refactor(gateway): second-pass compaction of delivery ledger/router, control socket, dead targets

gateway/control_socket.py

@@ -53,9 +53,7 @@ def _fallback_socket_path(home: Path) -> Path:
     then ``/tmp`` (POSIX); if nothing fits the tempdir candidate is returned anyway — bind fails
     non-fatally and consumers use the scan layer."""
     name = f"hermes-gw-{_home_hash(home)}.sock"
-    candidates = [Path(tempfile.gettempdir()) / name]
-    if not _IS_WINDOWS:
-        candidates.append(Path("/tmp") / name)
+    candidates = [Path(tempfile.gettempdir()) / name] + ([] if _IS_WINDOWS else [Path("/tmp") / name])
     return next((c for c in candidates if _fits_sun_path(c)), candidates[0])
 
 
@@ -70,12 +68,11 @@ def resolve_client_socket_path(home: Path) -> Optional[Path]:
     direct = Path(home) / _SOCKET_FILENAME
     if direct.exists():
         return direct
-    pointer = Path(home) / _POINTER_FILENAME
     with contextlib.suppress(OSError):
-        if pointer.is_file():
-            target = pointer.read_text(encoding="utf-8").strip()
-            if target and Path(target).exists():
-                return Path(target)
+        pointer = Path(home) / _POINTER_FILENAME
+        target = pointer.read_text(encoding="utf-8").strip() if pointer.is_file() else ""
+        if target and Path(target).exists():
+            return Path(target)
     return None
 
 
@@ -89,9 +86,7 @@ def _detect_supervisor() -> str:
         return "launchd"
     if env.get("HERMES_DESKTOP_MANAGED"):
         return "desktop"
-    if "--external-supervisor" in sys.argv:
-        return "external"
-    return "manual"
+    return "external" if "--external-supervisor" in sys.argv else "manual"
 
 
 def build_identify_payload() -> dict[str, Any]:
@@ -102,9 +97,7 @@ def build_identify_payload() -> dict[str, Any]:
         "protocol": CONTROL_PROTOCOL_VERSION,
         **{k: record.get(k) for k in ("kind", "pid", "start_time", "hermes_home")},
         "profile": _profile_label_for_home(record.get("hermes_home") or ""),
-        "supervisor": _detect_supervisor(),
-    }
-    payload.update(_get_code_identity_fields())
+        "supervisor": _detect_supervisor(), **_get_code_identity_fields()}
     with contextlib.suppress(Exception):
         # served_profiles (multiplex mode) is stamped into runtime status by the runner.
         served = (read_runtime_status() or {}).get("served_profiles")
@@ -135,9 +128,8 @@ class GatewayControlServer:
         self._pipe_server: Any = None  # Windows proactor pipe server
         self._bind_path: Optional[Path] = None
         self._pointer_file: Optional[Path] = None
-        self._handlers: dict[str, Callable[[], dict[str, Any]]] = {"identify": build_identify_payload,
-                                                                   "status": build_status_payload}
-        self._handlers.update(verb_handlers or {})
+        self._handlers: dict[str, Callable[[], dict[str, Any]]] = {
+            "identify": build_identify_payload, "status": build_status_payload, **(verb_handlers or {})}
 
     async def start(self) -> bool:
         """Bind and start serving. Returns True on success, False otherwise."""
@@ -196,10 +188,9 @@ class GatewayControlServer:
 
     def cleanup_files(self) -> None:
         """Best-effort removal of socket + pointer files (atexit-safe)."""
-        for path in (self._bind_path, self._pointer_file):
-            if path is not None:
-                with contextlib.suppress(OSError):
-                    path.unlink(missing_ok=True)
+        for path in filter(None, (self._bind_path, self._pointer_file)):
+            with contextlib.suppress(OSError):
+                path.unlink(missing_ok=True)
 
     def handle_request_line(self, raw: bytes) -> bytes:
         """One JSON request line -> one JSON response line. Never raises (shared by POSIX + pipe)."""
@@ -261,11 +252,9 @@ class _PipeControlProtocol(asyncio.Protocol):
         self._buffer.extend(data)
         if len(self._buffer) > _MAX_REQUEST_BYTES:
             self._transport.close()
-            return
-        if b"\n" in self._buffer:
-            line, _, _ = bytes(self._buffer).partition(b"\n")
+        elif b"\n" in self._buffer:
             try:
-                self._transport.write(self._server.handle_request_line(line))
+                self._transport.write(self._server.handle_request_line(bytes(self._buffer).partition(b"\n")[0]))
             finally:
                 self._transport.close()
 
@@ -278,15 +267,10 @@ def query_gateway_control(home: Path, verb: str, *, timeout: float = _DEFAULT_CL
     query = _query_windows_pipe if _IS_WINDOWS else _query_unix_socket
     try:
         raw = query(Path(home), request, timeout)
+        response = json.loads(raw.decode("utf-8")) if raw else None
     except Exception:
         return None
-    try:
-        response = json.loads(raw.decode("utf-8")) if raw else None
-    except (ValueError, UnicodeDecodeError):
-        return None
-    if not isinstance(response, dict) or response.get("ok") is not True:
-        return None
-    result = response.get("result")
+    result = response.get("result") if isinstance(response, dict) and response.get("ok") is True else None
     return result if isinstance(result, dict) else None
 
 
@@ -295,30 +279,24 @@ def _read_response_line(read: Callable[[], bytes], deadline: float) -> Optional[
     chunks: list[bytes] = []
     while time.monotonic() < deadline:
         chunk = read()
-        if not chunk:
-            break
         chunks.append(chunk)
-        if b"\n" in chunk:
+        if not chunk or b"\n" in chunk:
             break
         if sum(len(c) for c in chunks) > _MAX_RESPONSE_BYTES:
             return None
-    line, _, _ = b"".join(chunks).partition(b"\n")
-    return line or None
+    return b"".join(chunks).partition(b"\n")[0] or None
 
 
 def _query_unix_socket(home: Path, request: bytes, timeout: float) -> Optional[bytes]:
     path = resolve_client_socket_path(home)
     if path is None:
         return None
-    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+    # OSError covers ConnectionRefusedError / FileNotFoundError on connect and socket.timeout on read.
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock, contextlib.suppress(OSError):
         sock.settimeout(timeout)
-        try:
-            sock.connect(str(path))
-        except OSError:  # incl. ConnectionRefusedError / FileNotFoundError
-            return None
+        sock.connect(str(path))
         sock.sendall(request)
-        with contextlib.suppress(socket.timeout):
-            return _read_response_line(lambda: sock.recv(65536), time.monotonic() + timeout)
+        return _read_response_line(lambda: sock.recv(65536), time.monotonic() + timeout)
     return None
 
 

gateway/dead_targets.py

@@ -40,9 +40,8 @@ def classify_dead_error(error_text: Optional[str]) -> Optional[str]:
     except Exception:  # pragma: no cover - import guard
         return None
     kind = classify_send_error(None, error_text=error_text)
-    if kind not in _DEAD_ERROR_KINDS or (kind == "not_found" and not is_chat_level_not_found(error_text=error_text)):
-        return None
-    return kind
+    dead = kind in _DEAD_ERROR_KINDS and (kind != "not_found" or is_chat_level_not_found(error_text=error_text))
+    return kind if dead else None
 
 
 class DeadTargetRegistry:
@@ -54,13 +53,10 @@ class DeadTargetRegistry:
         self._dead: Dict[str, Dict[str, object]] = {}
         self._path = path if path is not None else get_hermes_home() / "gateway" / "dead_targets.json"
         try:
-            if self._path.exists():
-                raw = json.loads(self._path.read_text(encoding="utf-8"))
-                if isinstance(raw, dict):
-                    self._dead = {k: v for k, v in raw.items() if isinstance(v, dict)}
+            raw = json.loads(self._path.read_text(encoding="utf-8")) if self._path.exists() else {}
+            self._dead = {k: v for k, v in raw.items() if isinstance(v, dict)} if isinstance(raw, dict) else {}
         except (OSError, ValueError) as exc:
             logger.debug("dead_targets: could not load %s (%s) — starting empty", self._path, exc)
-            self._dead = {}
 
     def _flush_locked(self) -> None:
         try:
@@ -96,9 +92,8 @@ class DeadTargetRegistry:
             return False
         key = _normalize(platform, chat_id)
         with self._lock:
-            if key in self._dead:
-                del self._dead[key]
-                self._flush_locked()
-                logger.info("dead_targets: cleared %s (delivery succeeded again)", key)
-                return True
-        return False
+            if self._dead.pop(key, None) is None:
+                return False
+            self._flush_locked()
+            logger.info("dead_targets: cleared %s (delivery succeeded again)", key)
+        return True

gateway/delivery.py

@@ -51,9 +51,8 @@ class DeliveryTransport:
     async def send(self, logical_platform: Platform, chat_id: str, content: str,
                    metadata: Optional[Dict[str, Any]]) -> Any:
         """Send through this transport while preserving the logical platform."""
-        if self.is_relay:
-            return await self.adapter.send_for_platform(logical_platform, chat_id, content, metadata=metadata)
-        return await self.adapter.send(chat_id, content, metadata=metadata)
+        return await (self.adapter.send_for_platform(logical_platform, chat_id, content, metadata=metadata)
+                      if self.is_relay else self.adapter.send(chat_id, content, metadata=metadata))
 
 
 def resolve_delivery_transport(platform: Platform, config: GatewayConfig,
@@ -63,15 +62,12 @@ def resolve_delivery_transport(platform: Platform, config: GatewayConfig,
     logical platform, so restart-time delivery is independent of per-chat caches without letting Relay
     hijack unrelated platform targets."""
     live_adapters = adapters or {}
-    native = live_adapters.get(platform)
-    native_config = config.platforms.get(platform)
+    native, native_config = live_adapters.get(platform), config.platforms.get(platform)
     # Explicitly supplied live adapters with no config block are honored, but an
     # explicitly disabled native adapter never shadows an enabled Relay transport.
     if native is not None and (native_config is None or native_config.enabled):
         return DeliveryTransport(native, native_config, platform)
-
-    relay = live_adapters.get(Platform.RELAY)
-    relay_config = config.platforms.get(Platform.RELAY)
+    relay, relay_config = live_adapters.get(Platform.RELAY), config.platforms.get(Platform.RELAY)
     fronts_platform = getattr(relay, "fronts_platform", None)
     if (relay is not None and (relay_config is None or relay_config.enabled)
             and callable(fronts_platform) and fronts_platform(platform)):
@@ -99,9 +95,7 @@ def _looks_like_int(value: Optional[str]) -> bool:
 def _send_result_error(result: Any) -> Optional[str]:
     """Error string of a failed SendResult object / plain result dict ("" if none), or None on success."""
     get = result.get if isinstance(result, dict) else (lambda name, default=None: getattr(result, name, default))
-    if get("success", True) is not False:
-        return None
-    return str(get("error") or "")
+    return None if get("success", True) is not False else str(get("error") or "")
 
 
 @dataclass
@@ -116,23 +110,18 @@ class DeliveryTarget:
     @classmethod
     def parse(cls, target: str, origin: Optional[SessionSource] = None) -> "DeliveryTarget":
         """Parse "origin" | "local" | "<platform>" | "<platform>:<chat_id>[:<thread_id>]"."""
-        target_stripped, target_lower = target.strip(), target.strip().lower()
-        if target_lower == "origin" and origin:
-            return cls(platform=origin.platform, chat_id=origin.chat_id, thread_id=origin.thread_id, is_origin=True)
-        if target_lower == "origin":
-            return cls(platform=Platform.LOCAL, is_origin=True)
-        if target_lower == "local":
-            return cls(platform=Platform.LOCAL)
+        target = target.strip()
+        if target.lower() == "origin":
+            return (cls(platform=origin.platform, chat_id=origin.chat_id, thread_id=origin.thread_id, is_origin=True)
+                    if origin else cls(platform=Platform.LOCAL, is_origin=True))
         # Platform names are case-insensitive; chat/thread ids keep case. Unknown platforms -> local.
-        parts = target_stripped.split(":", 2)
+        parts = target.split(":", 2)
         try:
             platform = Platform(parts[0].lower())
         except ValueError:
             return cls(platform=Platform.LOCAL)
-        if len(parts) == 1:
-            return cls(platform=platform)
-        return cls(platform=platform, chat_id=parts[1],
-                   thread_id=parts[2] if len(parts) > 2 else None, is_explicit=True)
+        return (cls(platform=platform, chat_id=parts[1], thread_id=parts[2] if len(parts) > 2 else None, is_explicit=True)
+                if len(parts) > 1 else cls(platform=platform))
 
     def to_string(self) -> str:
         """Convert back to string format."""
@@ -221,9 +210,8 @@ class DeliveryRouter:
     def _filter_silence_narration_enabled(self) -> bool:
         """``HERMES_FILTER_SILENCE_NARRATION`` env overrides the ``gateway.filter_silence_narration`` flag."""
         env = os.getenv("HERMES_FILTER_SILENCE_NARRATION")
-        if env is None:
-            return bool(getattr(self.config, "filter_silence_narration", True))
-        return env.strip().lower() in ("1", "true", "yes", "on")
+        return (bool(getattr(self.config, "filter_silence_narration", True)) if env is None
+                else env.strip().lower() in ("1", "true", "yes", "on"))
 
     def _cap_oversized_output(self, adapter: Any, content: str, job_id: str) -> str:
         """Audit-save oversized cron output; truncate it for non-chunking adapters. Above MAX_PLATFORM_OUTPUT
@@ -245,8 +233,7 @@ class DeliveryRouter:
             return content
         # The footer needs a valid path: if the best-effort save failed, retry
         # (a failure now is a real delivery problem and propagates).
-        if saved_path is None:
-            saved_path = self._save_full_output(content, job_id)
+        saved_path = saved_path or self._save_full_output(content, job_id)
         footer = f"\n\n... [truncated, full output saved to {saved_path}]"
         logger.info("Cron output truncated (%d chars) — full output: %s", len(content), saved_path)
         return content[:max(0, MAX_PLATFORM_OUTPUT - len(footer))] + footer
@@ -276,41 +263,39 @@ class DeliveryRouter:
             return {"success": True, "filtered": "silence_narration", "delivered": False}
 
         send_metadata = dict(metadata or {})
-        if transport.is_relay:
-            home = self.config.get_home_channel(target.platform)
-            if home is not None and home.chat_id == target.chat_id:
-                send_metadata.update({k: v for k, v in (("user_id", home.user_id), ("scope_id", home.scope_id)) if v})
+        home = self.config.get_home_channel(target.platform) if transport.is_relay else None
+        if home is not None and home.chat_id == target.chat_id:
+            send_metadata.update({k: v for k, v in (("user_id", home.user_id), ("scope_id", home.scope_id)) if v})
 
         # Caller-supplied thread routing always wins over target.thread_id.
-        named_telegram_private_topic_name: Optional[str] = None
+        named_topic: Optional[str] = None  # named Telegram private topic created for this send
         thread_id = target.thread_id
         if thread_id and not any(key in send_metadata for key in _THREAD_ROUTING_KEYS):
-            is_telegram_private = target.platform == Platform.TELEGRAM and looks_like_telegram_private_chat_id(target.chat_id)
             send_metadata["thread_id"] = thread_id
-            if is_telegram_private and not _looks_like_int(thread_id):
-                # Named topic: create via createForumTopic, use message_thread_id directly.
-                named_telegram_private_topic_name = thread_id
-                send_metadata["thread_id"] = await _ensure_named_dm_topic(adapter, target.chat_id, thread_id, refresh=False)
-                send_metadata["telegram_dm_topic_created_for_send"] = True
-            elif is_telegram_private:
-                # Legacy numeric private topic ids not created by this send path need a reply
-                # anchor to stay visible in the requested lane.
-                if send_metadata.get("telegram_reply_to_message_id") is None:
+            if target.platform == Platform.TELEGRAM and looks_like_telegram_private_chat_id(target.chat_id):
+                if not _looks_like_int(thread_id):
+                    # Named topic: create via createForumTopic, use message_thread_id directly.
+                    named_topic = thread_id
+                    send_metadata["thread_id"] = await _ensure_named_dm_topic(adapter, target.chat_id, thread_id, refresh=False)
+                    send_metadata["telegram_dm_topic_created_for_send"] = True
+                elif send_metadata.get("telegram_reply_to_message_id") is None:
+                    # Legacy numeric private topic ids not created by this send path need a reply
+                    # anchor to stay visible in the requested lane.
                     raise RuntimeError(
                         "Telegram private DM topic delivery requires telegram_reply_to_message_id; "
                         "send to the bare chat or provide a reply anchor"
                     )
-                send_metadata["telegram_dm_topic_reply_fallback"] = True
+                else:
+                    send_metadata["telegram_dm_topic_reply_fallback"] = True
 
-        result = await transport.send(target.platform, target.chat_id, content, metadata=send_metadata or None)
-        error = _send_result_error(result)
-        if error is not None and named_telegram_private_topic_name and "thread not found" in error.lower():
-            send_metadata["thread_id"] = await _ensure_named_dm_topic(
-                adapter, target.chat_id, named_telegram_private_topic_name, refresh=True
-            )
-            send_metadata["telegram_dm_topic_created_for_send"] = True
+        for retry in (False, True):
             result = await transport.send(target.platform, target.chat_id, content, metadata=send_metadata or None)
             error = _send_result_error(result)
+            if retry or error is None or not named_topic or "thread not found" not in error.lower():
+                break
+            # The named topic vanished under us: recreate it once and resend.
+            send_metadata["thread_id"] = await _ensure_named_dm_topic(adapter, target.chat_id, named_topic, refresh=True)
+            send_metadata["telegram_dm_topic_created_for_send"] = True
         if error is not None:
             raise RuntimeError(error or f"{target.platform.value} delivery failed")
         return result

gateway/delivery_ledger.py

@@ -18,7 +18,7 @@ import os
 import sqlite3
 import threading
 import time
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 from typing import Any, Dict, Iterator, List, Optional
 
 from hermes_constants import get_hermes_home
@@ -82,8 +82,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
             adapter_profile TEXT
         )"""
     )
-    columns = {row[1] for row in conn.execute("PRAGMA table_info(delivery_obligations)")}
-    if "adapter_profile" not in columns:
+    if "adapter_profile" not in {row[1] for row in conn.execute("PRAGMA table_info(delivery_obligations)")}:
         try:
             conn.execute("ALTER TABLE delivery_obligations ADD COLUMN adapter_profile TEXT")
         except sqlite3.OperationalError as exc:
@@ -98,20 +97,21 @@ def _transaction() -> Iterator[sqlite3.Connection]:
     context manager only commits/rolls back, so ``with _connect()`` alone leaks a connection (and its
     WAL/SHM fds) per call — ``record_obligation`` runs on every final response; exhausts RLIMIT_NOFILE."""
     conn = _connect()
+    with closing(conn), conn:
+        yield conn
+
+
+def _start_time(pid: int) -> Optional[int]:
     try:
-        with conn:
-            yield conn
-    finally:
-        conn.close()
+        from gateway.status import get_process_start_time  # lazy: tests monkeypatch gateway.status
+        return get_process_start_time(pid)
+    except Exception:
+        return None
 
 
 def _owner_stamp() -> tuple[int, Optional[int]]:
     pid = os.getpid()
-    try:
-        from gateway.status import get_process_start_time
-        return pid, get_process_start_time(pid)
-    except Exception:
-        return pid, None
+    return pid, _start_time(pid)
 
 
 def _owner_alive(pid: Any, started_at: Any) -> bool:
@@ -122,16 +122,11 @@ def _owner_alive(pid: Any, started_at: Any) -> bool:
         pid = int(pid)
     except (TypeError, ValueError):
         return False
-    try:
-        from gateway.status import get_process_start_time
-        current_start = get_process_start_time(pid)
-    except Exception:
-        current_start = None
+    current_start = _start_time(pid)
     if current_start is None:
-        # Start time unreadable: alive iff the pid exists. Route through the cross-platform probe —
-        # ``os.kill(pid, 0)`` on Windows is NOT a no-op (bpo-14484: it maps to
-        # ``GenerateConsoleCtrlEvent(0, pid)`` and could Ctrl+C the gateway's own console group).
-        # ``_pid_exists`` keeps EPERM-means-alive semantics (pid exists, owned by another user).
+        # Start time unreadable: alive iff the pid exists. Route through the cross-platform probe — on Windows
+        # ``os.kill(pid, 0)`` is NOT a no-op (bpo-14484: it maps to ``GenerateConsoleCtrlEvent(0, pid)`` and could
+        # Ctrl+C the gateway's own console group). ``_pid_exists`` keeps EPERM-means-alive (pid owned by another user).
         try:
             from gateway.status import _pid_exists
         except Exception:
@@ -139,11 +134,9 @@ def _owner_alive(pid: Any, started_at: Any) -> bool:
                 return False  # never fall back to a raw sig-0 probe on Windows
             try:
                 os.kill(pid, 0)  # windows-footgun: ok — POSIX-only fallback branch
-            except PermissionError:
                 return True
-            except OSError:  # incl. ProcessLookupError
-                return False
-            return True
+            except OSError as exc:  # incl. ProcessLookupError; EPERM means the pid exists
+                return isinstance(exc, PermissionError)
         try:
             return bool(_pid_exists(pid))
         except Exception:
@@ -163,8 +156,7 @@ def compute_obligation_id(session_key: str, message_ref: str, content: str) -> s
 def record_obligation(*, obligation_id: str, session_key: str, platform: str, chat_id: str,
                       thread_id: Optional[str], content: str, adapter_profile: Optional[str] = None) -> None:
     """Record a final response as owed to the platform (state='pending')."""
-    now = time.time()
-    pid, started = _owner_stamp()
+    now, (pid, started) = time.time(), _owner_stamp()
     with _DB_LOCK, _transaction() as conn:
         conn.execute(
             """INSERT OR REPLACE INTO delivery_obligations
