From 4f5f0fd57c4b19320d5f7d3793136cbdb434e845 Mon Sep 17 00:00:00 2001
From: AhmadHammad21 <ahmedhammad8765@gmail.com>
Date: Wed, 20 May 2026 18:02:50 +0300
Subject: [PATCH 1/3] refactor: extract shared investigation runner and improve
 monitoring reliability

- Extract _run_investigation into src/agent/investigation_runner.py, eliminating
  ~130 lines of copy-paste between poller.py and event_consumer.py
- Synthesise a failed result when agent completes without calling submit_investigation
  (handles models with weak tool-use support like Gemma) so the event is always
  persisted and notified rather than silently dropped
- Add SSE push at GET /api/monitoring/stream; MonitoringPage switches from 30s
  polling to real-time alert delivery via EventSource with 30s fallback on error
- Fix services grid to survive restarts by falling back to DB-derived status when
  in-memory dict is empty; fix status mapping so completed investigations show
  as healthy (green) instead of amber
- Isolate poller boto3 calls to a dedicated ThreadPoolExecutor(max_workers=4)
  so poller ticks cannot saturate the shared default pool used by API requests
---
 frontend/src/pages/MonitoringPage.tsx |  44 ++++++-
 src/agent/event_consumer.py           | 134 +-------------------
 src/agent/investigation_runner.py     | 170 ++++++++++++++++++++++++++
 src/agent/monitor_store.py            |  64 +++++++++-
 src/agent/poller.py                   | 144 ++--------------------
 src/api/routers/monitoring.py         |  43 ++++++-
 6 files changed, 327 insertions(+), 272 deletions(-)
 create mode 100644 src/agent/investigation_runner.py

diff --git a/frontend/src/pages/MonitoringPage.tsx b/frontend/src/pages/MonitoringPage.tsx
index 6f66c7a..b9c85e6 100644
--- a/frontend/src/pages/MonitoringPage.tsx
+++ b/frontend/src/pages/MonitoringPage.tsx
@@ -60,7 +60,7 @@ export default function MonitoringPage() {
         toast.error(d.detail || 'Failed to send test event');
         return;
       }
-      toast.success('Test Lambda alarm sent — investigation will appear here in ~30s');
+      toast.success('Test Lambda alarm sent — investigation will appear here automatically');
     } catch {
       toast.error('Failed to send test event');
     } finally {
@@ -76,7 +76,47 @@ export default function MonitoringPage() {
       .finally(() => setLoading(false));
   };
 
-  useEffect(() => { load(); const id = setInterval(load, 30000); return () => clearInterval(id); }, []);
+  useEffect(() => {
+    load();
+
+    // Subscribe to real-time alert push via SSE
+    let es: EventSource | null = null;
+    let fallbackId: ReturnType<typeof setInterval> | null = null;
+
+    const connectSSE = () => {
+      es = new EventSource('/api/monitoring/stream');
+      es.onmessage = (e) => {
+        try {
+          const data = JSON.parse(e.data);
+          if (data.type === 'alert') {
+            setAlerts(prev =>
+              prev.some(a => a.id === data.alert.id) ? prev : [data.alert, ...prev]
+            );
+          }
+        } catch { /* ignore parse errors */ }
+      };
+      es.onerror = () => {
+        es?.close();
+        es = null;
+        // Fall back to polling if SSE fails
+        if (!fallbackId) fallbackId = setInterval(load, 30000);
+      };
+    };
+
+    connectSSE();
+
+    // Always poll services (not SSE-pushed yet)
+    const serviceId = setInterval(
+      () => fetchServices().then(setServices).catch(() => {}),
+      30000,
+    );
+
+    return () => {
+      es?.close();
+      if (fallbackId) clearInterval(fallbackId);
+      clearInterval(serviceId);
+    };
+  }, []);
 
   const healthy = services.filter(s => s.status === 'healthy').length;
   const errored = services.filter(s => s.status === 'error').length;
diff --git a/src/agent/event_consumer.py b/src/agent/event_consumer.py
index 5a1654c..061a875 100644
--- a/src/agent/event_consumer.py
+++ b/src/agent/event_consumer.py
@@ -111,137 +111,9 @@ def _build_investigation_prompt(event: dict) -> str:
 
 
 async def _run_investigation(prompt: str, session_id: str) -> dict[str, Any] | None:
-    """Run a full agent investigation using the same streaming path as the chat endpoint."""
-    from agent.core import get_agent
-    from agent.turns import save_turn
-
-    def _f(obj: Any, key: str, default: Any = None) -> Any:
-        return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)
-
-    config = {
-        "configurable": {"thread_id": session_id},
-        "recursion_limit": settings.max_tool_calls * 3 + 15,
-    }
-
-    tc_accum: dict[int, dict[str, Any]] = {}
-    pending_calls: dict[str, dict[str, Any]] = {}
-    tool_calls_log: list[dict[str, Any]] = []
-    investigation_result: dict[str, Any] | None = None
-    response_text = ""
-    usage_meta: Any = None
-
-    try:
-        async with asyncio.timeout(settings.investigation_timeout):
-            async for chunk, _meta in get_agent().astream(
-                {"messages": [{"role": "user", "content": prompt}]},
-                config=config,
-                stream_mode="messages",
-            ):
-                um = getattr(chunk, "usage_metadata", None)
-                if um:
-                    usage_meta = um
-
-                for tcc in getattr(chunk, "tool_call_chunks", []) or []:
-                    idx = _f(tcc, "index", 0)
-                    if idx not in tc_accum:
-                        tc_accum[idx] = {"id": "", "name": "", "args_str": ""}
-                    if tc_id := _f(tcc, "id"):
-                        tc_accum[idx]["id"] = tc_id
-                    if name := (_f(tcc, "name") or ""):
-                        tc_accum[idx]["name"] += name
-                    if args := (_f(tcc, "args") or ""):
-                        tc_accum[idx]["args_str"] += args
-
-                for tc in getattr(chunk, "tool_calls", []) or []:
-                    tc_id = _f(tc, "id") or ""
-                    name = _f(tc, "name") or ""
-                    args = _f(tc, "args") or {}
-                    if tc_id and name:
-                        pending_calls[tc_id] = {
-                            "tool": name,
-                            "args": args if isinstance(args, dict) else {},
-                        }
-
-                content = getattr(chunk, "content", "")
-                tc_id = getattr(chunk, "tool_call_id", None)
-
-                if content and isinstance(content, str) and not tc_id:
-                    response_text += content
-
-                if tc_id:
-                    for entry in tc_accum.values():
-                        if eid := entry["id"]:
-                            try:
-                                eargs: Any = (
-                                    json.loads(entry["args_str"]) if entry["args_str"] else {}
-                                )
-                            except json.JSONDecodeError:
-                                eargs = {}
-                            pending_calls[eid] = {"tool": entry["name"], "args": eargs}
-                    tc_accum.clear()
-
-                    call_info = pending_calls.pop(
-                        tc_id,
-                        {"tool": getattr(chunk, "name", None) or "unknown", "args": {}},
-                    )
-                    try:
-                        result = json.loads(content)
-                    except (json.JSONDecodeError, TypeError):
-                        result = {"raw": str(content)[:500]}
-
-                    tool_calls_log.append({
-                        "tool": call_info["tool"],
-                        "args": call_info["args"],
-                        "result": result,
-                    })
-                    if call_info["tool"] == "submit_investigation":
-                        investigation_result = call_info["args"]
-
-    except Exception as e:
-        logger.error("Agent investigation failed: {}", e)
-        investigation_result = {
-            "_status": "failed",
-            "root_cause_summary": f"Investigation failed: {e}",
-            "confidence": "LOW",
-            "mitigation_steps": [
-                "Re-investigate manually via the chat — use a higher MAX_TOOL_CALLS if needed.",
-            ],
-        }
-
-    # Build readable assistant summary
-    if (
-        investigation_result
-        and investigation_result.get("root_cause_summary")
-        and "_status" not in investigation_result
-    ):
-        steps = investigation_result.get("mitigation_steps", [])
-        steps_text = "\n".join(f"- {s}" for s in steps) if steps else "_No steps provided._"
-        confidence = investigation_result.get("confidence", "MEDIUM")
-        assistant_text = (
-            f"**Root Cause ({confidence} confidence):** "
-            f"{investigation_result['root_cause_summary']}\n\n"
-            f"**Mitigation Steps:**\n{steps_text}"
-        )
-    elif response_text.strip():
-        assistant_text = response_text.strip()
-    else:
-        assistant_text = "Investigation completed. See tool calls for details."
-
-    await save_turn(
-        session_id=session_id,
-        user_message=prompt,
-        assistant_text=assistant_text,
-        tool_calls_log=tool_calls_log,
-        usage={
-            "model": settings.llm_model,
-            "input_tokens": _f(usage_meta, "input_tokens", 0) or 0,
-            "output_tokens": _f(usage_meta, "output_tokens", 0) or 0,
-            "latency_ms": 0,
-        },
-        source="event",
-    )
-
-    return investigation_result
+    from agent.investigation_runner import run_investigation
+    result, _tool_calls_log = await run_investigation(prompt, session_id)
+    return result
 
 
 def _extract_aws_error(event: dict) -> str:
