From b6f5a3e2a56dba44bdad08962463a25677730a68 Mon Sep 17 00:00:00 2001
From: "loki-hermes-agent[bot]"
 <3150032+loki-hermes-agent[bot]@users.noreply.github.com>
Date: Sun, 7 Jun 2026 00:33:22 +0300
Subject: [PATCH 1/2] feat: implement context snapshot fetch (Task 8)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Add context_fetcher.py with Prometheus and Docker socket queries
- Implement fallback chain: Docker → Prometheus → unavailable
- Query peer count and validator count based on client type
- Integrate context snapshot into WebhookHandler before processing
- Add 10 unit tests for context fetcher module
- Update implementation plan to mark Task 8 complete
---
 docs/hermes-implementation-plan.md            |  19 +--
 .../src/webhook_receiver/context_fetcher.py   | 154 ++++++++++++++++++
 .../src/webhook_receiver/server.py            |   9 +
 .../tests/test_context_fetcher.py             | 146 +++++++++++++++++
 4 files changed, 316 insertions(+), 12 deletions(-)
 create mode 100644 webhook-receiver/src/webhook_receiver/context_fetcher.py
 create mode 100644 webhook-receiver/tests/test_context_fetcher.py

diff --git a/docs/hermes-implementation-plan.md b/docs/hermes-implementation-plan.md
index 18ce830..a122b65 100644
--- a/docs/hermes-implementation-plan.md
+++ b/docs/hermes-implementation-plan.md
@@ -86,18 +86,13 @@
 
 ### Task 8: Implement context snapshot fetch
 
-**Objective:** Pre-fetch cheap context (peer count, sync status, container status) at receive time with Prometheus fallback.
-
-**Files:**
-- Modify: `webhook-receiver/src/server.py`
-- Create: `webhook-receiver/src/context_fetcher.py`
-
-**Steps:**
-1. Implement Prometheus queries for peer count, sync status, etc.
-2. Implement Docker socket queries for container status
-3. Implement fallback chain: primary source → Prometheus last value → "unavailable"
-4. Write tests
-5. Verify context snapshot is attached to HermesAlert
+[x] Complete — Created `webhook-receiver/src/webhook_receiver/context_fetcher.py`:
+  - `_query_prometheus`: Queries Prometheus for instant vector values (2s timeout)
+  - `_get_docker_container_status`: Queries Docker socket for container state
+  - `fetch_context_snapshot`: Orchestrates fallback chain (Docker → Prometheus → "unavailable")
+  - Queries peer count and validator count based on client type (lighthouse, prysm, teku, etc.)
+  - Integrated into `WebhookHandler` — context snapshot attached to each `HermesAlert` before processing
+  - 10 unit tests covering success, failure, and fallback scenarios
 
 ---
 
diff --git a/webhook-receiver/src/webhook_receiver/context_fetcher.py b/webhook-receiver/src/webhook_receiver/context_fetcher.py
new file mode 100644
index 0000000..2c400f0
--- /dev/null
+++ b/webhook-receiver/src/webhook_receiver/context_fetcher.py
@@ -0,0 +1,154 @@
+"""Context snapshot fetcher for webhook receiver.
+
+Pre-fetches cheap context (peer count, sync status, container status)
+at receive time with Prometheus fallback.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import socket
+import urllib.parse
+import urllib.request
+import urllib.error
+from typing import Optional, Tuple
+
+from .types import ContextSnapshot
+
+log = logging.getLogger(__name__)
+
+PROMETHEUS_URL = os.environ.get("PROMETHEUS_URL", "http://prometheus:9090")
+DOCKER_SOCKET_PATH = os.environ.get("DOCKER_SOCKET_PATH", "/var/run/docker.sock")
+
+
+def _query_prometheus(query: str) -> Optional[float]:
+    """Query Prometheus for a single instant vector value."""
+    url = f"{PROMETHEUS_URL}/api/v1/query?query={urllib.parse.quote(query)}"
+    try:
+        req = urllib.request.Request(url, method="GET")
+        with urllib.request.urlopen(req, timeout=2) as resp:
+            data = json.loads(resp.read())
+            result = data.get("data", {}).get("result", [])
+            if result and len(result) > 0:
+                return float(result[0]["value"][1])
+    except (
+        urllib.error.URLError,
+        json.JSONDecodeError,
+        KeyError,
+        ValueError,
+        TypeError,
+    ) as e:
+        log.debug("Prometheus query failed for %s: %s", query, e)
+    return None
+
+
+def _get_docker_container_status(container_name: str) -> Tuple[Optional[str], Optional[str]]:
+    """
+    Query Docker socket for container status.
+    Returns (status, note).
+    """
+    api_path = f"/containers/{container_name}/json"
+    request = f"GET {api_path} HTTP/1.1\r\nHost: docker\r\n\r\n"
+    
+    try:
+        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+            sock.settimeout(2.0)
+            sock.connect(DOCKER_SOCKET_PATH)
+            sock.sendall(request.encode('utf-8'))
+            
+            response = b""
+            while True:
+                chunk = sock.recv(4096)
+                if not chunk:
+                    break
+                response += chunk
+                
+            header_end = response.find(b"\r\n\r\n")
+            if header_end == -1:
+                return None, "malformed response"
+            
+            body = response[header_end + 4:]
+            data = json.loads(body.decode('utf-8'))
+            state = data.get("State", {})
+            status = state.get("Status", "unknown")
+            
+            if status == "running":
+                return "running", None
+            else:
+                return status, f"container is {status}"
+    except (socket.error, json.JSONDecodeError, UnicodeDecodeError, OSError) as e:
+        log.debug("Docker socket query failed for %s: %s", container_name, e)
+        return None, f"docker socket error: {type(e).__name__}"
+
+
+def fetch_context_snapshot(
+    host: str,
+    container: Optional[str] = None,
+    client: Optional[str] = None,
+) -> ContextSnapshot:
+    """
+    Fetch context snapshot for an alert.
+    
+    Fallback chain:
+    1. Primary source (Docker socket for container status, direct metrics)
+    2. Prometheus last value
+    3. "unavailable"
+    """
+    snapshot = ContextSnapshot()
+    fallback_used = False
+    
+    # 1. Container status
+    if container:
+        status, note = _get_docker_container_status(container)
+        if status:
+            snapshot.container_status = status
+            if note:
+                snapshot.container_status_note = note
+        else:
+            # Fallback to Prometheus: check if container is up
+            prom_up = _query_prometheus(f'up{{container="{container}"}}')
+            if prom_up == 1.0:
+                snapshot.container_status = "running"
+                snapshot.container_status_note = "docker socket unreachable, using prometheus fallback"
+                fallback_used = True
+            else:
+                snapshot.container_status = "unavailable"
+                snapshot.container_status_note = "docker socket and prometheus unreachable"
+                fallback_used = True
+
+    # 2. Peer count
+    peer_queries = []
+    if client == "lighthouse":
+        peer_queries = ["lighthouse_peers", "beacon_peers"]
+    elif client in ("prysm", "teku", "nimbus", "lodestar"):
+        peer_queries = ["beacon_peers", "p2p_peers"]
+    
+    for q in peer_queries:
+        peers = _query_prometheus(q)
+        if peers is not None:
+            snapshot.peer_count = int(peers)
+            break
+    else:
+        if peer_queries:
+            fallback_used = True
+
+    # 3. Validator count
+    val_queries = []
+    if client == "lighthouse":
+        val_queries = ["lighthouse_validator_count", "validator_count"]
+    elif client in ("prysm", "teku"):
+        val_queries = ["beacon_validators_total", "validator_count"]
+    
+    for q in val_queries:
+        val_count = _query_prometheus(q)
+        if val_count is not None:
+            snapshot.validator_count = int(val_count)
+            break
+    else:
+        if val_queries:
+            fallback_used = True
+
+    snapshot.prometheus_fallback_used = fallback_used
+    return snapshot
diff --git a/webhook-receiver/src/webhook_receiver/server.py b/webhook-receiver/src/webhook_receiver/server.py
index 93f2c1a..73b33bd 100644
--- a/webhook-receiver/src/webhook_receiver/server.py
+++ b/webhook-receiver/src/webhook_receiver/server.py
@@ -14,6 +14,7 @@
 
 from aiohttp import web
 
+from .context_fetcher import fetch_context_snapshot
 from .dedup import DedupLookup, should_process
 from .schema import ValidationError, validate_alertmanager_payload
 from .storm_protection import StormTracker
@@ -78,6 +79,14 @@ async def handle_webhook(self, request: web.Request) -> web.Response:
         if not alerts:
             return web.json_response({"status": "ok", "alerts_processed": 0})
 
+        # Pre-fetch context snapshot for each alert
+        for alert in alerts:
+            alert.context_snapshot = fetch_context_snapshot(
+                host=alert.host,
+                container=alert.container,
+                client=alert.client,
+            )
+
         offsets = []
         deduped_ids = []
         bundled_ids = []
diff --git a/webhook-receiver/tests/test_context_fetcher.py b/webhook-receiver/tests/test_context_fetcher.py
new file mode 100644
index 0000000..6167748
--- /dev/null
+++ b/webhook-receiver/tests/test_context_fetcher.py
@@ -0,0 +1,146 @@
+"""Tests for context_fetcher module."""
+
+import json
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from webhook_receiver.context_fetcher import (
+    _query_prometheus,
+    _get_docker_container_status,
+    fetch_context_snapshot,
+)
+
+
+class TestQueryPrometheus:
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_success(self, mock_urlopen):
+        mock_resp = MagicMock()
+        mock_resp.read.return_value = json.dumps({
+            "data": {
+                "result": [{"value": [1704067200, "42.5"]}]
+            }
+        }).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_resp
+        
+        result = _query_prometheus("lighthouse_peers")
+        assert result == 42.5
+
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_failure(self, mock_urlopen):
+        import urllib.error
+        mock_urlopen.side_effect = urllib.error.URLError("Connection refused")
+        result = _query_prometheus("lighthouse_peers")
+        assert result is None
+
+    @patch("webhook_receiver.context_fetcher.urllib.request.urlopen")
+    def test_query_prometheus_empty_result(self, mock_urlopen):
+        mock_resp = MagicMock()
+        mock_resp.read.return_value = json.dumps({
+            "data": {
+                "result": []
+            }
+        }).encode()
+        mock_urlopen.return_value.__enter__.return_value = mock_resp
+        
+        result = _query_prometheus("lighthouse_peers")
+        assert result is None
+
+
+class TestGetDockerContainerStatus:
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_running(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+        
+        response = (
+            b"HTTP/1.1 200 OK\r\n"
+            b"Content-Type: application/json\r\n"
+            b"\r\n"
+            b'{"State": {"Status": "running"}}'
+        )
+        mock_sock.recv.side_effect = [response, b""]
+        
+        status, note = _get_docker_container_status("consensus")
+        assert status == "running"
+        assert note is None
+
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_exited(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+        
+        response = (
+            b"HTTP/1.1 200 OK\r\n"
+            b"Content-Type: application/json\r\n"
+            b"\r\n"
+            b'{"State": {"Status": "exited"}}'
+        )
+        mock_sock.recv.side_effect = [response, b""]
+        
+        status, note = _get_docker_container_status("consensus")
+        assert status == "exited"
+        assert "exited" in note
+
+    @patch("webhook_receiver.context_fetcher.socket.socket")
+    def test_docker_status_connection_refused(self, mock_socket_cls):
+        mock_sock = MagicMock()
+        mock_sock.connect.side_effect = ConnectionRefusedError()
+        mock_socket_cls.return_value.__enter__.return_value = mock_sock
+        
+        status, note = _get_docker_container_status("consensus")
+        assert status is None
+        assert "ConnectionRefusedError" in note
+
+
+class TestFetchContextSnapshot:
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_success_no_fallback(self, mock_prom, mock_docker):
+        mock_docker.return_value = ("running", None)
+        mock_prom.return_value = 50.0
+        
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+        
+        assert snapshot.container_status == "running"
+        assert snapshot.container_status_note is None
+        assert snapshot.peer_count == 50
+        assert snapshot.prometheus_fallback_used is False
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_docker_fallback_to_prometheus(self, mock_prom, mock_docker):
+        mock_docker.return_value = (None, "docker socket error")
+        # Calls: 1. container up, 2. lighthouse_peers, 3. lighthouse_validator_count
+        mock_prom.side_effect = [1.0, 50.0, 3.0]
+        
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+        
+        assert snapshot.container_status == "running"
+        assert "prometheus fallback" in snapshot.container_status_note
+        assert snapshot.peer_count == 50
+        assert snapshot.validator_count == 3
+        assert snapshot.prometheus_fallback_used is True
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_all_unavailable(self, mock_prom, mock_docker):
+        mock_docker.return_value = (None, "error")
+        mock_prom.return_value = None
+        
+        snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
+        
+        assert snapshot.container_status == "unavailable"
+        assert snapshot.peer_count is None
+        assert snapshot.prometheus_fallback_used is True
+
+    @patch("webhook_receiver.context_fetcher._get_docker_container_status")
+    @patch("webhook_receiver.context_fetcher._query_prometheus")
+    def test_fetch_no_container(self, mock_prom, mock_docker):
+        mock_prom.return_value = 50.0
+        
+        snapshot = fetch_context_snapshot("host1", container=None, client="lighthouse")
+        
+        assert snapshot.container_status is None
+        assert snapshot.peer_count == 50
+        assert snapshot.prometheus_fallback_used is False

From eb557028ebdc4abd6ee98ea8e5d22df8002d3b20 Mon Sep 17 00:00:00 2001
From: "loki-hermes-agent[bot]"
 <3150032+loki-hermes-agent[bot]@users.noreply.github.com>
Date: Sun, 7 Jun 2026 00:51:18 +0300
Subject: [PATCH 2/2] fix: address review feedback on context snapshot fetch

- Add Connection: close to Docker socket request to prevent hanging
- Remove broken Prometheus container fallback (up{container=...} doesn't work in eth-docker)
- Add nimbus and lodestar to validator count queries
- Add test assertion to verify Connection: close header is sent
---
 .../src/webhook_receiver/context_fetcher.py   | 22 +++++++++----------
 .../tests/test_context_fetcher.py             | 12 ++++++----
 2 files changed, 18 insertions(+), 16 deletions(-)

diff --git a/webhook-receiver/src/webhook_receiver/context_fetcher.py b/webhook-receiver/src/webhook_receiver/context_fetcher.py
index 2c400f0..665e83f 100644
--- a/webhook-receiver/src/webhook_receiver/context_fetcher.py
+++ b/webhook-receiver/src/webhook_receiver/context_fetcher.py
@@ -50,7 +50,12 @@ def _get_docker_container_status(container_name: str) -> Tuple[Optional[str], Op
     Returns (status, note).
     """
     api_path = f"/containers/{container_name}/json"
-    request = f"GET {api_path} HTTP/1.1\r\nHost: docker\r\n\r\n"
+    request = (
+        f"GET {api_path} HTTP/1.1\r\n"
+        f"Host: docker\r\n"
+        f"Connection: close\r\n"
+        f"\r\n"
+    )
     
     try:
         with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
@@ -107,16 +112,9 @@ def fetch_context_snapshot(
             if note:
                 snapshot.container_status_note = note
         else:
-            # Fallback to Prometheus: check if container is up
-            prom_up = _query_prometheus(f'up{{container="{container}"}}')
-            if prom_up == 1.0:
-                snapshot.container_status = "running"
-                snapshot.container_status_note = "docker socket unreachable, using prometheus fallback"
-                fallback_used = True
-            else:
-                snapshot.container_status = "unavailable"
-                snapshot.container_status_note = "docker socket and prometheus unreachable"
-                fallback_used = True
+            snapshot.container_status = "unavailable"
+            snapshot.container_status_note = "docker socket unreachable"
+            fallback_used = True
 
     # 2. Peer count
     peer_queries = []
@@ -138,7 +136,7 @@ def fetch_context_snapshot(
     val_queries = []
     if client == "lighthouse":
         val_queries = ["lighthouse_validator_count", "validator_count"]
-    elif client in ("prysm", "teku"):
+    elif client in ("prysm", "teku", "nimbus", "lodestar"):
         val_queries = ["beacon_validators_total", "validator_count"]
     
     for q in val_queries:
diff --git a/webhook-receiver/tests/test_context_fetcher.py b/webhook-receiver/tests/test_context_fetcher.py
index 6167748..e1fed3f 100644
--- a/webhook-receiver/tests/test_context_fetcher.py
+++ b/webhook-receiver/tests/test_context_fetcher.py
@@ -64,6 +64,10 @@ def test_docker_status_running(self, mock_socket_cls):
         status, note = _get_docker_container_status("consensus")
         assert status == "running"
         assert note is None
+        
+        # Verify Connection: close header is sent to prevent hanging
+        sent_request = mock_sock.sendall.call_args[0][0].decode('utf-8')
+        assert "Connection: close\r\n" in sent_request
 
     @patch("webhook_receiver.context_fetcher.socket.socket")
     def test_docker_status_exited(self, mock_socket_cls):
@@ -111,13 +115,13 @@ def test_fetch_success_no_fallback(self, mock_prom, mock_docker):
     @patch("webhook_receiver.context_fetcher._query_prometheus")
     def test_fetch_docker_fallback_to_prometheus(self, mock_prom, mock_docker):
         mock_docker.return_value = (None, "docker socket error")
-        # Calls: 1. container up, 2. lighthouse_peers, 3. lighthouse_validator_count
-        mock_prom.side_effect = [1.0, 50.0, 3.0]
+        # Calls: 1. lighthouse_peers, 2. lighthouse_validator_count
+        mock_prom.side_effect = [50.0, 3.0]
         
         snapshot = fetch_context_snapshot("host1", "consensus", "lighthouse")
         
-        assert snapshot.container_status == "running"
-        assert "prometheus fallback" in snapshot.container_status_note
+        assert snapshot.container_status == "unavailable"
+        assert snapshot.container_status_note == "docker socket unreachable"
         assert snapshot.peer_count == 50
         assert snapshot.validator_count == 3
         assert snapshot.prometheus_fallback_used is True