@@ -173,8 +165,7 @@ def record_obligation(*, obligation_id: str, session_key: str, platform: str, ch
                 owner_pid, owner_started_at, adapter_profile)
                VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?, ?)""",
             (obligation_id, session_key, platform, str(chat_id), str(thread_id) if thread_id else None,
-             content, now, now, pid, started, str(adapter_profile).strip() if adapter_profile else "default"),
-        )
+             content, now, now, pid, started, str(adapter_profile).strip() if adapter_profile else "default"))
     _prune()
 
 
@@ -207,8 +198,7 @@ def release_runtime_claim(obligation_id: str, error: str = "") -> bool:
                    updated_at=?, last_error=?
                WHERE obligation_id=? AND state='attempting'
                  AND owner_pid IS ? AND owner_started_at IS ?""",
-            (time.time(), error[:500] if error else None, obligation_id, pid, started),
-        )
+            (time.time(), error[:500] if error else None, obligation_id, pid, started))
     return bool(cursor.rowcount)
 
 
@@ -218,27 +208,16 @@ def _update_state(obligation_id: str, state: str, error: str = "") -> None:
             """UPDATE delivery_obligations
                SET state=?, updated_at=?, last_error=?
                WHERE obligation_id=?""",
-            (state, time.time(), error[:500] if error else None, obligation_id),
-        )
-
-
-def _exhausted(attempts: int, created_at: float, now: float) -> bool:
-    return attempts >= MAX_ATTEMPTS or (now - created_at) > STALE_AFTER_SECONDS
+            (state, time.time(), error[:500] if error else None, obligation_id))
 
 
 def _claimed_row(oid, session_key, platform, chat_id, thread_id, content, attempts, profile, *,
                  needs_marker: bool, runtime: bool = False) -> Dict[str, Any]:
     """Claimed-row dict handed back for redelivery; ``runtime`` adds the reconnect-marker fields."""
-    row: Dict[str, Any] = {"obligation_id": oid, "session_key": session_key, "platform": platform,
-                           "chat_id": chat_id, "thread_id": thread_id, "content": content,
-                           "needs_marker": needs_marker}
-    if runtime:
-        row["marker"] = RECONNECTED_MARKER
-    row["profile"] = profile
-    if runtime:
-        row["runtime_recovery"] = True
-    row["attempts"] = attempts + 1
-    return row
+    return {"obligation_id": oid, "session_key": session_key, "platform": platform, "chat_id": chat_id,
+            "thread_id": thread_id, "content": content, "needs_marker": needs_marker,
+            **({"marker": RECONNECTED_MARKER} if runtime else {}), "profile": profile,
+            **({"runtime_recovery": True} if runtime else {}), "attempts": attempts + 1}
 
 
 def sweep_recoverable(now: Optional[float] = None, *, deliverable_platforms: Optional[set] = None,
@@ -253,8 +232,7 @@ def sweep_recoverable(now: Optional[float] = None, *, deliverable_platforms: Opt
     and hits the cap having never been sent once (the stale cutoff still bounds untouched rows).
     ``deliverable_targets`` further scopes multiplexed gateways by exact ``(platform, adapter_profile)``
     so one connected bot cannot spend another disconnected bot's retry budget."""
-    now = now if now is not None else time.time()
-    pid, started = _owner_stamp()
+    now, (pid, started) = now if now is not None else time.time(), _owner_stamp()
     claimed: List[Dict[str, Any]] = []
     with _DB_LOCK, _transaction() as conn:
         rows = conn.execute(
@@ -268,24 +246,20 @@ def sweep_recoverable(now: Optional[float] = None, *, deliverable_platforms: Opt
              owner_pid, owner_started_at, adapter_profile) in rows:
             if _owner_alive(owner_pid, owner_started_at):
                 continue  # a live gateway still owns this row
-            if _exhausted(attempts, created_at, now):
+            if attempts >= MAX_ATTEMPTS or (now - created_at) > STALE_AFTER_SECONDS:  # exhausted -> abandoned
                 conn.execute(
                     """UPDATE delivery_obligations
-                       SET state='abandoned', updated_at=? WHERE obligation_id=?""",
-                    (now, oid),
-                )
+                       SET state='abandoned', updated_at=? WHERE obligation_id=?""", (now, oid))
                 continue
-            if deliverable_platforms is not None and platform not in deliverable_platforms:
+            if ((deliverable_platforms is not None and platform not in deliverable_platforms)
+                    or (deliverable_targets is not None and (platform, adapter_profile) not in deliverable_targets)):
                 continue  # no adapter this boot — claiming would spend an attempt on a no-op
-            if deliverable_targets is not None and (platform, adapter_profile) not in deliverable_targets:
-                continue
             cursor = conn.execute(
                 """UPDATE delivery_obligations
                    SET owner_pid=?, owner_started_at=?, attempts=attempts+1,
                        updated_at=?
                    WHERE obligation_id=? AND (owner_pid IS ? OR owner_pid=?)""",
-                (pid, started, now, oid, owner_pid, owner_pid),
-            )
+                (pid, started, now, oid, owner_pid, owner_pid))
             if cursor.rowcount:  # pending = never started, redeliver plainly; else carry marker
                 claimed.append(_claimed_row(oid, session_key, platform, chat_id, thread_id, content, attempts,
                                             adapter_profile, needs_marker=state != "pending"))
@@ -304,8 +278,7 @@ def sweep_failed_for_runtime(platform: str, now: Optional[float] = None, *,
     allowlisted transient errors, same attempts/staleness bounds, every update guarded by the prior owner
     stamp and ``failed`` state. Claimed rows always carry the reconnect marker (the failed send's ack is
     not safe to infer)."""
-    now = now if now is not None else time.time()
-    pid, started = _owner_stamp()
+    now, (pid, started) = now if now is not None else time.time(), _owner_stamp()
     if started is None:  # PID alone cannot distinguish this process from a stale row left after PID
         return []        # reuse; runtime replay is optional, so fail closed (startup recovery remains).
     expected_profile = "default" if not profile or profile == "default" else str(profile)
@@ -316,32 +289,26 @@ def sweep_failed_for_runtime(platform: str, now: Optional[float] = None, *,
                       content, attempts, created_at, owner_pid,
                       owner_started_at, last_error, adapter_profile
                FROM delivery_obligations
-               WHERE state='failed' AND platform=?""",
-            (platform,),
-        ).fetchall()
+               WHERE state='failed' AND platform=?""", (platform,)).fetchall()
         for (oid, session_key, row_platform, chat_id, thread_id, content, attempts, created_at,
              owner_pid, owner_started_at, last_error, adapter_profile) in rows:
             # Exact process-start matching prevents PID reuse from stealing work.
             if (adapter_profile != expected_profile or owner_pid != pid or owner_started_at != started
                     or str(last_error or "").strip().lower() not in _RUNTIME_RETRYABLE_ERRORS):
                 continue
-            owner_guard = (oid, owner_pid, owner_started_at)
-            if _exhausted(attempts, created_at, now):
+            owner_guard = (now, oid, owner_pid, owner_started_at)
+            if attempts >= MAX_ATTEMPTS or (now - created_at) > STALE_AFTER_SECONDS:  # exhausted -> abandoned
                 conn.execute(
                     """UPDATE delivery_obligations
                        SET state='abandoned', updated_at=?
                        WHERE obligation_id=? AND state='failed'
-                         AND owner_pid IS ? AND owner_started_at IS ?""",
-                    (now, *owner_guard),
-                )
+                         AND owner_pid IS ? AND owner_started_at IS ?""", owner_guard)
                 continue
             cursor = conn.execute(
                 """UPDATE delivery_obligations
                    SET state='attempting', attempts=attempts+1, updated_at=?
                    WHERE obligation_id=? AND state='failed'
-                     AND owner_pid IS ? AND owner_started_at IS ?""",
-                (now, *owner_guard),
-            )
+                     AND owner_pid IS ? AND owner_started_at IS ?""", owner_guard)
             if cursor.rowcount:
                 claimed.append(_claimed_row(oid, session_key, row_platform, chat_id, thread_id, content,
                                             attempts, adapter_profile, needs_marker=True, runtime=True))
@@ -354,12 +321,9 @@ def _prune(now: Optional[float] = None) -> None:
         with _transaction() as conn:
             conn.execute(
                 """DELETE FROM delivery_obligations
-                   WHERE state IN ('delivered', 'abandoned') AND updated_at < ?""",
-                (now - _RETENTION_SECONDS,),
-            )
+                   WHERE state IN ('delivered', 'abandoned') AND updated_at < ?""", (now - _RETENTION_SECONDS,))
             total = conn.execute("SELECT COUNT(*) FROM delivery_obligations").fetchone()[0]
-            excess = max(0, total - _MAX_ROWS)
-            if excess:
+            if total > _MAX_ROWS:
                 conn.execute(
                     """DELETE FROM delivery_obligations WHERE obligation_id IN (
                          SELECT obligation_id FROM delivery_obligations
@@ -368,9 +332,7 @@ def _prune(now: Optional[float] = None) -> None:
                                     WHEN 'abandoned' THEN 1
                                     ELSE 2
                                   END, updated_at ASC
-                         LIMIT ?)""",
-                    (excess,),
-                )
+                         LIMIT ?)""", (total - _MAX_ROWS,))
     except Exception:
         logger.debug("delivery ledger prune failed", exc_info=True)
 
@@ -382,8 +344,6 @@ def ledger_enabled(config: Optional[Dict[str, Any]] = None) -> bool:
             from hermes_cli.config import load_config
             config = load_config()
         value = (config.get("gateway") or {}).get("delivery_ledger", True)
-        if isinstance(value, str):
-            return value.strip().lower() not in {"false", "0", "no", "off"}
-        return bool(value)
+        return value.strip().lower() not in {"false", "0", "no", "off"} if isinstance(value, str) else bool(value)
     except Exception:
         return True

gateway/rich_sent_store.py

@@ -40,8 +40,7 @@ def record(chat_id, message_id, text: Optional[str]) -> None:
         data = _load(path)
         data[f"{chat_id}:{message_id}"] = {"t": text[:_MAX_TEXT_CHARS], "ts": int(time.time())}
         if len(data) > _MAX_ENTRIES:  # trim oldest by timestamp
-            by_age = sorted(data.items(), key=lambda kv: kv[1].get("ts", 0))
-            for k, _ in by_age[: len(data) - _MAX_ENTRIES]:
+            for k, _ in sorted(data.items(), key=lambda kv: kv[1].get("ts", 0))[: len(data) - _MAX_ENTRIES]:
                 data.pop(k, None)
         tmp = f"{path}.tmp.{os.getpid()}"
         with open(tmp, "w", encoding="utf-8") as fh: