diff --git a/src/le_agent_sdk/agent/manager.py b/src/le_agent_sdk/agent/manager.py
index 15ea7b7..bc04eb7 100644
--- a/src/le_agent_sdk/agent/manager.py
+++ b/src/le_agent_sdk/agent/manager.py
@@ -82,7 +82,31 @@ def _is_event_authentic(event: dict[str, Any]) -> bool:
         fault affecting every event, and silently returning zero results would
         misrepresent it as "nothing found".
         """
-        if NostrEvent.verify(event):
+        try:
+            verified = NostrEvent.verify(event)
+        except RuntimeError:
+            # Environment/crypto-backend fault affecting every event — propagate
+            # (fail closed, loudly). CryptoBackendUnavailableError is a
+            # RuntimeError subclass, and a hostile event cannot induce a
+            # RuntimeError from verify() (malformed input raises KeyError/
+            # TypeError, dropped below), so propagating RuntimeError surfaces a
+            # backend outage without letting one bad event abort the batch.
+            # Downgrading it to "unauthentic" would silently return zero results
+            # and misrepresent a backend outage as "nothing found".
+            raise
+        except Exception as exc:
+            # A hostile/malformed relay event (missing committed field, wrong
+            # types) can make verify() raise while computing the id. Drop only
+            # that event — one bad event must not abort the whole query.
+            logger.warning(
+                "Dropping malformed relay event %.16s...: %s. "
+                "The relay may be malicious or misbehaving.",
+                event.get("id", "<no id>"),
+                exc,
+            )
+            return False
+
+        if verified:
             return True
 
         logger.warning(
@@ -166,6 +190,16 @@ async def _query_relays(
         for result in results:
             if isinstance(result, list):
                 for event in result:
+                    if not isinstance(event, dict):
+                        # A hostile relay can send a non-dict payload (str/list).
+                        # Drop it before `.get` raises AttributeError out of the
+                        # whole query — one bad event must not DoS discovery.
+                        logger.warning(
+                            "Dropping non-dict relay payload of type %s. "
+                            "The relay may be malicious or misbehaving.",
+                            type(event).__name__,
+                        )
+                        continue
                     event_id = event.get("id", "")
                     if event_id and event_id not in seen_ids:
                         seen_ids.add(event_id)
@@ -219,10 +253,25 @@ async def discover(
         )
 
         raw_events = await self._query_relays([nostr_filter], timeout=timeout)
-        return [
-            AgentCapability.from_nostr_event(e)
-            for e in self._filter_authentic(raw_events)
-        ]
+
+        # Parse each authenticated event INDEPENDENTLY. A malformed tag on one
+        # event (e.g. an unparseable `price` amount) must not abort the whole
+        # batch: a single hostile relay publishing one bad capability event
+        # would otherwise DoS discovery for every agent. Fail closed, loudly —
+        # the offending event is skipped and the skip is logged as a WARNING,
+        # never silently swallowed.
+        capabilities: list[AgentCapability] = []
+        for event in self._filter_authentic(raw_events):
+            try:
+                capabilities.append(AgentCapability.from_nostr_event(event))
+            except Exception as exc:
+                logger.warning(
+                    "Skipping malformed capability event %.16s...: %s. "
+                    "The relay may be malicious or misbehaving.",
+                    event.get("id", "<no id>"),
+                    exc,
+                )
+        return capabilities
 
     async def publish_capability(self, capability: AgentCapability) -> str:
         """Publish a capability advertisement to relays.
diff --git a/tests/test_security_regressions.py b/tests/test_security_regressions.py
index a78f60b..d1d1a51 100644
--- a/tests/test_security_regressions.py
+++ b/tests/test_security_regressions.py
@@ -11,6 +11,7 @@
   5. AgentManager trusting unverified relay events.
 """
 
+import logging
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -437,6 +438,155 @@ async def test_listen_requests_drops_forged_events(self):
         assert len(received) == 1
         assert received[0].pubkey == "aa" * 32, "forged request was yielded to caller"
 
+    @pytest.mark.asyncio
+    async def test_discover_skips_one_malformed_price_and_keeps_the_batch(self, caplog):
+        """Finding 6 (ledger #41): one bad `price` tag must not DoS discovery.
+
+        A single hostile relay publishing one capability event with an
+        unparseable amount (e.g. ``["price", "abc"]``) used to raise ValueError
+        out of discover()'s list comprehension, dropping EVERY capability in the
+        batch — including all the well-formed ones. Parsing must be per-event:
+        the malformed event is skipped (failed closed) and logged (loudly), and
+        the valid capabilities are still returned.
+
+        The malformed event is placed in the MIDDLE of the batch so a naive
+        fix that only survives a trailing bad event would still fail this.
+        """
+        valid_a = {
+            "id": "good-a",
+            "pubkey": "aa" * 32,
+            "created_at": 1700000000,
+            "kind": 38400,
+            "content": "Valid A",
+            "tags": [["d", "svc-a"], ["price", "100", "sats", "per-request"]],
+            "sig": "",
+        }
+        malformed = {
+            "id": "bad-mid",
+            "pubkey": "cc" * 32,
+            "created_at": 1700000001,
+            "kind": 38400,
+            "content": "Malformed price",
+            "tags": [["d", "svc-bad"], ["price", "abc"]],
+            "sig": "",
+        }
+        valid_b = {
+            "id": "good-b",
+            "pubkey": "bb" * 32,
+            "created_at": 1700000002,
+            "kind": 38400,
+            "content": "Valid B",
+            "tags": [["d", "svc-b"], ["price", "200"]],
+            "sig": "",
+        }
+
+        mgr = AgentManager()
+        with patch.object(mgr, "_query_relays", new_callable=AsyncMock) as mock_query:
+            mock_query.return_value = [valid_a, malformed, valid_b]
+            # Authenticity is orthogonal to parsing: stub verify() True so every
+            # event reaches the parse path (the drop-on-forgery path is tested
+            # separately above).
+            with patch.object(NostrEvent, "verify", return_value=True):
+                with caplog.at_level(logging.WARNING):
+                    caps = await mgr.discover()
+
+        # The batch must NOT abort: both well-formed capabilities survive.
+        assert len(caps) == 2
+        assert {c.service_id for c in caps} == {"svc-a", "svc-b"}
+        # The malformed event is skipped, not included.
+        assert "svc-bad" not in {c.service_id for c in caps}
+        # ...and its rejection is loud: a WARNING naming the offending event id.
+        assert any(
+            record.levelno == logging.WARNING and "bad-mid" in record.getMessage()
+            for record in caplog.records
+        ), "malformed event was skipped silently instead of logged"
+
+    @pytest.mark.asyncio
+    async def test_discover_skips_field_missing_event_and_keeps_batch(self, caplog):
+        """Vector A (ledger #41): a dict missing committed fields must not DoS.
+
+        _is_event_authentic -> NostrEvent.verify -> compute_id subscripts
+        pubkey/created_at/kind/tags/content. A relay event missing any of them
+        raised KeyError out of _filter_authentic, killing the whole discover()
+        batch — the same single-hostile-event DoS the price fix set out to
+        close, one step earlier in the pipeline. The malformed event must be
+        dropped (failed closed) and logged (loudly); the two genuinely-signed
+        capabilities must still come back.
+
+        These are REAL signed events (coincurve is available in the test env),
+        and verify() is deliberately NOT stubbed so the malformed event actually
+        reaches the id-computation that raises. Poison event is mid-batch.
+        """
+        signed_a = NostrEvent.create(
+            kind=38400,
+            content="Valid A",
+            tags=[["d", "svc-a"], ["price", "100"]],
+            private_key="11" * 32,
+        )
+        signed_b = NostrEvent.create(
+            kind=38400,
+            content="Valid B",
+            tags=[["d", "svc-b"], ["price", "200"]],
+            private_key="22" * 32,
+        )
+        # Missing pubkey/created_at/tags/content -> verify() raises KeyError.
+        field_missing = {"id": "bad-missing", "kind": 38400}
+
+        mgr = AgentManager()
+        with patch.object(mgr, "_query_relays", new_callable=AsyncMock) as mock_query:
+            mock_query.return_value = [signed_a, field_missing, signed_b]
+            with caplog.at_level(logging.WARNING):
+                caps = await mgr.discover()
+
+        assert len(caps) == 2
+        assert {c.service_id for c in caps} == {"svc-a", "svc-b"}
+        assert "bad-missing" not in {c.event_id for c in caps}
+        assert any(
+            record.levelno == logging.WARNING and "bad-missing" in record.getMessage()
+            for record in caplog.records
+        ), "field-missing event was dropped silently instead of logged"
+
+    @pytest.mark.asyncio
+    async def test_discover_skips_non_dict_relay_payload_and_keeps_batch(self, caplog):
+        """Vector B (ledger #41): a non-dict relay payload must not DoS.
+
+        _query_relays did ``event.get("id", "")`` over whatever the relay
+        returned. A hostile relay sending a str/list instead of an event dict
+        raised AttributeError out of discover(). It must be dropped before the
+        ``.get``, keeping the valid capabilities.
+
+        _query_relay (the PER-relay method) is patched so the REAL _query_relays
+        runs its new isinstance guard. Poison payload is mid-batch.
+        """
+        signed_a = NostrEvent.create(
+            kind=38400,
+            content="Valid A",
+            tags=[["d", "svc-a"], ["price", "100"]],
+            private_key="11" * 32,
+        )
+        signed_b = NostrEvent.create(
+            kind=38400,
+            content="Valid B",
+            tags=[["d", "svc-b"], ["price", "200"]],
+            private_key="22" * 32,
+        )
+
+        mgr = AgentManager()
+
+        async def fake_query_relay(url, filters, timeout):
+            return [signed_a, "not-a-dict", signed_b]
+
+        with patch.object(mgr, "_query_relay", side_effect=fake_query_relay):
+            with caplog.at_level(logging.WARNING):
+                caps = await mgr.discover()
+
+        assert len(caps) == 2
+        assert {c.service_id for c in caps} == {"svc-a", "svc-b"}
+        assert any(
+            record.levelno == logging.WARNING and "non-dict" in record.getMessage().lower()
+            for record in caplog.records
+        ), "non-dict relay payload was dropped silently instead of logged"
+
     @pytest.mark.asyncio
     async def test_listen_requests_surfaces_missing_dep_without_reconnect_storm(self):
         """A missing dep must not be mistaken for a relay fault and retried."""