fix(gateway): don't mark an entire chat dead on thread/message-level not_found

#55115 added the dead-target registry so confirmed-dead delivery targets are
short-circuited. Its documented scope (gateway/dead_targets.py) is deliberately
narrow: only *whole-chat* deaths -- the `forbidden` and chat-level `not_found`
(`chat not found`) kinds -- should be recorded; "Thread/topic-level not_found is
NOT recorded here ... a deleted topic does not mean the parent chat is dead."

But the implementation doesn't honor that scope. classify_send_error collapses
chat-level "chat not found" AND thread/message-level not_found ("thread not
found", "topic_deleted", "message_id_invalid", "message to edit/reply not
found") into one "not_found" kind, _DEAD_ERROR_KINDS contains "not_found"
wholesale, and deliver()'s except marks the PARENT chat_id dead. So a single
deleted Telegram topic or edited-away message permanently marks the entire chat
(and every future scheduled / cron / agent delivery to it) dead -- silently. The
adapter self-heal the docstring relies on only covers the non-private-group
thread retry; named-DM-topic and message-level failures propagate to deliver()'s
except and wrongly kill the whole chat.

Add is_chat_level_not_found() (factoring the not_found substrings into chat-level
vs sub-chat-level constants) and gate the delivery dead-path: a "not_found" only
marks the target dead when it is chat-level. classify_send_error's public
contract is unchanged (still returns "not_found" for every shape); only the
mark_dead decision is refined, restoring the registry's documented scope.

Cross-platform: telegram/slack/discord delivery all flow through
classify_send_error -> mark_dead. Adds regression tests through the real
deliver() path plus helper/classifier units.

gateway/delivery.py

@@ -123,11 +123,19 @@ def _classify_dead_from_error_text(error_text: Optional[str]) -> Optional[str]:
     if not error_text:
         return None
     try:
-        from .platforms.base import classify_send_error
+        from .platforms.base import classify_send_error, is_chat_level_not_found
     except Exception:  # pragma: no cover - import guard
         return None
     kind = classify_send_error(None, error_text=error_text)
-    return kind if DeadTargetRegistry.is_dead_error_kind(kind) else None
+    if not DeadTargetRegistry.is_dead_error_kind(kind):
+        return None
+    # ``not_found`` collapses chat-level and thread/topic/message-level failures.
+    # Only a whole-chat not_found means the target is dead — a deleted forum topic
+    # or an edited-away message must not mark the entire chat (and all of its future
+    # deliveries) dead.  See gateway.dead_targets' documented scope.
+    if kind == "not_found" and not is_chat_level_not_found(error_text):
+        return None
+    return kind
 
 
 @dataclass

gateway/platforms/base.py

@@ -1914,6 +1914,22 @@ SEND_ERROR_KINDS = frozenset(
     }
 )
 
+# ``not_found`` substrings split by blast radius.  A *chat-level* not_found means
+# the chat/user/group itself is gone, so the whole target is dead.  A
+# *thread/topic/message-level* not_found (a deleted forum topic, an edited-away
+# message) leaves the parent chat reachable — it must NOT mark the whole chat
+# dead.  ``classify_send_error`` collapses both into ``"not_found"``;
+# ``is_chat_level_not_found`` recovers the distinction for the dead-target path.
+# See gateway.dead_targets.
+_CHAT_LEVEL_NOT_FOUND_SUBSTRINGS = ("chat not found",)
+_SUBCHAT_NOT_FOUND_SUBSTRINGS = (
+    "message to edit not found",
+    "message to reply not found",
+    "thread not found",
+    "topic_deleted",
+    "message_id_invalid",
+)
+
 
 def classify_send_error(exc: Optional[BaseException], error_text: str = "") -> str:
     """Map a send exception / error string to a :data:`SEND_ERROR_KINDS` value.
@@ -1953,13 +1969,8 @@ def classify_send_error(exc: Optional[BaseException], error_text: str = "") -> s
         or "not a member" in blob
     ):
         return "forbidden"
-    if (
-        "chat not found" in blob
-        or "message to edit not found" in blob
-        or "message to reply not found" in blob
-        or "thread not found" in blob
-        or "topic_deleted" in blob
-        or "message_id_invalid" in blob
+    if any(s in blob for s in _CHAT_LEVEL_NOT_FOUND_SUBSTRINGS) or any(
+        s in blob for s in _SUBCHAT_NOT_FOUND_SUBSTRINGS
     ):
         return "not_found"
     if (
@@ -1977,6 +1988,27 @@ def classify_send_error(exc: Optional[BaseException], error_text: str = "") -> s
     return "unknown"
 
 
+def is_chat_level_not_found(error_text: str = "", exc: Optional[BaseException] = None) -> bool:
+    """Whether a ``not_found`` failure means the *whole chat* is gone.
+
+    :func:`classify_send_error` collapses chat-level and thread/topic/message-level
+    not_found into the single ``"not_found"`` kind.  Only the chat-level case (the
+    chat/user/group no longer exists) should mark a delivery target dead; a deleted
+    forum topic or an edited-away message leaves the parent chat reachable.  When
+    both a chat-level and a sub-chat marker are present, the sub-chat reading wins
+    (conservative: never kill a chat that may still be reachable).
+    """
+    parts = []
+    if error_text:
+        parts.append(error_text)
+    if exc is not None:
+        parts.append(str(exc))
+    blob = " ".join(parts).lower()
+    if any(s in blob for s in _SUBCHAT_NOT_FOUND_SUBSTRINGS):
+        return False
+    return any(s in blob for s in _CHAT_LEVEL_NOT_FOUND_SUBSTRINGS)
+
+
 class EphemeralReply(str):
     """System-notice reply that auto-deletes after a TTL.
 

tests/gateway/test_dead_targets.py

@@ -167,3 +167,82 @@ async def test_shared_registry_is_used_when_injected(isolate):
     # Injected registry's pre-existing flag short-circuits before any send.
     assert res["telegram:500"]["skipped"] == "dead_target"
     assert adapter.calls == []
+
+
+# --------------------------------------------------------------------------
+# not_found blast radius: chat-level kills the chat, thread/message-level must not
+# --------------------------------------------------------------------------
+
+class RaisingAdapter:
+    """Raises a fixed error message on every send."""
+
+    def __init__(self, message):
+        self.message = message
+        self.calls = []
+
+    async def send(self, chat_id, content, metadata=None):
+        self.calls.append(chat_id)
+        raise RuntimeError(self.message)
+
+
+_SUBCHAT_NOT_FOUND_MESSAGES = [
+    "Bad Request: message thread not found",
+    "Bad Request: TOPIC_DELETED",
+    "Bad Request: message to edit not found",
+    "Bad Request: message to reply not found",
+    "Bad Request: MESSAGE_ID_INVALID",
+]
+
+
+@pytest.mark.asyncio
+async def test_chat_level_not_found_marks_target_dead(isolate):
+    # "chat not found" -> the whole chat/user/group is gone, so it is dead
+    # (same blast radius as forbidden).
+    adapter = RaisingAdapter("Bad Request: chat not found")
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    target = DeliveryTarget.parse("telegram:100")
+
+    res = await router.deliver("hi", [target])
+    assert res["telegram:100"]["success"] is False
+    assert router.dead_targets.is_dead("telegram", "100") is True
+
+
+@pytest.mark.parametrize("message", _SUBCHAT_NOT_FOUND_MESSAGES)
+@pytest.mark.asyncio
+async def test_thread_or_message_level_not_found_does_not_mark_chat_dead(isolate, message):
+    # A deleted forum topic / edited-away message is NOT a whole-chat death: marking
+    # the parent chat dead would silently short-circuit every future delivery to it.
+    adapter = RaisingAdapter(message)
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    target = DeliveryTarget.parse("telegram:200")
+
+    res = await router.deliver("hi", [target])
+    assert res["telegram:200"]["success"] is False
+    assert router.dead_targets.is_dead("telegram", "200") is False
+
+
+class TestNotFoundBlastRadius:
+    def test_is_chat_level_not_found_chat_level(self):
+        from gateway.platforms.base import is_chat_level_not_found
+
+        assert is_chat_level_not_found("Bad Request: chat not found") is True
+
+    @pytest.mark.parametrize("message", _SUBCHAT_NOT_FOUND_MESSAGES)
+    def test_is_chat_level_not_found_subchat(self, message):
+        from gateway.platforms.base import is_chat_level_not_found
+
+        assert is_chat_level_not_found(message) is False
+
+    def test_subchat_marker_wins_when_both_present(self):
+        from gateway.platforms.base import is_chat_level_not_found
+
+        # Conservative: if a sub-chat marker is present, never kill the whole chat.
+        assert is_chat_level_not_found("chat not found; message thread not found") is False
+
+    def test_classify_dead_from_error_text_gates_not_found(self):
+        from gateway.delivery import _classify_dead_from_error_text
+
+        assert _classify_dead_from_error_text("Forbidden: bot was blocked by the user") == "forbidden"
+        assert _classify_dead_from_error_text("Bad Request: chat not found") == "not_found"
+        assert _classify_dead_from_error_text("Bad Request: message thread not found") is None
+        assert _classify_dead_from_error_text("httpx.ReadTimeout: connection timed out") is None