diff --git a/src/agent/investigation_runner.py b/src/agent/investigation_runner.py
new file mode 100644
index 0000000..32c8e7e
--- /dev/null
+++ b/src/agent/investigation_runner.py
@@ -0,0 +1,170 @@
+"""Shared investigation runner — used by both the poller and event consumer."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from loguru import logger
+
+from config import settings
+
+
+async def run_investigation(
+    prompt: str, session_id: str
+) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]:
+    """Run a full agent investigation. Returns (result, tool_calls_log).
+
+    Both the poller and the event consumer use this. result is None only when the
+    LLM never called submit_investigation (e.g. MAX_TOOL_CALLS hit without finishing).
+    On agent exception, result is returned with _status="failed" so callers can still
+    persist a record and send notifications.
+    """
+    from agent.core import get_agent
+    from agent.turns import save_turn
+
+    def _f(obj: Any, key: str, default: Any = None) -> Any:
+        return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)
+
+    config = {
+        "configurable": {"thread_id": session_id},
+        "recursion_limit": settings.max_tool_calls * 3 + 15,
+    }
+
+    tc_accum: dict[int, dict[str, Any]] = {}
+    pending_calls: dict[str, dict[str, Any]] = {}
+    tool_calls_log: list[dict[str, Any]] = []
+    investigation_result: dict[str, Any] | None = None
+    response_text = ""
+    usage_meta: Any = None
+
+    try:
+        async with asyncio.timeout(settings.investigation_timeout):
+            async for chunk, _meta in get_agent().astream(
+                {"messages": [{"role": "user", "content": prompt}]},
+                config=config,
+                stream_mode="messages",
+            ):
+                um = getattr(chunk, "usage_metadata", None)
+                if um:
+                    usage_meta = um
+
+                for tcc in getattr(chunk, "tool_call_chunks", []) or []:
+                    idx = _f(tcc, "index", 0)
+                    if idx not in tc_accum:
+                        tc_accum[idx] = {"id": "", "name": "", "args_str": ""}
+                    if tc_id := _f(tcc, "id"):
+                        tc_accum[idx]["id"] = tc_id
+                    if name := (_f(tcc, "name") or ""):
+                        tc_accum[idx]["name"] += name
+                    if args := (_f(tcc, "args") or ""):
+                        tc_accum[idx]["args_str"] += args
+
+                for tc in getattr(chunk, "tool_calls", []) or []:
+                    tc_id = _f(tc, "id") or ""
+                    name = _f(tc, "name") or ""
+                    args = _f(tc, "args") or {}
+                    if tc_id and name:
+                        pending_calls[tc_id] = {
+                            "tool": name,
+                            "args": args if isinstance(args, dict) else {},
+                        }
+
+                content = getattr(chunk, "content", "")
+                tc_id = getattr(chunk, "tool_call_id", None)
+
+                if content and isinstance(content, str) and not tc_id:
+                    response_text += content
+
+                if tc_id:
+                    for entry in tc_accum.values():
+                        if eid := entry["id"]:
+                            try:
+                                eargs: Any = (
+                                    json.loads(entry["args_str"]) if entry["args_str"] else {}
+                                )
+                            except json.JSONDecodeError:
+                                eargs = {}
+                            pending_calls[eid] = {"tool": entry["name"], "args": eargs}
+                    tc_accum.clear()
+
+                    call_info = pending_calls.pop(
+                        tc_id,
+                        {"tool": getattr(chunk, "name", None) or "unknown", "args": {}},
+                    )
+                    try:
+                        result = json.loads(content)
+                    except (json.JSONDecodeError, TypeError):
+                        result = {"raw": str(content)[:500]}
+
+                    tool_calls_log.append({
+                        "tool": call_info["tool"],
+                        "args": call_info["args"],
+                        "result": result,
+                    })
+                    if call_info["tool"] == "submit_investigation":
+                        investigation_result = call_info["args"]
+
+    except Exception as e:
+        logger.error("Investigation failed: {}", e)
+        investigation_result = {
+            "_status": "failed",
+            "root_cause_summary": f"Investigation failed: {e}",
+            "confidence": "LOW",
+            "mitigation_steps": [
+                "Re-investigate manually via the chat — use a higher MAX_TOOL_CALLS if needed.",
+            ],
+        }
+
+    # Agent completed without calling submit_investigation (e.g. hit recursion limit or
+    # the model doesn't support tool use reliably). Synthesize a failed result so callers
+    # always get a non-None value and can still persist + notify.
+    if investigation_result is None:
+        summary = response_text.strip() or "Agent completed without producing a structured result."
+        logger.warning(
+            "Investigation ended without submit_investigation — synthesising failed result"
+        )
+        investigation_result = {
+            "_status": "failed",
+            "root_cause_summary": summary[:500],
+            "confidence": "LOW",
+            "mitigation_steps": [
+                "Re-investigate manually via the chat.",
+                "Consider switching to a model with stronger tool-use support.",
+            ],
+        }
+
+    if (
+        investigation_result
+        and investigation_result.get("root_cause_summary")
+        and "_status" not in investigation_result
+    ):
+        steps = investigation_result.get("mitigation_steps", [])
+        steps_text = "\n".join(f"- {s}" for s in steps) if steps else "_No steps provided._"
+        confidence = investigation_result.get("confidence", "MEDIUM")
+        assistant_text = (
+            f"**Root Cause ({confidence} confidence):** "
+            f"{investigation_result['root_cause_summary']}\n\n"
+            f"**Mitigation Steps:**\n{steps_text}"
+        )
+    elif response_text.strip():
+        assistant_text = response_text.strip()
+    else:
+        assistant_text = "Investigation completed. See tool calls for details."
+
+    await save_turn(
+        session_id=session_id,
+        user_message=prompt,
+        assistant_text=assistant_text,
+        tool_calls_log=tool_calls_log,
+        usage={
+            "model": settings.llm_model,
+            "input_tokens": _f(usage_meta, "input_tokens", 0) or 0,
+            "output_tokens": _f(usage_meta, "output_tokens", 0) or 0,
+            "latency_ms": 0,
+        },
+        source="event",
+    )
+
+    return investigation_result, tool_calls_log
diff --git a/src/agent/monitor_store.py b/src/agent/monitor_store.py
index 4006217..ef652f6 100644
--- a/src/agent/monitor_store.py
+++ b/src/agent/monitor_store.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from datetime import UTC, datetime
 
 from loguru import logger
@@ -9,6 +10,27 @@
 # In-memory service status (ephemeral — just current state)
 _services: dict[str, dict] = {}
 
+# SSE subscribers — one Queue per connected /api/monitoring/stream client
+_alert_subscribers: set[asyncio.Queue[dict | None]] = set()
+
+
+def subscribe_alerts() -> asyncio.Queue[dict | None]:
+    q: asyncio.Queue[dict | None] = asyncio.Queue(maxsize=50)
+    _alert_subscribers.add(q)
+    return q
+
+
+def unsubscribe_alerts(q: asyncio.Queue[dict | None]) -> None:
+    _alert_subscribers.discard(q)
+
+
+def _broadcast_alert(alert: dict) -> None:
+    for q in list(_alert_subscribers):
+        try:
+            q.put_nowait(alert)
+        except asyncio.QueueFull:
+            pass
+
 
 async def add_alert(
     service: str,
@@ -22,14 +44,30 @@ async def add_alert(
     trigger_source: str | None = None,
     evidence: list | None = None,
 ) -> str:
-    """Persist an alert to the DB backend."""
+    """Persist an alert to the DB backend and broadcast to SSE subscribers."""
     from agent.db import db
 
     try:
-        return await db.add_alert(
+        alert_id = await db.add_alert(
             service, error, resolution, confidence, sns_sent,
             dedup_key, status, session_id, trigger_source, evidence,
         )
+        if alert_id and _alert_subscribers:
+            _broadcast_alert({
+                "id": alert_id,
+                "service": service,
+                "error": error,
+                "resolution": resolution,
+                "confidence": confidence,
+                "sns_sent": sns_sent,
+                "timestamp": datetime.now(UTC).isoformat(),
+                "dedup_key": dedup_key,
+                "status": status,
+                "trigger_source": trigger_source,
+                "session_id": session_id,
+                "evidence": evidence or [],
+            })
+        return alert_id
     except Exception as e:
         logger.error("Failed to persist alert: {}", e)
         return ""
@@ -109,7 +147,7 @@ async def is_incident_claimed(incident_key: str, within_minutes: int = 3) -> boo
 def update_service(name: str, status: str, error: str | None = None) -> None:
     _services[name] = {
         "name": name,
-        "status": status,
+        "status": "error" if status == "failed" else "healthy",
         "last_check": datetime.now(UTC).isoformat(),
         "last_error": error,
     }
@@ -127,5 +165,21 @@ async def get_alert_async(alert_id: str) -> dict | None:
     return await db.get_alert(alert_id)
 
 
-def get_services() -> list[dict]:
-    return list(_services.values())
+async def get_services() -> list[dict]:
+    """Return service health. Uses in-memory dict if populated; falls back to DB on restart."""
+    if _services:
+        return list(_services.values())
+    # After restart the dict is empty — derive last-known status per service from persisted alerts.
+    alerts = await get_alerts_async(limit=200)
+    seen: dict[str, dict] = {}
+    for a in alerts:
+        svc = a.get("service") or ""
+        if svc and svc not in seen:
+            alert_status = a.get("status", "completed")
+            seen[svc] = {
+                "name": svc,
+                "status": "error" if alert_status == "failed" else "healthy",
+                "last_check": a.get("timestamp") or "",
+                "last_error": a.get("error") if alert_status == "failed" else None,
+            }
+    return list(seen.values())
diff --git a/src/agent/poller.py b/src/agent/poller.py
index 4997279..3805265 100644
--- a/src/agent/poller.py
+++ b/src/agent/poller.py
@@ -10,13 +10,20 @@
 from __future__ import annotations
 
 import asyncio
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 from loguru import logger
 
-from agent.incident_keys import alarm_incident_key, lambda_error_incident_key, lambda_metric_incident_key
+from agent.incident_keys import (
+    alarm_incident_key,
+    lambda_error_incident_key,
+)
 from config import settings
 
+# Dedicated pool so poller boto3 calls never compete with API request threads
+_POLLER_POOL = ThreadPoolExecutor(max_workers=4, thread_name_prefix="poller")
+
 
 def _claim_window_minutes() -> int:
     return max(1, settings.poll_reinvestigate_hours * 60)
@@ -25,135 +32,8 @@ def _claim_window_minutes() -> int:
 async def _run_investigation(
     prompt: str, session_id: str
 ) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]:
-    """Run a full agent investigation using the same streaming path as the chat endpoint."""
-    import json
-
-    from agent.core import get_agent
-    from agent.turns import save_turn
-
-    def _f(obj: Any, key: str, default: Any = None) -> Any:
-        return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)
-
-    config = {
-        "configurable": {"thread_id": session_id},
-        "recursion_limit": settings.max_tool_calls * 3 + 15,
-    }
-
-    tc_accum: dict[int, dict[str, Any]] = {}
-    pending_calls: dict[str, dict[str, Any]] = {}
-    tool_calls_log: list[dict[str, Any]] = []
-    investigation_result: dict[str, Any] | None = None
-    response_text = ""
-    usage_meta: Any = None
-
-    try:
-        async with asyncio.timeout(settings.investigation_timeout):
-            async for chunk, _meta in get_agent().astream(
-                {"messages": [{"role": "user", "content": prompt}]},
-                config=config,
-                stream_mode="messages",
-            ):
-                um = getattr(chunk, "usage_metadata", None)
-                if um:
-                    usage_meta = um
-
-                for tcc in getattr(chunk, "tool_call_chunks", []) or []:
-                    idx = _f(tcc, "index", 0)
-                    if idx not in tc_accum:
-                        tc_accum[idx] = {"id": "", "name": "", "args_str": ""}
-                    if tc_id := _f(tcc, "id"):
-                        tc_accum[idx]["id"] = tc_id
-                    if name := (_f(tcc, "name") or ""):
-                        tc_accum[idx]["name"] += name
-                    if args := (_f(tcc, "args") or ""):
-                        tc_accum[idx]["args_str"] += args
-
-                for tc in getattr(chunk, "tool_calls", []) or []:
-                    tc_id = _f(tc, "id") or ""
-                    name = _f(tc, "name") or ""
-                    args = _f(tc, "args") or {}
-                    if tc_id and name:
-                        pending_calls[tc_id] = {
-                            "tool": name,
-                            "args": args if isinstance(args, dict) else {},
-                        }
-
-                content = getattr(chunk, "content", "")
-                tc_id = getattr(chunk, "tool_call_id", None)
-
-                if content and isinstance(content, str) and not tc_id:
-                    response_text += content
-
-                if tc_id:
-                    for entry in tc_accum.values():
-                        if eid := entry["id"]:
-                            try:
-                                eargs: Any = (
-                                    json.loads(entry["args_str"]) if entry["args_str"] else {}
-                                )
-                            except json.JSONDecodeError:
-                                eargs = {}
-                            pending_calls[eid] = {"tool": entry["name"], "args": eargs}
-                    tc_accum.clear()
-
-                    call_info = pending_calls.pop(
-                        tc_id,
-                        {"tool": getattr(chunk, "name", None) or "unknown", "args": {}},
-                    )
-                    try:
-                        result = json.loads(content)
-                    except (json.JSONDecodeError, TypeError):
-                        result = {"raw": str(content)[:500]}
-
-                    tool_calls_log.append({
-                        "tool": call_info["tool"],
-                        "args": call_info["args"],
-                        "result": result,
-                    })
-                    if call_info["tool"] == "submit_investigation":
-                        investigation_result = call_info["args"]
-
-    except Exception as e:
-        logger.error("Poller investigation failed: {}", e)
-        investigation_result = {
-            "_status": "failed",
-            "root_cause_summary": f"Investigation failed: {e}",
-            "confidence": "LOW",
-            "mitigation_steps": [
-                "Re-investigate manually via the chat — use a higher MAX_TOOL_CALLS if needed.",
-            ],
-        }
-
-    if (investigation_result and investigation_result.get("root_cause_summary")
-            and "_status" not in investigation_result):
-        steps = investigation_result.get("mitigation_steps", [])
-        steps_text = "\n".join(f"- {s}" for s in steps) if steps else "_No steps provided._"
-        confidence = investigation_result.get("confidence", "MEDIUM")
-        assistant_text = (
-            f"**Root Cause ({confidence} confidence):** "
-            f"{investigation_result['root_cause_summary']}\n\n"
-            f"**Mitigation Steps:**\n{steps_text}"
-        )
-    elif response_text.strip():
-        assistant_text = response_text.strip()
-    else:
-        assistant_text = "Investigation completed. See tool calls for details."
-
-    await save_turn(
-        session_id=session_id,
-        user_message=prompt,
-        assistant_text=assistant_text,
-        tool_calls_log=tool_calls_log,
-        usage={
-            "model": settings.llm_model,
-            "input_tokens": _f(usage_meta, "input_tokens", 0) or 0,
-            "output_tokens": _f(usage_meta, "output_tokens", 0) or 0,
-            "latency_ms": 0,
-        },
-        source="event",
-    )
-
-    return investigation_result, tool_calls_log
+    from agent.investigation_runner import run_investigation
+    return await run_investigation(prompt, session_id)
 
 
 async def _persist_and_notify(
@@ -231,7 +111,7 @@ async def _check_alarms() -> None:
     )
     from tools.cloudwatch import get_alarms
 
-    data = await asyncio.get_event_loop().run_in_executor(None, get_alarms, "ALARM")
+    data = await asyncio.get_event_loop().run_in_executor(_POLLER_POOL, get_alarms, "ALARM")
     alarms = data.get("alarms", [])
     logger.debug("Poller: {} alarm(s) in ALARM state", len(alarms))
     for alarm in alarms:
@@ -305,7 +185,7 @@ async def _check_lambda_errors() -> None:
     )
     from tools.lambda_ import get_lambda_error_rate, list_lambda_functions
 
-    funcs = await asyncio.get_event_loop().run_in_executor(None, list_lambda_functions)
+    funcs = await asyncio.get_event_loop().run_in_executor(_POLLER_POOL, list_lambda_functions)
     fn_list = funcs.get("functions", [])[:20]
     logger.debug("Poller: checking error rates for {} Lambda function(s)", len(fn_list))
     for fn in fn_list:  # cap at 20 to avoid runaway API calls
diff --git a/src/api/routers/monitoring.py b/src/api/routers/monitoring.py
index 0aac9b7..ef59c9f 100644
--- a/src/api/routers/monitoring.py
+++ b/src/api/routers/monitoring.py
@@ -2,11 +2,20 @@
 
 from __future__ import annotations
 
+import asyncio
+import json
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import StreamingResponse
 
-from agent.monitor_store import get_alerts_async, get_alert_async, get_services
+from agent.monitor_store import (
+    get_alert_async,
+    get_alerts_async,
+    get_services,  # async — returns in-memory or falls back to DB
+    subscribe_alerts,
+    unsubscribe_alerts,
+)
 from api.auth import get_current_user
 
 router = APIRouter(prefix="/api/monitoring", tags=["monitoring"])
@@ -16,7 +25,7 @@
 async def services(
     _user: Annotated[dict | None, Depends(get_current_user)],
 ) -> list:
-    return get_services()
+    return await get_services()
 
 
 @router.get("/alerts")
@@ -36,3 +45,33 @@ async def alert_detail(
     if not alert:
         raise HTTPException(status_code=404, detail="Alert not found")
     return alert
+
+
+@router.get("/stream")
+async def monitoring_stream(
+    _user: Annotated[dict | None, Depends(get_current_user)],
+) -> StreamingResponse:
+    """SSE stream — pushes new alerts in real time as they are persisted."""
+
+    async def generate():
+        q = subscribe_alerts()
+        try:
+            yield f"data: {json.dumps({'type': 'connected'})}\n\n"
+            while True:
+                try:
+                    alert = await asyncio.wait_for(q.get(), timeout=20.0)
+                    if alert is None:
+                        break
+                    yield f"data: {json.dumps({'type': 'alert', 'alert': alert})}\n\n"
+                except TimeoutError:
+                    yield f"data: {json.dumps({'type': 'heartbeat'})}\n\n"
+        except asyncio.CancelledError:
+            pass
+        finally:
+            unsubscribe_alerts(q)
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )

From aaef6934c1adc4f073d92d7c05f4887391530f21 Mon Sep 17 00:00:00 2001
From: AhmadHammad21 <ahmedhammad8765@gmail.com>
Date: Wed, 20 May 2026 18:40:18 +0300
Subject: [PATCH 2/3] feat: remove SNS integration

SNS was fully optional (poller never used it, event consumer gated on
sns_topic_arn being set) and added IAM surface area with no benefit over
the existing Slack and Telegram channels.

- Delete src/tools/sns.py
- Remove SNS block from event_consumer._deliver(); sns_sent hardcoded False
- Remove sns_topic_arn from appsettings, init_store (_default + accessor),
  SetupBody, and /api/init/setup handler
- Remove SNS permission check from permission_checker.check_permissions()
- Remove SNS field from SettingsPage AWS Configuration tab and InitPage wizard
- Remove SNS_TOPIC_ARN from .env.example
- Clean up permission_checker line lengths while touching the file
---
 .env.example                        |   3 +-
 frontend/src/pages/InitPage.tsx     |  18 +--
 frontend/src/pages/SettingsPage.tsx |   7 +-
 scripts/test_dlq.py                 | 182 +++++++++++++++++++++++++
 src/agent/event_consumer.py         |  40 +-----
 src/agent/init_store.py             |   4 -
 src/agent/permission_checker.py     |  44 +++----
 src/api/routers/init.py             |   6 +-
 src/config/appsettings.py           |   5 +-
 src/tools/sns.py                    |  23 ----
 tests/integration/test_dlq_logic.py | 197 ++++++++++++++++++++++++++++
 11 files changed, 411 insertions(+), 118 deletions(-)
 create mode 100644 scripts/test_dlq.py
 delete mode 100644 src/tools/sns.py
 create mode 100644 tests/integration/test_dlq_logic.py

diff --git a/.env.example b/.env.example
index a626f4b..95e5a1b 100644
--- a/.env.example
+++ b/.env.example
@@ -70,7 +70,6 @@ CHECKPOINT_BACKEND=memory
 # POLL_ERROR_THRESHOLD=5.0        # Lambda error rate % that triggers an investigation
 # POLL_REINVESTIGATE_HOURS=1      # don't re-investigate the same alarm within N hours
 
-# Event-driven detection — configure SNS/SQS via Settings → AWS Configuration after login
-# SNS_TOPIC_ARN=arn:aws:sns:us-east-1:123456789012:opendevops-alerts
+# Event-driven detection — configure SQS via Settings → AWS Configuration after login
 # SQS_QUEUE_URL=https://sqs.us-east-1.amazonaws.com/123456789012/opendevops
 # EVENT_CONSUMER_ENABLED=false
diff --git a/frontend/src/pages/InitPage.tsx b/frontend/src/pages/InitPage.tsx
index cefa2bc..e07c18c 100644
--- a/frontend/src/pages/InitPage.tsx
+++ b/frontend/src/pages/InitPage.tsx
@@ -57,7 +57,6 @@ export default function InitPage() {
 
   // Step 2
   const [region,   setRegion]   = useState('us-east-1');
-  const [snsArn,   setSnsArn]   = useState('');
 
   // Step 3
   const [perms,    setPerms]    = useState<Record<string, PermResult>>({});
@@ -78,7 +77,6 @@ export default function InitPage() {
       .then(r => r.json())
       .then(d => {
         setRegion(d.aws_region || 'us-east-1');
-        setSnsArn(d.sns_topic_arn || '');
         if (d.initialized && (d.has_user || !d.auth_enabled)) {
           navigate('/', { replace: true });
           return;
@@ -114,7 +112,7 @@ export default function InitPage() {
       const r = await apiFetch('/api/init/setup', {
         method: 'POST',
         headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ aws_region: region, sns_topic_arn: snsArn, sqs_queue_url: '', org_name: orgName.trim() }),
+        body: JSON.stringify({ aws_region: region, sqs_queue_url: '', org_name: orgName.trim() }),
       });
       if (!r.ok) { setError('Failed to save configuration'); return; }
       setStep(3);
@@ -279,20 +277,6 @@ export default function InitPage() {
                     ))}
                   </select>
                 </div>
-                <div>
-                  <label className="block text-sm font-medium text-gray-700 dark:text-[#D4D4D8] mb-2">
-                    SNS Topic ARN <span className="text-gray-400 dark:text-[#52525B] font-normal">(optional)</span>
-                  </label>
-                  <input
-                    value={snsArn}
-                    onChange={e => setSnsArn(e.target.value)}
-                    placeholder="arn:aws:sns:us-east-1:123456789012:alerts"
-                    className={inp}
-                  />
-                  <p className="mt-1.5 text-xs text-gray-400 dark:text-[#52525B]">
-                    Leave empty to skip SNS notifications — alerts will still appear in the monitoring page.
-                  </p>
-                </div>
               </div>
               {error && <p className="mt-4 text-sm text-red-500">{error}</p>}
               <button onClick={saveConfig} disabled={loading}
diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx
index 4086993..19256a6 100644
--- a/frontend/src/pages/SettingsPage.tsx
+++ b/frontend/src/pages/SettingsPage.tsx
@@ -49,7 +49,6 @@ export default function SettingsPage() {
   const [telegramTesting,    setTelegramTesting]    = useState(false);
 
   // AWS tab state
-  const [snsArn,        setSnsArn]        = useState('');
   const [sqsUrl,        setSqsUrl]        = useState('');
   const [awsRegion,     setAwsRegion]     = useState('');
   const [awsSaving,     setAwsSaving]     = useState(false);
@@ -91,7 +90,6 @@ export default function SettingsPage() {
     apiFetch('/api/init/status')
       .then(r => r.json())
       .then(d => {
-        setSnsArn(d.sns_topic_arn || '');
         setSqsUrl(d.sqs_queue_url || '');
         setAwsRegion(d.aws_region || '');
         setInfraEnabled(!!d.event_infra_enabled);
@@ -108,7 +106,7 @@ export default function SettingsPage() {
       const r = await apiFetch('/api/init/setup', {
         method: 'POST',
         headers: { 'Content-Type': 'application/json' },
-        body: JSON.stringify({ sns_topic_arn: snsArn, sqs_queue_url: sqsUrl, aws_region: awsRegion }),
+        body: JSON.stringify({ sqs_queue_url: sqsUrl, aws_region: awsRegion }),
       });
       if (!r.ok) throw new Error('save failed');
       toast.success('AWS configuration saved');
@@ -412,9 +410,8 @@ export default function SettingsPage() {
 
               <div className="divide-y divide-gray-100 dark:divide-[#27272F]">
                 {[
-                  { label: 'SNS Topic ARN', hint: 'Alerts are published here after each investigation', value: snsArn, set: setSnsArn, placeholder: 'arn:aws:sns:us-east-1:123456789012:alerts' },
                   { label: 'SQS Queue URL', hint: 'Event consumer polls this queue for CloudWatch alarms', value: sqsUrl, set: setSqsUrl, placeholder: 'https://sqs.us-east-1.amazonaws.com/123456789012/opendevops' },
-                  { label: 'AWS Region',    hint: 'Region for SNS, SQS, and EventBridge resources', value: awsRegion, set: setAwsRegion, placeholder: 'us-east-1' },
+                  { label: 'AWS Region',    hint: 'Region for SQS and EventBridge resources', value: awsRegion, set: setAwsRegion, placeholder: 'us-east-1' },
                 ].map(({ label, hint, value, set, placeholder }) => (
                   <div key={label} className="flex items-start gap-4 px-4 py-[11px]">
                     <div className="flex-[0_0_200px] pt-0.5">
diff --git a/scripts/test_dlq.py b/scripts/test_dlq.py
new file mode 100644
index 0000000..ef54d6a
--- /dev/null
+++ b/scripts/test_dlq.py
@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+test_dlq.py — verify the SQS → DLQ redrive policy works end-to-end.
+
+Sends a poison message (invalid JSON) to the main queue, then simulates
+maxReceiveCount failed processing attempts by receiving the message and
+immediately releasing it back each time without deleting it. After the
+limit is hit, AWS automatically moves the message to the DLQ.
+
+IMPORTANT: Stop the app server before running this script. If the event
+consumer is running it will compete for the poison message, hold it for
+the full visibility timeout (180s), and the script will not be able to
+reach it. The consumer's behaviour (not deleting bad JSON) is already
+covered by the pytest suite — this script only verifies AWS redrive.
+
+Requirements: AWS profile with sqs:SendMessage, sqs:ReceiveMessage,
+              sqs:ChangeMessageVisibility, sqs:DeleteMessage on both queues.
+
+Usage:
+  uv run python scripts/test_dlq.py --profile my-write-profile
+  uv run python scripts/test_dlq.py --profile my-write-profile --region eu-west-1
+  uv run python scripts/test_dlq.py --profile my-write-profile --no-cleanup
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+
+import boto3
+from botocore.exceptions import ClientError
+
+QUEUE_NAME = "opendevops-agent-events"
+DLQ_QUEUE_NAME = "opendevops-agent-events-dlq"
+MAX_RECEIVE_COUNT = 5
+
+
+def _get_queue_url(sqs, name: str) -> str:
+    return sqs.get_queue_url(QueueName=name)["QueueUrl"]
+
+
+def _get_queue_attrs(sqs, queue_url: str) -> dict:
+    return sqs.get_queue_attributes(
+        QueueUrl=queue_url,
+        AttributeNames=["RedrivePolicy", "ApproximateNumberOfMessagesNotVisible"],
+    )["Attributes"]
+
+
+def _get_max_receive_count(attrs: dict) -> int:
+    policy = json.loads(attrs.get("RedrivePolicy", "{}"))
+    return int(policy.get("maxReceiveCount", MAX_RECEIVE_COUNT))
+
+
+def main(profile: str, region: str, cleanup: bool) -> int:
+    session = boto3.Session(profile_name=profile) if profile else boto3.Session()
+    sqs = session.client("sqs", region_name=region)
+
+    print(f"Profile : {profile or 'default'}")
+    print(f"Region  : {region}\n")
+
+    try:
+        main_url = _get_queue_url(sqs, QUEUE_NAME)
+        dlq_url = _get_queue_url(sqs, DLQ_QUEUE_NAME)
+    except ClientError as e:
+        print(f"✗ Could not resolve queue URLs: {e}")
+        print("  Make sure the init wizard has been run (Settings → AWS Configuration).")
+        return 1
+
+    main_attrs = _get_queue_attrs(sqs, main_url)
+    max_receive = _get_max_receive_count(main_attrs)
+    in_flight = int(main_attrs.get("ApproximateNumberOfMessagesNotVisible", 0))
+
+    print(f"Main queue : {main_url}")
+    print(f"DLQ        : {dlq_url}")
+    print(f"maxReceiveCount : {max_receive}")
+
+    if in_flight > 0:
+        print(
+            f"\n⚠ WARNING: {in_flight} message(s) currently in-flight on the main queue."
+            "\n  If the event consumer is running it will compete for the poison message."
+            "\n  Stop the app server first, then re-run this script."
+        )
+        return 1
+
+    # Unique marker so we can identify our message among any others in the queue
+    marker = f"opendevops-dlq-test-{int(time.time())}"
+    poison = f"NOT_VALID_JSON [{marker}]"
+
+    try:
+        sqs.send_message(QueueUrl=main_url, MessageBody=poison)
+    except ClientError as e:
+        print(f"\n✗ Failed to send poison message: {e}")
+        print("  The profile needs sqs:SendMessage on the main queue.")
+        return 1
+
+    print(f"\nSent poison message : {poison}")
+    print(f"Simulating {max_receive} failed receive attempts...\n")
+
+    grabbed = False
+    for attempt in range(1, max_receive + 2):
+        resp = sqs.receive_message(
+            QueueUrl=main_url,
+            MaxNumberOfMessages=10,
+            VisibilityTimeout=2,
+            WaitTimeSeconds=5,
+            AttributeNames=["ApproximateReceiveCount"],
+        )
+        target = next(
+            (m for m in resp.get("Messages", []) if marker in m["Body"]), None
+        )
+
+        if target is None:
+            if grabbed:
+                print(f"  Attempt {attempt}: message gone from main queue — redrive triggered")
+            else:
+                # Message was never visible — likely grabbed by event consumer
+                print(
+                    f"  Attempt {attempt}: message not visible (in-flight)."
+                    "\n  The event consumer may have picked it up."
+                    "\n  Stop the app server and re-run this script."
+                )
+                return 1
+            break
+
+        grabbed = True
+        count = target.get("Attributes", {}).get("ApproximateReceiveCount", "?")
+        print(f"  Attempt {attempt}: received (ApproximateReceiveCount={count})")
+
+        # Simulate failed processing: release immediately without deleting
+        try:
+            sqs.change_message_visibility(
+                QueueUrl=main_url,
+                ReceiptHandle=target["ReceiptHandle"],
+                VisibilityTimeout=0,
+            )
+        except ClientError:
+            pass  # receipt handle may be invalid if visibility already expired
+
+        time.sleep(0.3)
+
+    print("\nWaiting 3s for AWS to complete the redrive...")
+    time.sleep(3)
+
+    print("Checking DLQ...")
+    dlq_resp = sqs.receive_message(
+        QueueUrl=dlq_url,
+        MaxNumberOfMessages=10,
+        VisibilityTimeout=0,
+        WaitTimeSeconds=5,
+    )
+    dlq_messages = dlq_resp.get("Messages", [])
+    hit = next((m for m in dlq_messages if marker in m["Body"]), None)
+
+    if hit:
+        print(f"\n✓ PASS — poison message found in DLQ after {max_receive} failed attempts.")
+        if cleanup:
+            sqs.delete_message(QueueUrl=dlq_url, ReceiptHandle=hit["ReceiptHandle"])
+            print("  Test message deleted from DLQ.")
+        else:
+            print("  Skipping cleanup (--no-cleanup). Message stays in DLQ.")
+        return 0
+
+    count = len(dlq_messages)
+    print(f"\n✗ FAIL — message not found in DLQ ({count} other message(s) present).")
+    print("  The message may still be in the main queue or the redrive is misconfigured.")
+    return 1
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Verify SQS → DLQ redrive end-to-end")
+    parser.add_argument("--profile", default="", help="AWS profile name (needs write access)")
+    parser.add_argument("--region", default="us-east-1")
+    parser.add_argument(
+        "--no-cleanup",
+        dest="cleanup",
+        action="store_false",
+        help="Leave the test message in the DLQ after the test",
+    )
+    args = parser.parse_args()
+    sys.exit(main(args.profile, args.region, args.cleanup))
diff --git a/src/agent/event_consumer.py b/src/agent/event_consumer.py
index 061a875..3f42431 100644
--- a/src/agent/event_consumer.py
+++ b/src/agent/event_consumer.py
@@ -13,7 +13,6 @@
 from agent.incident_keys import event_incident_key
 from agent.init_store import (
     get_runtime_aws_region,
-    get_runtime_sns_topic_arn,
     get_runtime_sqs_queue_url,
 )
 from agent.monitor_store import (
@@ -144,39 +143,6 @@ async def _deliver(result: dict[str, Any], event: dict, dedup_key: str, session_
     resolution = "\n".join(mitigation) if isinstance(mitigation, list) else str(mitigation)
     aws_error = _extract_aws_error(event)
 
-    sns_arn = get_runtime_sns_topic_arn()
-
-    sns_sent = False
-    sns_attempted = False
-    if sns_arn:
-        sns_attempted = True
-        try:
-            from tools.sns import publish_sns_alert
-
-            services_affected = result.get("services_affected", [])
-            evidence = result.get("evidence", [])
-            message = (
-                f"Service: {service}\n"
-                f"Root Cause: {root_cause}\n"
-                f"Evidence: {chr(10).join(evidence[:5]) if evidence else 'N/A'}\n"
-                f"Mitigation Steps:\n{resolution}\n"
-                "Services Affected: "
-                f"{', '.join(services_affected) if services_affected else service}\n"
-                f"Confidence: {confidence}\n"
-                f"Event Time: {event.get('time', '')}"
-            )
-            subject_prefix = "[FAILED]" if status == "failed" else f"[{confidence}]"
-            await asyncio.get_event_loop().run_in_executor(
-                None,
-                publish_sns_alert,
-                sns_arn,
-                f"{subject_prefix} {service} — {root_cause[:80]}",
-                message,
-            )
-            sns_sent = True
-        except Exception as e:
-            logger.error("SNS delivery failed: {}", e)
-
     slack_sent = False
     if settings.slack_webhook_url:
         try:
@@ -219,7 +185,7 @@ async def _deliver(result: dict[str, Any], event: dict, dedup_key: str, session_
         error=root_cause,
         resolution=resolution,
         confidence=confidence,
-        sns_sent=sns_sent,
+        sns_sent=False,
         dedup_key=dedup_key,
         status=status,
         session_id=session_id,
@@ -227,13 +193,11 @@ async def _deliver(result: dict[str, Any], event: dict, dedup_key: str, session_
         evidence=result.get("evidence", []),
     )
     if alert_id:
-        if sns_attempted:
-            await add_notification(alert_id, "sns", "delivered" if sns_sent else "failed")
         if settings.slack_webhook_url:
             await add_notification(alert_id, "slack", "delivered" if slack_sent else "failed")
         if settings.telegram_bot_token and settings.telegram_chat_id:
             await add_notification(alert_id, "telegram", "delivered" if telegram_sent else "failed")
-    logger.info("Alert delivered: service={} confidence={} sns={}", service, confidence, sns_sent)
+    logger.info("Alert delivered: service={} confidence={}", service, confidence)
 
 
 def _is_real_failure(source: str, detail: dict) -> bool:
diff --git a/src/agent/init_store.py b/src/agent/init_store.py
index 3c24496..7877b6e 100644
--- a/src/agent/init_store.py
+++ b/src/agent/init_store.py
@@ -35,7 +35,6 @@ def _default() -> dict:
         "setup_complete": False,
         "event_infra_enabled": False,
         "event_infra_managed": False,
-        "sns_topic_arn": "",
         "aws_region": settings.aws_region,
         "sqs_queue_url": "",
         "sqs_queue_arn": "",
@@ -50,7 +49,6 @@ def _default() -> dict:
             "ec2": None,
             "rds": None,
             "iam": None,
-            "sns": None,
             "sqs": None,
             "events": None,
         },
@@ -164,5 +162,3 @@ def get_runtime_sqs_queue_url() -> str:
     return str(settings.sqs_queue_url or load_init().get("sqs_queue_url") or "")
 
 
-def get_runtime_sns_topic_arn() -> str:
-    return str(settings.sns_topic_arn or load_init().get("sns_topic_arn") or "")
diff --git a/src/agent/permission_checker.py b/src/agent/permission_checker.py
index e5c85c7..61c31d8 100644
--- a/src/agent/permission_checker.py
+++ b/src/agent/permission_checker.py
@@ -9,7 +9,11 @@
 
 
 def _session() -> boto3.Session:
-    return boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session()
+    return (
+        boto3.Session(profile_name=settings.aws_profile)
+        if settings.aws_profile
+        else boto3.Session()
+    )
 
 
 def _check(fn) -> dict:
@@ -20,33 +24,29 @@ def _check(fn) -> dict:
         return {"passed": False, "error": str(e)}
 
 
-def check_permissions(sns_topic_arn: str = "", region: str | None = None) -> dict[str, dict]:
+def _c(s, service: str, region: str):
+    return s.client(service, region_name=region)
+
+
+def check_permissions(region: str | None = None) -> dict[str, dict]:
     """Run one lightweight read call per service. Returns {service: {passed, error}}."""
     s = _session()
     region = region or settings.aws_region
 
-    results: dict[str, dict] = {}
-
-    results["cloudwatch"] = _check(lambda: s.client("cloudwatch", region_name=region).describe_alarms(MaxRecords=1))
-    results["cloudtrail"] = _check(lambda: s.client("cloudtrail", region_name=region).lookup_events(MaxResults=1))
-    results["ecs"] = _check(lambda: s.client("ecs", region_name=region).list_clusters(maxResults=1))
-    results["lambda"] = _check(lambda: s.client("lambda", region_name=region).list_functions(MaxItems=1))
-    results["ec2"] = _check(lambda: s.client("ec2", region_name=region).describe_instances(MaxResults=5))
-    results["rds"] = _check(lambda: s.client("rds", region_name=region).describe_db_instances())
-    results["iam"] = _check(lambda: s.client("sts", region_name=region).get_caller_identity())
-
-    if sns_topic_arn:
-        results["sns"] = _check(
-            lambda: s.client("sns", region_name=region).get_topic_attributes(TopicArn=sns_topic_arn)
-        )
-    else:
-        results["sns"] = {"passed": None, "error": "No SNS topic ARN provided"}
-
-    results["sqs"] = _check(lambda: s.client("sqs", region_name=region).list_queues(MaxResults=1))
-    results["events"] = _check(lambda: s.client("events", region_name=region).list_rules(Limit=1))
+    results: dict[str, dict] = {
+        "cloudwatch": _check(lambda: _c(s, "cloudwatch", region).describe_alarms(MaxRecords=1)),
+        "cloudtrail": _check(lambda: _c(s, "cloudtrail", region).lookup_events(MaxResults=1)),
+        "ecs":        _check(lambda: _c(s, "ecs", region).list_clusters(maxResults=1)),
+        "lambda":     _check(lambda: _c(s, "lambda", region).list_functions(MaxItems=1)),
+        "ec2":        _check(lambda: _c(s, "ec2", region).describe_instances(MaxResults=5)),
+        "rds":        _check(lambda: _c(s, "rds", region).describe_db_instances()),
+        "iam":        _check(lambda: _c(s, "sts", region).get_caller_identity()),
+        "sqs":        _check(lambda: _c(s, "sqs", region).list_queues(MaxResults=1)),
+        "events":     _check(lambda: _c(s, "events", region).list_rules(Limit=1)),
+    }
 
     for svc, r in results.items():
-        status = "✓" if r["passed"] else ("?" if r["passed"] is None else "✗")
+        status = "✓" if r["passed"] else "✗"
         logger.debug("Permission check {}: {} {}", svc, status, r.get("error") or "")
 
     return results
diff --git a/src/api/routers/init.py b/src/api/routers/init.py
index a9f9024..3d732c4 100644
--- a/src/api/routers/init.py
+++ b/src/api/routers/init.py
@@ -1,4 +1,4 @@
-"""Init wizard API — first-run setup for SNS, SQS, and permission checks."""
+"""Init wizard API — first-run setup for SQS and permission checks."""
 
 from __future__ import annotations
 
@@ -31,7 +31,6 @@ class CreateUserBody(BaseModel):
 
 
 class SetupBody(BaseModel):
-    sns_topic_arn: str = ""
     aws_region: str = "us-east-1"
     sqs_queue_url: str = ""
     org_name: str = ""
@@ -96,7 +95,6 @@ async def setup(
     import re
 
     data = await load_init_async()
-    data["sns_topic_arn"] = body.sns_topic_arn.strip()
     data["aws_region"] = body.aws_region.strip() or settings.aws_region
     data["sqs_queue_url"] = body.sqs_queue_url.strip()
     data["event_infra_enabled"] = bool(data["sqs_queue_url"])
@@ -130,7 +128,7 @@ async def check_permissions_endpoint(
 
     data = await load_init_async()
     results = await asyncio.get_event_loop().run_in_executor(
-        None, _check, data.get("sns_topic_arn", ""), data.get("aws_region")
+        None, _check, data.get("aws_region")
     )
     data["permissions"] = {svc: r["passed"] for svc, r in results.items()}
     await save_init_async(data)
diff --git a/src/config/appsettings.py b/src/config/appsettings.py
index 19cf064..2364192 100644
--- a/src/config/appsettings.py
+++ b/src/config/appsettings.py
@@ -49,7 +49,7 @@ class Settings(BaseSettings):
     # Set summarization_enabled=false or summarization_threshold_chars=0 to disable.
     summarization_enabled: bool = True
     summarization_threshold_chars: int = 60_000   # ~15 K tokens; trigger compaction above this
-    summarization_keep_chars: int = 20_000        # ~5 K tokens of recent messages to preserve intact
+    summarization_keep_chars: int = 20_000        # ~5 K tokens of recent messages to keep intact
 
     # Slack — leave unset to disable notifications
     slack_webhook_url: str | None = None
@@ -67,8 +67,7 @@ class Settings(BaseSettings):
     jwt_secret: str | None = None
     jwt_expire_minutes: int = 1440  # 24 hours
 
-    # SNS / event consumer — leave unset to disable
-    sns_topic_arn: str | None = None
+    # Event consumer — leave unset to disable
     event_consumer_enabled: bool = False
     sqs_queue_url: str | None = None
 
diff --git a/src/tools/sns.py b/src/tools/sns.py
deleted file mode 100644
index 440ee6e..0000000
--- a/src/tools/sns.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""SNS tool: publish alerts."""
-
-import boto3
-from loguru import logger
-
-from agent.init_store import get_runtime_aws_region
-from config import settings
-
-
-def _sns_client():
-    session = (
-        boto3.Session(profile_name=settings.aws_profile)
-        if settings.aws_profile
-        else boto3.Session()
-    )
-    return session.client("sns", region_name=get_runtime_aws_region())
-
-
-def publish_sns_alert(topic_arn: str, subject: str, message: str) -> dict:
-    """Publish an alert to SNS."""
-    _sns_client().publish(TopicArn=topic_arn, Subject=subject[:100], Message=message)
-    logger.info("SNS alert published to {}", topic_arn)
-    return {"published": True, "topic_arn": topic_arn}
diff --git a/tests/integration/test_dlq_logic.py b/tests/integration/test_dlq_logic.py
new file mode 100644
index 0000000..c72bb37
--- /dev/null
+++ b/tests/integration/test_dlq_logic.py
@@ -0,0 +1,197 @@
+"""Unit tests for DLQ-related logic in the event consumer.
+
+Tests that the consumer correctly leaves messages in the queue when
+processing fails (should_delete stays False), which is what triggers
+the SQS redrive to the DLQ after maxReceiveCount attempts.
+
+Moto does not simulate the actual DLQ redrive — use scripts/test_dlq.py
+for end-to-end verification against real AWS.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_sqs_message(body: str, receipt: str = "receipt-abc") -> dict:
+    return {"Body": body, "ReceiptHandle": receipt}
+
+
+# ---------------------------------------------------------------------------
+# Tests: message NOT deleted on parse failure
+# ---------------------------------------------------------------------------
+
+def test_invalid_json_is_not_deleted():
+    """A message that cannot be parsed as JSON must stay in the queue."""
+    deleted = []
+    messages = [_make_sqs_message("NOT_VALID_JSON")]
+
+    for msg in messages:
+        should_delete = False
+        try:
+            json.loads(msg["Body"])
+            should_delete = True
+        except json.JSONDecodeError:
+            pass  # leave in queue → SQS will redrive after maxReceiveCount
+        if should_delete:
+            deleted.append(msg["ReceiptHandle"])
+
+    assert deleted == [], "Invalid JSON message should NOT be deleted"
+
+
+def test_valid_json_would_be_deleted_on_success():
+    """Control: a well-formed event that processes without error is deleted."""
+    deleted = []
+    messages = [_make_sqs_message(json.dumps({"source": "aws.lambda"}))]
+
+    for msg in messages:
+        should_delete = False
+        try:
+            json.loads(msg["Body"])
+            # Simulate successful processing
+            should_delete = True
+        except json.JSONDecodeError:
+            pass
+        if should_delete:
+            deleted.append(msg["ReceiptHandle"])
+
+    assert deleted == ["receipt-abc"]
+
+
+def test_processing_exception_leaves_message_in_queue():
+    """A message that raises during processing must not be deleted."""
+    deleted = []
+    messages = [_make_sqs_message(json.dumps({"source": "aws.cloudwatch"}))]
+
+    for msg in messages:
+        should_delete = False
+        try:
+            json.loads(msg["Body"])
+            raise RuntimeError("Investigation produced no structured result")
+        except json.JSONDecodeError:
+            pass
+        except Exception:
+            pass  # leave in queue
+        if should_delete:
+            deleted.append(msg["ReceiptHandle"])
+
+    assert deleted == [], "Failed processing must not delete the message"
+
+
+# ---------------------------------------------------------------------------
+# Tests: event_infra redrive policy constants
+# ---------------------------------------------------------------------------
+
+def test_max_receive_count_constant():
+    """MAX_RECEIVE_COUNT must match what we verified in AWS."""
+    from agent.event_infra import MAX_RECEIVE_COUNT
+    assert MAX_RECEIVE_COUNT == "5", (
+        "maxReceiveCount changed — update the DLQ in AWS to match "
+        "(aws sqs set-queue-attributes --attribute-names RedrivePolicy)"
+    )
+
+
+def test_dlq_queue_name_constant():
+    from agent.event_infra import DLQ_QUEUE_NAME, QUEUE_NAME
+    assert DLQ_QUEUE_NAME == f"{QUEUE_NAME}-dlq", (
+        "DLQ name must be the main queue name with '-dlq' suffix"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Tests: event consumer loop deletes only on success
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_consumer_loop_skips_delete_on_bad_json(monkeypatch):
+    """event_consumer_loop must not delete a message that fails JSON parsing."""
+    import asyncio
+    from unittest.mock import MagicMock, patch
+
+    deleted_receipts: list[str] = []
+    received_once = False
+
+    def fake_receive(**_kwargs):
+        nonlocal received_once
+        if received_once:
+            # Return empty after first batch so the loop exits
+            raise asyncio.CancelledError()
+        received_once = True
+        return {"Messages": [{"Body": "NOT_VALID_JSON", "ReceiptHandle": "rh-bad"}]}
+
+    def fake_delete(QueueUrl, ReceiptHandle):  # noqa: N803
+        deleted_receipts.append(ReceiptHandle)
+
+    mock_sqs = MagicMock()
+    mock_sqs.receive_message.side_effect = fake_receive
+    mock_sqs.delete_message.side_effect = fake_delete
+
+    with (
+        patch("agent.event_consumer.get_runtime_sqs_queue_url", return_value="https://fake-url"),
+        patch("agent.event_consumer.get_runtime_aws_region", return_value="us-east-1"),
+        patch("boto3.Session") as mock_session,
+    ):
+        mock_session.return_value.client.return_value = mock_sqs
+        from agent.event_consumer import event_consumer_loop
+        try:
+            await event_consumer_loop()
+        except asyncio.CancelledError:
+            pass
+
+    assert "rh-bad" not in deleted_receipts, (
+        "Bad-JSON message receipt should never be passed to delete_message"
+    )
+
+
+@pytest.mark.asyncio
+async def test_consumer_loop_skips_delete_on_processing_error(monkeypatch):
+    """event_consumer_loop must not delete a message when _process_event raises."""
+    import asyncio
+    from unittest.mock import MagicMock, patch
+
+    deleted_receipts: list[str] = []
+    received_once = False
+
+    def fake_receive(**_kwargs):
+        nonlocal received_once
+        if received_once:
+            raise asyncio.CancelledError()
+        received_once = True
+        return {
+            "Messages": [{
+                "Body": json.dumps({"source": "aws.cloudwatch", "detail": {}}),
+                "ReceiptHandle": "rh-fail",
+            }]
+        }
+
+    def fake_delete(QueueUrl, ReceiptHandle):  # noqa: N803
+        deleted_receipts.append(ReceiptHandle)
+
+    async def fake_process_event(_event):
+        raise RuntimeError("Simulated processing failure")
+
+    mock_sqs = MagicMock()
+    mock_sqs.receive_message.side_effect = fake_receive
+    mock_sqs.delete_message.side_effect = fake_delete
+
+    with (
+        patch("agent.event_consumer.get_runtime_sqs_queue_url", return_value="https://fake-url"),
+        patch("agent.event_consumer.get_runtime_aws_region", return_value="us-east-1"),
+        patch("agent.event_consumer._process_event", side_effect=fake_process_event),
+        patch("boto3.Session") as mock_session,
+    ):
+        mock_session.return_value.client.return_value = mock_sqs
+        from agent.event_consumer import event_consumer_loop
+        try:
+            await event_consumer_loop()
+        except asyncio.CancelledError:
+            pass
+
+    assert "rh-fail" not in deleted_receipts, (
+        "Failed-processing message receipt should never be passed to delete_message"
+    )

From cb2275ea50de4a46b5847f9602ae6edcc29963a7 Mon Sep 17 00:00:00 2001
From: AhmadHammad21 <ahmedhammad8765@gmail.com>
Date: Wed, 20 May 2026 18:59:49 +0300
Subject: [PATCH 3/3] docs: update IAM policy split, remove SNS references, add
 SSE stream and investigation_runner notes

Two-policy IAM split (Operational reads *, Setup writes scoped to opendevops-*).
Removes all SNS mentions from docs, monitoring flow, and README roadmap.
Adds SSE stream endpoint to monitoring API reference.
Documents service health DB fallback, investigation_runner, and poller thread pool.
---
 README.md                 | 11 ++---
 docs/event_detection.md   | 52 +++++---------------
 docs/iam_setup.md         | 99 +++++++++++++++++++++++++++------------
 docs/monitoring.md        | 37 +++++----------
 docs/slack_and_polling.md |  5 ++
 5 files changed, 102 insertions(+), 102 deletions(-)

diff --git a/README.md b/README.md
index 40e1039..0f32e79 100644
--- a/README.md
+++ b/README.md
@@ -12,9 +12,8 @@ and gives actionable mitigation plans — without the AWS DevOps Agent price tag
 - **Streaming responses** — FastAPI SSE endpoint streams agent tokens in real time as the LLM reasons; tool calls appear as they complete
 - **Event-driven incident detection** — EventBridge → SQS → long-poll consumer; 9 EventBridge rules cover CloudWatch alarms, ECS task failures, Lambda async errors, RDS events, EC2 state changes, CodePipeline failures, and AWS Health events; uses a DLQ plus database-backed incident claims to avoid duplicate investigations; runs alongside the metric poller — see [docs/event_detection.md](docs/event_detection.md)
 - **Context enrichment** — before the LLM runs, deterministic boto3 calls fetch facts about the affected resource (alarm details, recent logs, function config, etc.) to reduce tool call count and speed up investigations
-- **SNS alert delivery** — after each event-driven investigation, findings are published to a configured SNS topic as structured JSON; fires alongside Slack and Telegram so all channels receive results simultaneously
-- **Monitoring dashboard** — live incident feed showing all event-driven investigations: confidence level (or FAILED badge), affected service, root cause summary, SNS status; each alert links back to its original investigation session via **View investigation** so you can follow up without losing context — see [docs/monitoring.md](docs/monitoring.md)
-- **AWS Configuration settings tab** — admin-only editable tab in Settings for SNS Topic ARN, SQS Queue URL, and AWS Region; shared org-wide via database-backed app config; includes an inline IAM permission checker per service
+- **Monitoring dashboard** — live incident feed showing all event-driven investigations: confidence level (or FAILED badge), affected service, root cause summary; each alert links back to its original investigation session via **View investigation** so you can follow up without losing context; real-time SSE push keeps the page live without polling — see [docs/monitoring.md](docs/monitoring.md)
+- **AWS Configuration settings tab** — admin-only editable tab in Settings for SQS Queue URL and AWS Region; shared org-wide via database-backed app config; includes an inline IAM permission checker per service
 - **Web UI** — React + Vite SPA served by FastAPI:
   - **Chat page** — streaming responses, collapsible tool call inspector, cost/latency card, stop button; supports `?prompt=` deeplink for pre-seeded investigations from the Monitoring dashboard
   - **Session history sidebar** — lists all past conversations; click any to resume with full tool call inspector and cost card restored; new chat and delete (soft) buttons
@@ -251,16 +250,14 @@ docs/                  # Feature reference — auth, schema, skills, databases,
 - [x] **Fun streaming labels** — contextual loading copy ("Digging through CloudTrail…", "Lemonizing metrics…", "Cooking up a root cause…")
 - [x] **Slack & Telegram notifications** — reactive: posts after every investigation to Slack (Block Kit) and/or Telegram (HTML bot message); proactive: background poller checks CloudWatch alarms and Lambda error rates, auto-investigates, and delivers to both channels; set `SLACK_WEBHOOK_URL` and/or `TELEGRAM_BOT_TOKEN`+`TELEGRAM_CHAT_ID` in `.env`; see [docs/telegram.md](docs/telegram.md)
 - [x] **Event-driven incident detection** — EventBridge → SQS → long-poll consumer; 9 EventBridge rules covering CloudWatch alarms, ECS, Lambda, RDS, EC2, CodePipeline, and AWS Health; runs in parallel with the metric poller; see [docs/event_detection.md](docs/event_detection.md)
-- [x] **SNS alert delivery** — findings published to SNS after each event-driven investigation; fires alongside Slack; structured JSON payload for easy downstream processing
 - [x] **Context enrichment** — deterministic boto3 calls per event type before LLM runs; reduces tool call count by front-loading relevant resource facts
-- [x] **Monitoring dashboard** — live incident feed, per-service health summary, alert detail page; "View investigation" opens the original agent session for follow-up; failed investigations flagged separately; see [docs/monitoring.md](docs/monitoring.md)
-- [x] **AWS Configuration settings tab** — admin-only editable tab for SNS/SQS/region config; shared org-wide via database-backed app config; inline IAM permission checker
+- [x] **Monitoring dashboard** — live incident feed with real-time SSE push, per-service health summary (DB-backed, survives restarts), alert detail page; "View investigation" opens the original agent session for follow-up; failed investigations flagged separately; see [docs/monitoring.md](docs/monitoring.md)
+- [x] **AWS Configuration settings tab** — admin-only editable tab for SQS/region config; shared org-wide via database-backed app config; inline IAM permission checker
 
 ### Later
 - [ ] **Observability** — OpenTelemetry traces for agent steps, tool call latency, LLM token usage
 - [ ] **Follow-up question suggestions** — after each investigation completes, generate 3 suggested follow-up questions in the background and surface them in the UI as clickable chips
 - [ ] **Session / user feedback loop** — thumbs up/down on investigations, feed signals back to the agent and to an internal quality dashboard
-- [ ] **Telegram integration** — bot that accepts `/investigate` commands and streams findings back to a chat or group
 - [ ] **Knowledge base** — attach internal runbooks, post-mortems, and architecture docs so the agent grounds answers in org-specific context
 - [ ] **Multi-account AWS** — support multiple AWS profiles per org via `aws_profiles` table (schema already in place)
 - [ ] **Multi-cloud support** — extend tooling to GCP (Cloud Monitoring, Cloud Logging, GKE) and Azure (Monitor, Log Analytics, AKS); unified incident investigation across providers
diff --git a/docs/event_detection.md b/docs/event_detection.md
index 26e20f2..6df6605 100644
--- a/docs/event_detection.md
+++ b/docs/event_detection.md
@@ -19,9 +19,9 @@ AWS Service (ECS, Lambda, RDS, EC2…)
       → SQS queue (OpenDevOps, with DLQ)
           → SQS long-poll consumer (event_consumer_loop)
               → context_collectors (deterministic boto3 enrichment)
-              → agent.ainvoke (LLM investigation)
-              → monitor_store.add_alert (persist result)
-              → SNS publish + Slack notify
+              → investigation_runner.run_investigation (LLM investigation)
+              → monitor_store.add_alert (persist result + SSE push)
+              → Slack / Telegram notify
 ```
 
 ---
@@ -30,14 +30,13 @@ AWS Service (ECS, Lambda, RDS, EC2…)
 
 ### Option A — Init wizard (recommended)
 
-Go to **Settings → AWS Configuration** after first login. Enter your AWS Region and optional SNS Topic ARN, save, then click **Run checks** to verify IAM permissions. Click **Create Infrastructure** to create the SQS queue, EventBridge rules, and aggregate Lambda CloudWatch alarm.
+Go to **Settings → AWS Configuration** after first login. Enter your AWS Region, save, then click **Run checks** to verify IAM permissions. Click **Create Infrastructure** to create the SQS queue, EventBridge rules, and aggregate Lambda CloudWatch alarm.
 
 ### Option B — Manual `.env`
 
 ```bash
-SQS_QUEUE_URL=https://sqs.us-east-1.amazonaws.com/123456789012/opendevops-events
+SQS_QUEUE_URL=https://sqs.us-east-1.amazonaws.com/123456789012/opendevops-agent-events
 EVENT_CONSUMER_ENABLED=true        # or leave unset — consumer also starts if SQS_QUEUE_URL is set
-SNS_TOPIC_ARN=arn:aws:sns:us-east-1:123456789012:opendevops-alerts
 ```
 
 The consumer starts automatically on server startup if either `EVENT_CONSUMER_ENABLED=true`, `SQS_QUEUE_URL` is set, or the init wizard has enabled event infrastructure in persistent app config.
@@ -86,9 +85,9 @@ Context is appended to the investigation prompt, capped at 3 000 chars to stay w
 2. Filters noise: only processes events where `_is_real_failure()` returns true (e.g. skips EC2 `running` state changes, healthy RDS events)
 3. Deduplicates: builds a canonical incident key and atomically claims it in the database before the agent runs
 4. Calls `collect_context(event)` to enrich the prompt
-5. Generates a `session_id` and runs a full agent investigation via `agent.ainvoke`, persisting the session and messages to the database (`source = 'event'`)
-6. Persists the result to the `alerts` table (linked to the session via `session_id`) via `monitor_store.add_alert()`
-7. Delivers findings via SNS and Slack (if configured); failed investigations are flagged with status `'failed'` but are still persisted and notified
+5. Generates a `session_id` and runs a full agent investigation via `investigation_runner.run_investigation`, persisting the session and messages to the database (`source = 'event'`)
+6. Persists the result to the `alerts` table (linked to the session via `session_id`) via `monitor_store.add_alert()`, which also pushes the alert to all active SSE subscribers
+7. Delivers findings via Slack and Telegram (if configured); failed investigations are flagged with status `'failed'` but are still persisted and notified. If the LLM completes the loop without calling `submit_investigation` (e.g. a weak model), a synthetic `_status: failed` result is produced so the alert is always persisted.
 8. Deletes the SQS message only after successful processing, an intentional ignore, or a duplicate claim. Processing failures are left on the queue for retry and eventual DLQ redrive.
 
 The consumer is started as an `asyncio.Task` in the FastAPI lifespan and shut down cleanly on server stop.
@@ -99,37 +98,12 @@ Deduplication uses canonical incident keys rather than hashes of whole event pay
 
 ---
 
-## SNS Alert Delivery
-
-After each event-driven investigation, the result is published to the configured SNS topic (`SNS_TOPIC_ARN`). The message is a JSON payload:
-
-```json
-{
-  "service": "ECS / my-api",
-  "confidence": "HIGH",
-  "error": "Container exited with code 137 (OOM killed)",
-  "resolution": "Increase task memory from 512 MB to 1024 MB..."
-}
-```
-
-SNS subscribers (email, Lambda, SQS fan-out, etc.) receive this payload. Configure subscribers in your AWS console.
-
-Slack notifications fire alongside SNS when `SLACK_WEBHOOK_URL` is set — both channels always receive the same result.
-
----
-
 ## Permission Requirements
 
-The IAM role/user used by the agent needs these additional permissions for event-driven detection:
+The IAM role/user needs the permissions described in [iam_setup.md](iam_setup.md). For event-driven detection specifically:
 
-```
-sns:Publish, sns:GetTopicAttributes
-sqs:CreateQueue, sqs:SetQueueAttributes, sqs:GetQueueAttributes
-sqs:ReceiveMessage, sqs:DeleteMessage, sqs:DeleteQueue, sqs:ListQueues
-events:PutRule, events:PutTargets, events:ListRules
-events:RemoveTargets, events:DeleteRule
-cloudwatch:PutMetricAlarm, cloudwatch:DeleteAlarms, cloudwatch:DescribeAlarms
-```
+- **Policy 1 (Operational)** — `SQSConsume` statement provides `ReceiveMessage`, `DeleteMessage`, `ChangeMessageVisibility` on `opendevops-agent-events`
+- **Policy 2 (Setup)** — `SQSSetup`, `EventBridgeSetup`, and `CloudWatchAlarmSetup` statements with write actions scoped to `opendevops-*` resources
 
 Use **Settings → AWS Configuration → Run checks** to validate all permissions before going live.
 
@@ -158,12 +132,12 @@ For SQLite, schema migrations are applied automatically on first start.
 | File | Purpose |
 |---|---|
 | `src/agent/event_consumer.py` | SQS long-poll loop, filtering, prompt building, delivery |
+| `src/agent/investigation_runner.py` | Shared investigation runner (streaming loop, failed-result fallback) |
 | `src/agent/event_infra.py` | Create/teardown SQS queue + EventBridge rules |
 | `src/agent/context_collectors.py` | Per-event-type boto3 context enrichment |
-| `src/agent/monitor_store.py` | In-process service status + DB-backed alerts, notifications, and incident claims |
+| `src/agent/monitor_store.py` | In-process service status + DB-backed alerts, notifications, and incident claims; SSE subscriber broadcast |
 | `src/agent/init_store.py` | Persist init config to DB-backed `app_config` |
 | `src/agent/permission_checker.py` | IAM permission validation |
-| `src/tools/sns.py` | SNS publish wrapper |
 | `src/api/routers/init.py` | Setup wizard API endpoints |
 | `migrations/005_alerts.sql` | PostgreSQL alerts table + indexes |
 | `migrations/012_incident_claims.sql` | PostgreSQL incident claim table for atomic deduplication |
diff --git a/docs/iam_setup.md b/docs/iam_setup.md
index 4ee5a57..8f7da2d 100644
--- a/docs/iam_setup.md
+++ b/docs/iam_setup.md
@@ -21,34 +21,46 @@ For a typical self-hosted deployment, create an IAM user and paste the keys into
 1. Open **IAM → Users → Create user** in the AWS console
 2. Username: `opendevops-agent` (or any name you prefer)
 3. Select **"Attach policies directly"**
-4. Create and attach the inline policy below (see [Step 2](#step-2--iam-policy))
+4. Create and attach **both** customer-managed policies from [Step 2](#step-2--iam-policies) below
 5. Open the user → **Security credentials → Create access key**
 6. Choose **"Other"** as the use case
 7. Copy the **Access key ID** and **Secret access key** — you won't see the secret again
 
+If you only want the chat interface and manual investigations (no event-driven monitoring), attach only **Policy 1** and skip Policy 2.
+
 ---
 
-## Step 2 — IAM policy
+## Step 2 — IAM policies
+
+Two policies keep the least-privilege boundary clean:
+
+- **Policy 1 (Operational)** — read access across all resources, plus runtime SQS queue polling on the specific event queue. Required for all OpenDevOps features.
+- **Policy 2 (Setup)** — write actions scoped to `opendevops-*` resources only. Required only if you use the Settings → AWS Configuration setup wizard to create SQS/EventBridge/CloudWatch infrastructure.
 
-Create a **customer-managed policy** named `OpenDevOpsAgentPolicy` with this JSON:
+### Policy 1 — OpenDevOpsAgentOperational
 
 ```json
 {
   "Version": "2012-10-17",
   "Statement": [
     {
-      "Sid": "ReadOnlyInvestigation",
+      "Sid": "ReadAll",
       "Effect": "Allow",
       "Action": [
         "cloudwatch:DescribeAlarms",
         "cloudwatch:DescribeAlarmHistory",
         "cloudwatch:GetMetricStatistics",
         "cloudwatch:GetMetricData",
+        "cloudwatch:ListMetrics",
         "logs:DescribeLogGroups",
+        "logs:DescribeLogStreams",
         "logs:FilterLogEvents",
+        "logs:GetLogEvents",
         "logs:StartQuery",
+        "logs:StopQuery",
         "logs:GetQueryResults",
         "cloudtrail:LookupEvents",
+        "cloudtrail:GetTrailStatus",
         "ecs:ListClusters",
         "ecs:DescribeClusters",
         "ecs:ListServices",
@@ -64,42 +76,66 @@ Create a **customer-managed policy** named `OpenDevOpsAgentPolicy` with this JSO
         "rds:DescribeEvents",
         "iam:ListAttachedRolePolicies",
         "iam:ListRolePolicies",
-        "sts:GetCallerIdentity"
+        "sts:GetCallerIdentity",
+        "sqs:ListQueues",
+        "sqs:GetQueueUrl",
+        "sqs:GetQueueAttributes",
+        "events:ListRules",
+        "events:DescribeRule",
+        "events:ListTargetsByRule"
       ],
       "Resource": "*"
     },
     {
-      "Sid": "EventMonitoringSetup",
+      "Sid": "SQSConsume",
+      "Effect": "Allow",
+      "Action": [
+        "sqs:ReceiveMessage",
+        "sqs:DeleteMessage",
+        "sqs:ChangeMessageVisibility"
+      ],
+      "Resource": "arn:aws:sqs:*:*:opendevops-agent-events"
+    }
+  ]
+}
+```
+
+### Policy 2 — OpenDevOpsAgentSetup
+
+```json
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "SQSSetup",
       "Effect": "Allow",
       "Action": [
         "sqs:CreateQueue",
         "sqs:DeleteQueue",
-        "sqs:GetQueueUrl",
-        "sqs:GetQueueAttributes",
         "sqs:SetQueueAttributes",
-        "sqs:ListQueues",
-        "sqs:ReceiveMessage",
-        "sqs:DeleteMessage",
-        "sqs:ChangeMessageVisibility",
-        "sqs:SendMessage",
+        "sqs:SendMessage"
+      ],
+      "Resource": "arn:aws:sqs:*:*:opendevops-*"
+    },
+    {
+      "Sid": "EventBridgeSetup",
+      "Effect": "Allow",
+      "Action": [
         "events:PutRule",
         "events:PutTargets",
-        "events:ListRules",
         "events:RemoveTargets",
-        "events:DeleteRule",
-        "cloudwatch:PutMetricAlarm",
-        "cloudwatch:DeleteAlarms"
+        "events:DeleteRule"
       ],
-      "Resource": "*"
+      "Resource": "arn:aws:events:*:*:rule/opendevops-*"
     },
     {
-      "Sid": "SNSNotifications",
+      "Sid": "CloudWatchAlarmSetup",
       "Effect": "Allow",
       "Action": [
-        "sns:Publish",
-        "sns:GetTopicAttributes"
+        "cloudwatch:PutMetricAlarm",
+        "cloudwatch:DeleteAlarms"
       ],
-      "Resource": "*"
+      "Resource": "arn:aws:cloudwatch:*:*:alarm:opendevops-*"
     }
   ]
 }
@@ -107,15 +143,19 @@ Create a **customer-managed policy** named `OpenDevOpsAgentPolicy` with this JSO
 
 ### What each block does
 
-| Block | Purpose |
-|---|---|
-| `ReadOnlyInvestigation` | Read-only access to CloudWatch, CloudTrail, ECS, Lambda, EC2, RDS, IAM — used by the investigation agent tools |
-| `EventMonitoringSetup` | Create/manage the SQS queue and EventBridge rules during the setup wizard; poll SQS at runtime for incoming events |
-| `SNSNotifications` | Optional — publish alert summaries to an SNS topic after investigations; check topic attributes during permission check. Remove this block if you're not using SNS. |
+| Block | Policy | Purpose |
+|---|---|---|
+| `ReadAll` | Operational | Read access across CloudWatch, logs, CloudTrail, ECS, Lambda, EC2, RDS, IAM, STS, SQS, EventBridge — used by all investigation tools and bash-based AWS CLI calls |
+| `SQSConsume` | Operational | Polls `opendevops-agent-events` at runtime for incoming EventBridge events |
+| `SQSSetup` | Setup | Creates and tears down the SQS queue and DLQ; sends test events via the pipeline test script |
+| `EventBridgeSetup` | Setup | Creates and tears down the 9 EventBridge rules that forward AWS events to SQS |
+| `CloudWatchAlarmSetup` | Setup | Creates and tears down the aggregate `opendevops-lambda-errors-aggregate` CloudWatch alarm |
+
+**Why reads use `*`:** The agent can issue arbitrary AWS CLI commands via its bash tool. Scoping reads to specific resource ARNs would silently break any investigation that touches a resource not on the allowlist. Write actions are scoped to `opendevops-*` because the app only creates infrastructure under that prefix and nothing else.
 
 ### Minimal policy (investigation only, no event monitoring)
 
-If you only want the chat interface and manual investigations — no automatic event detection — you can attach just the `ReadOnlyInvestigation` statement and skip the rest.
+If you only want the chat interface and manual investigations — no automatic event detection — attach only **Policy 1** (`ReadAll` + `SQSConsume`). The `SQSConsume` statement is harmless if you never create the queue.
 
 ---
 
@@ -156,7 +196,6 @@ After starting the app, the setup wizard's **Step 3 (Permission Check)** calls e
 | RDS | No |
 | EC2 | No |
 | IAM / STS | No |
-| SNS | No |
 | CloudTrail | No |
 
 Required services must pass for event monitoring to work. Optional services are used when available — missing permissions will simply return empty results for that service.
@@ -165,4 +204,4 @@ Required services must pass for event monitoring to work. Optional services are
 
 ## Using an IAM role instead of a user
 
-If running on EC2, ECS, or Lambda, attach the same policy to the instance/task role instead. Remove `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_PROFILE` from `.env` — the SDK picks up the role automatically via the instance metadata service.
+If running on EC2, ECS, or Lambda, attach the same policies to the instance/task role instead. Remove `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_PROFILE` from `.env` — the SDK picks up the role automatically via the instance metadata service.
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 89daf8a..b82bf75 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -90,15 +90,14 @@ AWS service emits an event (ECS task stopped, Lambda errors, RDS failover…)
           → collect_context()  — deterministic boto3 pre-fetch (logs, metrics, config)
           → agent.ainvoke()    — full ReAct investigation
               → add_alert()    — persists to alerts table → /monitoring
-              → SNS publish    — fan-out to email, Lambda, etc.
-              → notify_slack() — posts to Slack
+              → notify_slack() — posts to Slack (+ Telegram if configured)
 ```
 
 ### When to use
 
 - You want broad coverage beyond Lambda (ECS, RDS, EC2, GuardDuty, AWS Health)
 - You already have CloudWatch alarms configured on your resources
-- You want SNS fan-out for email/PagerDuty/custom integrations
+- You want near-real-time detection instead of polling-interval latency
 
 ### Why 2–5 minutes for alarm-based events?
 
@@ -158,22 +157,6 @@ Before the LLM runs, `collect_context()` makes deterministic boto3 calls to pre-
 
 ---
 
-## SQS vs SNS — What Each Does
-
-These are often confused. They serve different roles:
-
-| | SQS | SNS |
-|---|---|---|
-| **Direction** | Input to agent | Output from agent |
-| **Role** | Receives EventBridge events for the agent to process | Fan-out investigated results to subscribers |
-| **Who writes to it** | AWS EventBridge rules | Agent, after investigation completes |
-| **Who reads from it** | `event_consumer_loop()` | Email, Lambda, Telegram, custom HTTP endpoints |
-| **Required** | Yes, for event-driven detection | No — optional output channel |
-
-SQS is the inbox. SNS is the outbox. You need SQS for event-driven detection; SNS is optional and only relevant if you want to route investigation results to external systems beyond Slack.
-
----
-
 ## Choosing Between the Two
 
 | Scenario | Recommendation |
@@ -183,7 +166,6 @@ SQS is the inbox. SNS is the outbox. You need SQS for event-driven detection; SN
 | ECS, RDS, EC2, GuardDuty coverage | Event-driven required |
 | Already have CloudWatch alarms | Event-driven — reuses your existing alarms |
 | Want fastest Lambda detection | Both — polling at 1 min catches errors before the alarm trips |
-| Want email/PagerDuty alerts | Event-driven + SNS |
 
 ---
 
@@ -215,7 +197,6 @@ Each alert card shows:
 - **Service** — the affected AWS service and resource name
 - **Error** — the root cause summary from the agent
 - **Time** — when the event was detected
-- **SNS badge** — shown if the finding was published to SNS
 
 Alerts are sorted by most recent first. Click any card to open the Alert Detail page.
 
@@ -225,7 +206,7 @@ Shows the full investigation result:
 
 - Root cause in a highlighted panel
 - Resolution steps produced by the agent
-- Timestamp and SNS notification status
+- Timestamp and detection source
 
 **Investigate / View investigation button** — behaviour depends on whether the investigation session was persisted:
 
@@ -240,7 +221,7 @@ If you open a "View investigation" chat and send at least one follow-up message,
 
 ### Service Health Panel
 
-The top of the page shows tracked services and their last-known status (`healthy`, `error`, `unknown`). Updated in-memory each time the event consumer processes an event for a service.
+The top of the page shows tracked services and their last-known status (`healthy`, `error`, `unknown`). Updated in-memory each time the event consumer processes an event for a service. On app restart, the panel is rebuilt from the most recent alert per service in the database — so status survives restarts without needing to reprocess events.
 
 ### Send Test Event (admin only)
 
@@ -254,8 +235,11 @@ The **Send test event** button in the Monitoring page header pushes a synthetic
 GET  /api/monitoring/alerts?limit=50   → list recent alerts
 GET  /api/monitoring/alerts/{id}       → single alert detail
 GET  /api/monitoring/services          → service health summary
+GET  /api/monitoring/stream            → SSE stream of new alerts (real-time push)
 ```
 
+The SSE stream (`/api/monitoring/stream`) sends a `connected` event on open, an `alert` event for each new alert as it is added, and a `heartbeat` every 20 seconds to keep the connection alive. The frontend subscribes on page load and falls back to 30-second polling if the connection drops.
+
 ---
 
 ## Source Files
@@ -264,9 +248,10 @@ GET  /api/monitoring/services          → service health summary
 |---|---|
 | `frontend/src/pages/MonitoringPage.tsx` | Live incident feed UI |
 | `frontend/src/pages/AlertDetailPage.tsx` | Alert detail + investigate deeplink |
-| `src/api/routers/monitoring.py` | REST endpoints |
-| `src/agent/monitor_store.py` | In-process service status + DB-backed alert, notification, and incident claim persistence |
-| `src/agent/poller.py` | Proactive polling loop |
+| `src/api/routers/monitoring.py` | REST endpoints + SSE stream |
+| `src/agent/monitor_store.py` | In-process service status + DB-backed alert, notification, and incident claim persistence; SSE subscriber broadcast |
+| `src/agent/investigation_runner.py` | Shared investigation runner used by both poller and event consumer |
+| `src/agent/poller.py` | Proactive polling loop (bounded ThreadPoolExecutor, 4 workers) |
 | `src/agent/event_consumer.py` | SQS long-poll consumer |
 | `src/agent/event_infra.py` | Create/teardown SQS + EventBridge + alarm |
 | `src/agent/context_collectors.py` | Per-event-type boto3 context enrichment |
diff --git a/docs/slack_and_polling.md b/docs/slack_and_polling.md
index 26b2683..a7f7410 100644
--- a/docs/slack_and_polling.md
+++ b/docs/slack_and_polling.md
@@ -78,6 +78,9 @@ Calls `list_lambda_functions` (capped at 20 functions) and checks each one's err
 ### Dedup
 The poller uses canonical incident keys and an atomic database claim before the agent runs. If the cooldown (`POLL_REINVESTIGATE_HOURS`) has not elapsed, the alarm or function is skipped. Autonomous polling requires SQLite or PostgreSQL; memory mode does not start the poller.
 
+### Thread isolation
+Boto3 calls in the poller run inside a bounded `ThreadPoolExecutor(max_workers=4)` named `poller`. This prevents poller threads from consuming the default executor used by the rest of the app, and caps concurrent boto3 calls to four threads even if multiple alarms fire simultaneously.
+
 ---
 
 ## When is `submit_investigation` called?
@@ -94,6 +97,8 @@ So the agent calls it exactly like any other tool — it decides when it has eno
 
 The function itself just returns `"Investigation result recorded."` — the real value is in the structured args the LLM was forced to populate.
 
+If the LLM completes the ReAct loop without ever calling `submit_investigation` (a failure mode seen with some weaker models), the investigation runner synthesizes a `_status: failed` result so the alert is always persisted to the monitoring page instead of being silently dropped.
+
 ---
 
 ## Message Format