From 6df4bb09cfcb8303beaf40598469ea761afd08ca Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:13:08 -0700 Subject: [PATCH] refactor(gateway): second-pass compaction of delivery ledger/router, control socket, dead targets --- gateway/control_socket.py | 66 +++++++------------- gateway/dead_targets.py | 23 +++---- gateway/delivery.py | 87 +++++++++++--------------- gateway/delivery_ledger.py | 124 +++++++++++++------------------------ gateway/rich_sent_store.py | 3 +- 5 files changed, 110 insertions(+), 193 deletions(-) diff --git a/gateway/control_socket.py b/gateway/control_socket.py index c135febe37..6a767a7a01 100644 --- a/gateway/control_socket.py +++ b/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 diff --git a/gateway/dead_targets.py b/gateway/dead_targets.py index 93805a149c..2b47b3083b 100644 --- a/gateway/dead_targets.py +++ b/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 diff --git a/gateway/delivery.py b/gateway/delivery.py index 7ecc856154..b8478da44f 100644 --- a/gateway/delivery.py +++ b/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" | "" | ":[:]".""" - 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 diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index db65b5bfe1..fac8e2b219 100644 --- a/gateway/delivery_ledger.py +++ b/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 diff --git a/gateway/rich_sent_store.py b/gateway/rich_sent_store.py index 74cdb981a4..645c542db7 100644 --- a/gateway/rich_sent_store.py +++ b/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: