From 356aea3560700c3f630b88b33d4fd46fb6604deb Mon Sep 17 00:00:00 2001 From: Ivan Date: Mon, 27 Apr 2026 11:15:08 -0500 Subject: [PATCH] feat(announce_manager): add aspect store configuration and announce storage logic --- meshchatx/src/backend/announce_manager.py | 23 +++ .../src/backend/auto_propagation_manager.py | 17 +- meshchatx/src/backend/community_interfaces.py | 50 +++++- .../backend/community_interfaces_directory.py | 28 ++++ meshchatx/src/backend/config_manager.py | 69 +++++++- meshchatx/src/backend/database/messages.py | 62 +++++++- meshchatx/src/backend/database/schema.py | 14 +- meshchatx/src/backend/identity_context.py | 18 ++- .../src/backend/local_message_retention.py | 70 ++++++++ meshchatx/src/backend/lxmf_utils.py | 6 +- meshchatx/src/backend/markdown_renderer.py | 2 +- .../src/backend/reticulum_pathfinding.py | 150 ++++++++++++++++++ meshchatx/src/backend/translator_handler.py | 71 ++++++--- 13 files changed, 542 insertions(+), 38 deletions(-) create mode 100644 meshchatx/src/backend/local_message_retention.py create mode 100644 meshchatx/src/backend/reticulum_pathfinding.py diff --git a/meshchatx/src/backend/announce_manager.py b/meshchatx/src/backend/announce_manager.py index 5bbe252..ff12522 100644 --- a/meshchatx/src/backend/announce_manager.py +++ b/meshchatx/src/backend/announce_manager.py @@ -18,6 +18,14 @@ _ASPECT_FETCH_LIMIT_KEYS = { "lxst.telephony": "announce_fetch_limit_lxmf_delivery", } +_ASPECT_STORE_ENABLE_KEYS = { + "lxmf.delivery": "announce_store_lxmf_delivery", + "lxst.telephony": "announce_store_lxst_telephony", + "nomadnetwork.node": "announce_store_nomadnetwork_node", + "lxmf.propagation": "announce_store_lxmf_propagation", + "git.repositories": "announce_store_git_repositories", +} + class AnnounceManager: def __init__(self, db: Database, config=None): @@ -50,6 +58,17 @@ class AnnounceManager: return 500 return min(v, 100_000) + def is_storing_announce_for_aspect(self, aspect, force_store: bool = False) -> bool: + if force_store or not self.config: + return True + key = _ASPECT_STORE_ENABLE_KEYS.get(aspect) + if not key: + return True + attr = getattr(self.config, key, None) + if attr is None: + return True + return bool(attr.get()) + def upsert_announce( self, reticulum, @@ -58,7 +77,11 @@ class AnnounceManager: aspect, app_data, announce_packet_hash, + force_store: bool = False, ): + if not self.is_storing_announce_for_aspect(aspect, force_store=force_store): + return + rssi = snr = quality = None if announce_packet_hash and reticulum: rssi = reticulum.get_packet_rssi(announce_packet_hash) diff --git a/meshchatx/src/backend/auto_propagation_manager.py b/meshchatx/src/backend/auto_propagation_manager.py index d9b1c9a..1da9b4c 100644 --- a/meshchatx/src/backend/auto_propagation_manager.py +++ b/meshchatx/src/backend/auto_propagation_manager.py @@ -8,6 +8,7 @@ from LXMF.LXMRouter import LXMRouter from meshchatx.src.backend.async_utils import AsyncUtils from meshchatx.src.backend.meshchat_utils import parse_lxmf_propagation_node_app_data +from meshchatx.src.backend import reticulum_pathfinding _PROP_FAILURE_STATES = frozenset( { @@ -57,15 +58,13 @@ class AutoPropagationManager: await asyncio.sleep(self._check_interval) async def _wait_for_path(self, dest_hash: bytes, timeout: float) -> bool: - if RNS.Transport.has_path(dest_hash): - return True - RNS.Transport.request_path(dest_hash) - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if RNS.Transport.has_path(dest_hash): - return True - await asyncio.sleep(POLL_INTERVAL_SECONDS) - return RNS.Transport.has_path(dest_hash) + r = self.app.reticulum if self.app and hasattr(self.app, "reticulum") else None + return await reticulum_pathfinding.wait_for_path( + r, + dest_hash, + timeout, + poll_interval=POLL_INTERVAL_SECONDS, + ) async def _probe_propagation_sync(self, node_hex: str) -> bool: ctx = self.context diff --git a/meshchatx/src/backend/community_interfaces.py b/meshchatx/src/backend/community_interfaces.py index 5069268..392668f 100644 --- a/meshchatx/src/backend/community_interfaces.py +++ b/meshchatx/src/backend/community_interfaces.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: 0BSD import json +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -8,16 +9,23 @@ _BUNDLED = Path(__file__).resolve().parent / "data" / "community_interfaces.json class CommunityInterfacesManager: - """Load suggested interface presets from bundled data or public/community_interfaces.json.""" + """Load suggested interface presets from cache, public override, or bundled data.""" - def __init__(self, public_override_path: str | None = None): + def __init__( + self, + public_override_path: str | None = None, + cache_path: str | Path | None = None, + ): self._public_override_path = public_override_path + self._cache_path = Path(cache_path) if cache_path else None self.interfaces = self._load_raw() def _candidate_paths(self) -> list[Path]: paths: list[Path] = [] if self._public_override_path: paths.append(Path(self._public_override_path)) + if self._cache_path: + paths.append(self._cache_path) paths.append(_BUNDLED) return paths @@ -67,5 +75,43 @@ class CommunityInterfacesManager: out["target_port"] = int(tp) return out + def refresh_from_directory( + self, url: str | None = None, timeout: float = 60.0 + ) -> dict[str, Any]: + from meshchatx.src.backend.community_interfaces_directory import ( + build_interfaces_from_directory_url, + ) + + interfaces, resolved_url = build_interfaces_from_directory_url( + url, timeout=timeout + ) + if not interfaces: + raise ValueError("Directory returned no usable interface presets") + if self._cache_path: + self._cache_path.parent.mkdir(parents=True, exist_ok=True) + doc = { + "_comment": "MeshChatX cache from in-app directory refresh. Load order is public " + "community_interfaces.json, then this cache, then bundled presets.", + "_source": resolved_url, + "_refreshed_at": datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat(), + "interfaces": interfaces, + } + self._cache_path.write_text( + json.dumps(doc, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + self.interfaces = self._load_raw() + else: + self.interfaces = [ + self._normalize_entry(dict(x)) + for x in interfaces + if isinstance(x, dict) + ] + if not self.interfaces: + raise ValueError("No interface presets after refresh") + return {"count": len(self.interfaces), "source": resolved_url} + async def get_interfaces(self) -> list[dict[str, Any]]: return [{**iface, "online": None, "last_check": 0} for iface in self.interfaces] diff --git a/meshchatx/src/backend/community_interfaces_directory.py b/meshchatx/src/backend/community_interfaces_directory.py index 2ed4fcc..f2cf940 100644 --- a/meshchatx/src/backend/community_interfaces_directory.py +++ b/meshchatx/src/backend/community_interfaces_directory.py @@ -4,7 +4,10 @@ from __future__ import annotations +import json import re +import urllib.error +import urllib.request from typing import Any DEFAULT_SUBMITTED_URL = ( @@ -13,6 +16,31 @@ DEFAULT_SUBMITTED_URL = ( DESCRIPTION = "directory.rns.recipes (user-submitted, online)" + +def fetch_directory_payload(url: str, *, timeout: float = 60.0) -> object: + req = urllib.request.Request( + url, + headers={ + "Accept": "application/json", + "User-Agent": "MeshChatX-community-interfaces/1.0 (+https://meshchatx.com/)", + }, + method="GET", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + +def build_interfaces_from_directory_url( + url: str | None = None, + *, + timeout: float = 60.0, +) -> tuple[list[dict[str, Any]], str]: + resolved = url or DEFAULT_SUBMITTED_URL + payload = fetch_directory_payload(resolved, timeout=timeout) + rows = rows_from_payload(payload) + return transform_directory_rows(rows), resolved + + _RE_REMOTE = re.compile(r"^\s*remote\s*=\s*(\S+)", re.MULTILINE | re.IGNORECASE) _RE_TARGET_HOST = re.compile( r"^\s*target_host\s*=\s*(\S+)", diff --git a/meshchatx/src/backend/config_manager.py b/meshchatx/src/backend/config_manager.py index 28d293d..2b5b97f 100644 --- a/meshchatx/src/backend/config_manager.py +++ b/meshchatx/src/backend/config_manager.py @@ -256,7 +256,16 @@ class ConfigManager: self.telemetry_enabled = self.BoolConfig(self, "telemetry_enabled", False) # translator config - self.translator_enabled = self.BoolConfig(self, "translator_enabled", False) + self.translator_argos_enabled = self.BoolConfig( + self, + "translator_argos_enabled", + False, + ) + self.translator_libretranslate_enabled = self.BoolConfig( + self, + "translator_libretranslate_enabled", + False, + ) self.libretranslate_url = self.StringConfig( self, "libretranslate_url", @@ -337,6 +346,33 @@ class ConfigManager: "#e5e7eb", ) + # When False, meshchat does not persist received announces of that type (all True by default). + self.announce_store_lxmf_delivery = self.BoolConfig( + self, + "announce_store_lxmf_delivery", + True, + ) + self.announce_store_lxst_telephony = self.BoolConfig( + self, + "announce_store_lxst_telephony", + True, + ) + self.announce_store_nomadnetwork_node = self.BoolConfig( + self, + "announce_store_nomadnetwork_node", + True, + ) + self.announce_store_lxmf_propagation = self.BoolConfig( + self, + "announce_store_lxmf_propagation", + True, + ) + self.announce_store_git_repositories = self.BoolConfig( + self, + "announce_store_git_repositories", + True, + ) + # announce caps: max rows stored per aspect (oldest dropped). Default 1000. self.announce_max_stored_lxmf_delivery = self.IntConfig( self, @@ -425,7 +461,29 @@ class ConfigManager: "[]", ) + self.local_message_auto_delete_enabled = self.BoolConfig( + self, + "local_message_auto_delete_enabled", + False, + ) + self.local_message_auto_delete_value = self.IntConfig( + self, + "local_message_auto_delete_value", + 30, + ) + self.local_message_auto_delete_unit = self.StringConfig( + self, + "local_message_auto_delete_unit", + "days", + ) + self.local_message_auto_delete_last_run_at = self.IntConfig( + self, + "local_message_auto_delete_last_run_at", + None, + ) + self._migrate_legacy_announce_limit_keys() + self._migrate_translator_from_legacy() def get(self, key: str, default_value=None) -> str | None: return self.db.config.get(key, default_value) @@ -433,6 +491,15 @@ class ConfigManager: def set(self, key: str, value: str | None): self.db.config.set(key, value) + def _migrate_translator_from_legacy(self): + old = self.db.config.get("translator_enabled", default=None) + a = self.db.config.get("translator_argos_enabled", default=None) + libre = self.db.config.get("translator_libretranslate_enabled", default=None) + if old is not None and a is None and libre is None: + v = "true" if str(old).lower() == "true" else "false" + self.db.config.set("translator_argos_enabled", v) + self.db.config.set("translator_libretranslate_enabled", v) + def _migrate_legacy_announce_limit_keys(self): pairs = [ ("announce_limit_lxmf_delivery", "announce_max_stored_lxmf_delivery"), diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py index b7e41a3..ad0cb18 100644 --- a/meshchatx/src/backend/database/messages.py +++ b/meshchatx/src/backend/database/messages.py @@ -37,11 +37,27 @@ class MessageDAO: "is_spam", "reply_to_hash", "attachments_stripped", + "path_hops_at_send", + "path_interface_at_send", + "path_finding_measure", + "path_row_hash_hex", ] columns = ", ".join(fields) placeholders = ", ".join(["?"] * len(fields)) - update_set = ", ".join([f"{f} = EXCLUDED.{f}" for f in fields if f != "hash"]) + update_fields = [ + f + for f in fields + if f != "hash" + and f + not in ( + "path_hops_at_send", + "path_interface_at_send", + "path_finding_measure", + "path_row_hash_hex", + ) + ] + update_set = ", ".join([f"{f} = EXCLUDED.{f}" for f in update_fields]) query = ( f"INSERT INTO lxmf_messages ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) " @@ -61,6 +77,17 @@ class MessageDAO: self.provider.execute(query, params) + def set_lxmf_message_path_at_send_if_unset( + self, message_hash, hops, interface_name + ): + """Store Reticulum path snapshot once (send or receive); never overwrites.""" + now = datetime.now(UTC).isoformat() + self.provider.execute( + "UPDATE lxmf_messages SET path_hops_at_send = ?, path_interface_at_send = ?, updated_at = ? " + "WHERE hash = ? AND path_hops_at_send IS NULL", + (hops, interface_name, now, message_hash), + ) + def update_lxmf_message_state( self, message_hash, @@ -166,6 +193,39 @@ class MessageDAO: self.set_peer_pinned(peer_hash, True) return True + def list_message_hashes_with_timestamp_before(self, cutoff_ts: float) -> list[str]: + rows = self.provider.fetchall( + "SELECT hash FROM lxmf_messages WHERE timestamp IS NOT NULL AND timestamp < ?", + (cutoff_ts,), + ) + return [r["hash"] for r in rows if r.get("hash")] + + def prune_conversation_metadata_for_peers_with_no_messages(self) -> None: + self.provider.execute( + """ + DELETE FROM lxmf_conversation_read_state + WHERE destination_hash NOT IN ( + SELECT DISTINCT peer_hash FROM lxmf_messages WHERE peer_hash IS NOT NULL + ) + """, + ) + self.provider.execute( + """ + DELETE FROM lxmf_conversation_folders + WHERE peer_hash NOT IN ( + SELECT DISTINCT peer_hash FROM lxmf_messages WHERE peer_hash IS NOT NULL + ) + """, + ) + self.provider.execute( + """ + DELETE FROM lxmf_conversation_pins + WHERE peer_hash NOT IN ( + SELECT DISTINCT peer_hash FROM lxmf_messages WHERE peer_hash IS NOT NULL + ) + """, + ) + def delete_lxmf_messages_by_hashes(self, message_hashes): if not message_hashes: return diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py index dd97a15..296e951 100644 --- a/meshchatx/src/backend/database/schema.py +++ b/meshchatx/src/backend/database/schema.py @@ -15,7 +15,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str: class DatabaseSchema: - LATEST_VERSION = 46 + LATEST_VERSION = 48 def __init__(self, provider: DatabaseProvider): self.provider = provider @@ -220,6 +220,10 @@ class DatabaseSchema: quality REAL, is_spam INTEGER DEFAULT 0, reply_to_hash TEXT, + path_hops_at_send INTEGER, + path_interface_at_send TEXT, + path_finding_measure TEXT, + path_row_hash_hex TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) @@ -1250,6 +1254,14 @@ class DatabaseSchema: "CREATE INDEX IF NOT EXISTS idx_user_stickers_pack ON user_stickers(pack_id, sort_order)", ) + if current_version < 47: + self._ensure_column("lxmf_messages", "path_hops_at_send", "INTEGER") + self._ensure_column("lxmf_messages", "path_interface_at_send", "TEXT") + + if current_version < 48: + self._ensure_column("lxmf_messages", "path_finding_measure", "TEXT") + self._ensure_column("lxmf_messages", "path_row_hash_hex", "TEXT") + # Update version in config self._safe_execute( """ diff --git a/meshchatx/src/backend/identity_context.py b/meshchatx/src/backend/identity_context.py index 7cb2ddc..6fd519b 100644 --- a/meshchatx/src/backend/identity_context.py +++ b/meshchatx/src/backend/identity_context.py @@ -273,10 +273,10 @@ class IdentityContext: ) libretranslate_url = self.config.libretranslate_url.get() - translator_enabled = self.config.translator_enabled.get() self.translator_handler = TranslatorHandler( libretranslate_url=libretranslate_url, - enabled=translator_enabled, + translator_argos_enabled=self.config.translator_argos_enabled.get(), + translator_libretranslate_enabled=self.config.translator_libretranslate_enabled.get(), ) self.bot_handler = BotHandler( @@ -339,6 +339,10 @@ class IdentityContext: self.community_interfaces_manager = CommunityInterfacesManager( public_override_path=self.app.get_public_path("community_interfaces.json"), + cache_path=os.path.join( + self.storage_path, + "community_interfaces_cache.json", + ), ) self.auto_propagation_manager = AutoPropagationManager( @@ -402,6 +406,16 @@ class IdentityContext: thread.daemon = True thread.start() + # start background thread for local (device-only) message age retention + thread = threading.Thread( + target=asyncio.run, + args=( + self.app.local_message_retention_loop(self.session_id, context=self), + ), + ) + thread.daemon = True + thread.start() + # start background thread for auto propagation node selection thread = threading.Thread( target=asyncio.run, diff --git a/meshchatx/src/backend/local_message_retention.py b/meshchatx/src/backend/local_message_retention.py new file mode 100644 index 0000000..6670fb8 --- /dev/null +++ b/meshchatx/src/backend/local_message_retention.py @@ -0,0 +1,70 @@ +# SPDX-License-Identifier: 0BSD +"""Local-only age-based deletion of stored LXMF rows (this device, no network signal).""" + +import logging +from collections.abc import Callable + +log = logging.getLogger(__name__) + +UNIT_DAYS = "days" +UNIT_MONTHS = "months" + +SECONDS_PER_DAY = 86400 +SECONDS_PER_MONTH = 30 * SECONDS_PER_DAY + +RETENTION_CHECK_INTERVAL_SECONDS = 3600 +LOCAL_RETENTION_STARTUP_GRACE_SECONDS = 120 + +MAX_VALUE_DAYS = 10_000 +MAX_VALUE_MONTHS = 120 + + +def normalize_unit(raw: str | None) -> str: + s = (raw or UNIT_DAYS).strip().lower() + if s in ("month", "months", "mo", "m"): + return UNIT_MONTHS + return UNIT_DAYS + + +def retention_window_seconds(value: int, unit: str) -> int: + try: + v = int(value) + except (TypeError, ValueError): + v = 1 + u = normalize_unit(unit) + if u == UNIT_MONTHS: + return max(1, min(v, MAX_VALUE_MONTHS)) * SECONDS_PER_MONTH + return max(1, min(v, MAX_VALUE_DAYS)) * SECONDS_PER_DAY + + +def local_message_retention_cutoff_ts(now: float, value: int, unit: str) -> float: + return float(now) - float(retention_window_seconds(value, unit)) + + +def apply_local_message_retention( + messages, + cancel_outbound: Callable[[bytes], None] | None, + *, + value: int, + unit: str, + now: float, +) -> int: + """Delete local LXMF message rows older than the retention window. + + Does not contact peers; only removes rows from the local database. + """ + cutoff = local_message_retention_cutoff_ts(now, value, unit) + hashes = messages.list_message_hashes_with_timestamp_before(cutoff) + if not hashes: + return 0 + if cancel_outbound is not None: + for h in hashes: + if not h or len(h) % 2 != 0: + continue + try: + cancel_outbound(bytes.fromhex(h)) + except Exception as exc: # noqa: BLE001 + log.debug("local_message_retention cancel_outbound: %s", exc) + messages.delete_lxmf_messages_by_hashes(hashes) + messages.prune_conversation_metadata_for_peers_with_no_messages() + return len(hashes) diff --git a/meshchatx/src/backend/lxmf_utils.py b/meshchatx/src/backend/lxmf_utils.py index c02fcc1..6c2dac9 100644 --- a/meshchatx/src/backend/lxmf_utils.py +++ b/meshchatx/src/backend/lxmf_utils.py @@ -510,7 +510,7 @@ def convert_db_lxmf_message_to_dict( "is_incoming": bool(db_lxmf_message["is_incoming"]), "state": db_lxmf_message["state"], "progress": db_lxmf_message["progress"], - "method": db_lxmf_message["method"], + "method": db_lxmf_message.get("method") or "unknown", "delivery_attempts": db_lxmf_message["delivery_attempts"], "next_delivery_attempt_at": db_lxmf_message["next_delivery_attempt_at"], "title": db_lxmf_message["title"], @@ -523,6 +523,10 @@ def convert_db_lxmf_message_to_dict( "is_spam": bool(db_lxmf_message["is_spam"]), "reply_to_hash": db_lxmf_message.get("reply_to_hash"), "attachments_stripped": bool(db_lxmf_message.get("attachments_stripped", 0)), + "path_hops_at_send": db_lxmf_message.get("path_hops_at_send"), + "path_interface_at_send": db_lxmf_message.get("path_interface_at_send"), + "path_finding_measure": db_lxmf_message.get("path_finding_measure"), + "path_row_hash_hex": db_lxmf_message.get("path_row_hash_hex"), "created_at": created_at, "updated_at": updated_at, "is_reaction": is_reaction, diff --git a/meshchatx/src/backend/markdown_renderer.py b/meshchatx/src/backend/markdown_renderer.py index ebf5f14..55a48e1 100644 --- a/meshchatx/src/backend/markdown_renderer.py +++ b/meshchatx/src/backend/markdown_renderer.py @@ -101,7 +101,7 @@ class MarkdownRenderer: # Inline code text = re.sub( r"`([^`]+)`", - r'\1', + r'\1', text, ) diff --git a/meshchatx/src/backend/reticulum_pathfinding.py b/meshchatx/src/backend/reticulum_pathfinding.py new file mode 100644 index 0000000..e70564c --- /dev/null +++ b/meshchatx/src/backend/reticulum_pathfinding.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: 0BSD + +import asyncio +import contextlib +import time +from dataclasses import dataclass +from typing import Any, Optional, Protocol + +import RNS + + +@dataclass(frozen=True) +class OutboundPathOutcome: + """Result of waiting for a Reticulum transport path before outbound LXMF.""" + + path_available: bool + prepare_measure: str + used_nudge: bool + + +def format_outbound_path_finding_measure(outcome: OutboundPathOutcome) -> str: + """Single string for storage/API: base measure, plus ``+nudge`` if a nudge was used.""" + base = outcome.prepare_measure + if outcome.used_nudge: + return f"{base}+nudge" + return base + + +IDX_PT_TIMESTAMP = 0 +IDX_PT_RVCD_IF = 5 + + +def _path_table_entry_is_expired_by_reticulum_rules(entry) -> bool: + ts = entry[IDX_PT_TIMESTAMP] + attached = entry[IDX_PT_RVCD_IF] + if attached is not None and hasattr(attached, "mode"): + iface = RNS.Interfaces.Interface.Interface + if attached.mode == iface.MODE_ACCESS_POINT: + dest_exp = ts + RNS.Transport.AP_PATH_TIME + elif attached.mode == iface.MODE_ROAMING: + dest_exp = ts + RNS.Transport.ROAMING_PATH_TIME + else: + dest_exp = ts + RNS.Transport.DESTINATION_TIMEOUT + else: + dest_exp = ts + RNS.Transport.DESTINATION_TIMEOUT + return time.time() > dest_exp + + +def transport_path_table_entry_is_expired(destination_hash: bytes) -> bool: + with RNS.Transport.path_table_lock: + if destination_hash not in RNS.Transport.path_table: + return True + entry = RNS.Transport.path_table[destination_hash] + return _path_table_entry_is_expired_by_reticulum_rules(entry) + + +def should_rediscover_path(destination_hash: bytes) -> bool: + if not RNS.Transport.has_path(destination_hash): + return True + if RNS.Transport.path_is_unresponsive(destination_hash): + return True + return transport_path_table_entry_is_expired(destination_hash) + + +def path_metadata_for_api(destination_hash: bytes) -> dict[str, bool]: + has = RNS.Transport.has_path(destination_hash) + if not has: + return { + "path_stale": True, + "path_unresponsive": False, + } + return { + "path_stale": transport_path_table_entry_is_expired(destination_hash), + "path_unresponsive": RNS.Transport.path_is_unresponsive(destination_hash), + } + + +def prepare_fresh_path_request( + reticulum: Optional["ReticulumLike"], destination_hash: bytes +) -> str: + """Ensure a path request is in flight if needed. + + Returns a stable label for what was done before waiting: + ``reused_valid_path`` (no new request), ``path_refresh_requested`` (dropped + or expired then requested), or ``new_path_requested`` (no prior path). + """ + if not should_rediscover_path(destination_hash): + return "reused_valid_path" + had_path = RNS.Transport.has_path(destination_hash) + if had_path: + if reticulum is not None: + with contextlib.suppress(Exception): + reticulum.drop_path(destination_hash) + else: + RNS.Transport.expire_path(destination_hash) + RNS.Transport.request_path(destination_hash) + return "path_refresh_requested" if had_path else "new_path_requested" + + +def nudge_path_request(destination_hash: bytes) -> None: + RNS.Transport.request_path(destination_hash) + + +def lxmf_path_wait_cap_seconds() -> float: + try: + base = float(RNS.Transport.PATH_REQUEST_TIMEOUT) + except Exception: + base = 30.0 + return max(30.0, min(base, 120.0)) + + +async def await_transport_path_for_outbound_lxmf( + reticulum: Optional["ReticulumLike"], + destination_hash_bytes: bytes, +) -> OutboundPathOutcome: + long_w = lxmf_path_wait_cap_seconds() + short_w = max(15.0, long_w * 0.5) + + measure = prepare_fresh_path_request(reticulum, destination_hash_bytes) + deadline = time.time() + long_w + while not RNS.Transport.has_path(destination_hash_bytes) and time.time() < deadline: + await asyncio.sleep(0.1) + if RNS.Transport.has_path(destination_hash_bytes): + return OutboundPathOutcome(True, measure, False) + + nudge_path_request(destination_hash_bytes) + deadline = time.time() + short_w + while not RNS.Transport.has_path(destination_hash_bytes) and time.time() < deadline: + await asyncio.sleep(0.1) + ok = RNS.Transport.has_path(destination_hash_bytes) + return OutboundPathOutcome(ok, measure, True) + + +class ReticulumLike(Protocol): + def drop_path(self, destination) -> Any: ... + + +async def wait_for_path( + reticulum: Optional["ReticulumLike"], + dest_hash: bytes, + path_wait_timeout_seconds: float, + poll_interval: float = 0.1, +) -> bool: + prepare_fresh_path_request(reticulum, dest_hash) + deadline = time.monotonic() + path_wait_timeout_seconds + while time.monotonic() < deadline: + if RNS.Transport.has_path(dest_hash): + return True + await asyncio.sleep(poll_interval) + return RNS.Transport.has_path(dest_hash) diff --git a/meshchatx/src/backend/translator_handler.py b/meshchatx/src/backend/translator_handler.py index cfe99e4..cb219ef 100644 --- a/meshchatx/src/backend/translator_handler.py +++ b/meshchatx/src/backend/translator_handler.py @@ -91,8 +91,14 @@ def _sync_run_coro(coro): class TranslatorHandler: - def __init__(self, libretranslate_url: str | None = None, enabled: bool = False): - self.enabled = enabled + def __init__( + self, + libretranslate_url: str | None = None, + translator_argos_enabled: bool = False, + translator_libretranslate_enabled: bool = False, + ): + self.translator_argos_enabled = translator_argos_enabled + self.translator_libretranslate_enabled = translator_libretranslate_enabled self.libretranslate_url = libretranslate_url or os.getenv( "LIBRETRANSLATE_URL", "http://localhost:5000", @@ -103,6 +109,18 @@ class TranslatorHandler: self.has_argos = self.has_argos_lib or self.has_argos_cli self.has_requests = HAS_AIOHTTP + def _any_backend_config_enabled(self) -> bool: + return self.translator_argos_enabled or self.translator_libretranslate_enabled + + @property + def enabled(self) -> bool: + return self._any_backend_config_enabled() + + @enabled.setter + def enabled(self, value: bool) -> None: + self.translator_argos_enabled = value + self.translator_libretranslate_enabled = value + async def _fetch_languages_async(self, url: str): base = url.rstrip("/") timeout = aiohttp.ClientTimeout(total=5) @@ -115,23 +133,26 @@ class TranslatorHandler: return await response.json() return None - def get_supported_languages(self, libretranslate_url: str | None = None): - languages = [] - if not self.enabled: - return languages + def get_translator_languages_response( + self, + libretranslate_url: str | None = None, + ) -> dict[str, Any]: + """List installed/reachable language pairs for UI; not gated on enable toggles.""" + languages: list[dict[str, str]] = [] + libretranslate_reachable = False url = libretranslate_url or self.libretranslate_url - if libretranslate_url is not None and str(libretranslate_url).strip(): - try: - url = normalize_loopback_http_service_base(libretranslate_url) - except UnsafeOutboundUrlError as e: - msg = str(e) - raise ValueError(msg) from e - if self.has_requests: + if libretranslate_url is not None and str(libretranslate_url).strip(): + try: + url = normalize_loopback_http_service_base(libretranslate_url) + except UnsafeOutboundUrlError as e: + msg = str(e) + raise ValueError(msg) from e try: libretranslate_langs = _sync_run_coro(self._fetch_languages_async(url)) if libretranslate_langs is not None: + libretranslate_reachable = True languages.extend( { "code": lang.get("code"), @@ -140,7 +161,6 @@ class TranslatorHandler: } for lang in libretranslate_langs ) - return languages except Exception as e: print(f"Failed to fetch LibreTranslate languages: {e}") @@ -169,7 +189,15 @@ class TranslatorHandler: except Exception as e: print(f"Failed to fetch Argos languages via CLI: {e}") - return languages + return { + "languages": languages, + "libretranslate_reachable": libretranslate_reachable, + } + + def get_supported_languages(self, libretranslate_url: str | None = None) -> list: + return self.get_translator_languages_response( + libretranslate_url=libretranslate_url, + )["languages"] def translate_text( self, @@ -179,7 +207,7 @@ class TranslatorHandler: use_argos: bool = False, libretranslate_url: str | None = None, ) -> dict[str, Any]: - if not self.enabled: + if not self._any_backend_config_enabled(): msg = "Translator is disabled" raise RuntimeError(msg) @@ -187,10 +215,13 @@ class TranslatorHandler: msg = "Text cannot be empty" raise ValueError(msg) - if use_argos and self.has_argos: + if use_argos: + if not self.translator_argos_enabled or not self.has_argos: + msg = "Argos translation is not enabled or not available" + raise RuntimeError(msg) return self._translate_argos(text, source_lang, target_lang) - if self.has_requests: + if self.translator_libretranslate_enabled and self.has_requests: try: url = libretranslate_url or self.libretranslate_url if libretranslate_url is not None and str(libretranslate_url).strip(): @@ -206,11 +237,11 @@ class TranslatorHandler: libretranslate_url=url, ) except Exception as e: - if self.has_argos: + if self.translator_argos_enabled and self.has_argos: return self._translate_argos(text, source_lang, target_lang) raise e - if self.has_argos: + if self.translator_argos_enabled and self.has_argos: return self._translate_argos(text, source_lang, target_lang) msg = "No translation backend available. Install aiohttp for LibreTranslate or argostranslate for local translation."