diff --git a/gateway/config.py b/gateway/config.py index ae149f81dc0b2..d0f1b5fa662ae 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -1113,6 +1113,8 @@ def _merge_platform_map(source_platforms: Any) -> None: os.environ["TELEGRAM_IGNORED_THREADS"] = str(ignored_threads) if "reactions" in telegram_cfg and not os.getenv("TELEGRAM_REACTIONS"): os.environ["TELEGRAM_REACTIONS"] = str(telegram_cfg["reactions"]).lower() + if "reaction_feedback" in telegram_cfg and not os.getenv("TELEGRAM_REACTION_FEEDBACK"): + os.environ["TELEGRAM_REACTION_FEEDBACK"] = str(telegram_cfg["reaction_feedback"]).lower() if "proxy_url" in telegram_cfg and not os.getenv("TELEGRAM_PROXY"): os.environ["TELEGRAM_PROXY"] = str(telegram_cfg["proxy_url"]).strip() # reply_to_mode: top-level preferred, falls back to extra.reply_to_mode @@ -1140,7 +1142,7 @@ def _merge_platform_map(source_platforms: Any) -> None: if isinstance(group_allowed_chats, list): group_allowed_chats = ",".join(str(v) for v in group_allowed_chats) os.environ["TELEGRAM_GROUP_ALLOWED_CHATS"] = str(group_allowed_chats) - for _telegram_extra_key in ("guest_mode", "disable_link_previews", "observe_unmentioned_group_messages"): + for _telegram_extra_key in ("guest_mode", "disable_link_previews", "observe_unmentioned_group_messages", "reaction_feedback"): if _telegram_extra_key in telegram_cfg: plat_data = platforms_data.setdefault(Platform.TELEGRAM.value, {}) if not isinstance(plat_data, dict): diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index aed7b71af9b54..cbc9134797c5c 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -35,6 +35,10 @@ ContextTypes, filters, ) + try: + from telegram.ext import MessageReactionHandler + except ImportError: + MessageReactionHandler = None from telegram.constants import ParseMode, ChatType from telegram.request import HTTPXRequest TELEGRAM_AVAILABLE = True @@ -50,6 +54,7 @@ CommandHandler = Any CallbackQueryHandler = Any TelegramMessageHandler = Any + MessageReactionHandler = None HTTPXRequest = Any filters = None ParseMode = None @@ -120,7 +125,7 @@ def check_telegram_requirements() -> bool: global TELEGRAM_AVAILABLE, Update, Bot, Message, InlineKeyboardButton global InlineKeyboardMarkup, LinkPreviewOptions, Application global CommandHandler, CallbackQueryHandler, TelegramMessageHandler - global ContextTypes, filters, ParseMode, ChatType, HTTPXRequest + global MessageReactionHandler, ContextTypes, filters, ParseMode, ChatType, HTTPXRequest if TELEGRAM_AVAILABLE: return True try: @@ -141,6 +146,10 @@ def check_telegram_requirements() -> bool: MessageHandler as _MH, ContextTypes as _CT, filters as _filters, ) + try: + from telegram.ext import MessageReactionHandler as _MRH + except ImportError: + _MRH = None from telegram.constants import ParseMode as _PM, ChatType as _CtT from telegram.request import HTTPXRequest as _HR except ImportError: @@ -155,6 +164,7 @@ def check_telegram_requirements() -> bool: CommandHandler = _CH CallbackQueryHandler = _CQH TelegramMessageHandler = _MH + MessageReactionHandler = _MRH ContextTypes = _CT filters = _filters ParseMode = _PM @@ -914,6 +924,67 @@ def _coerce_bool_extra(self, key: str, default: bool = False) -> bool: return default return bool(value) + def _record_reaction_feedback_target( + self, + chat_id: str, + message_id: str, + content: Optional[str], + metadata: Optional[Dict[str, Any]] = None, + *, + thread_id: Optional[str] = None, + ) -> None: + """Best-effort index of Hermes-sent messages for later reactions.""" + if not self._reaction_feedback_enabled(): + return + try: + from gateway import reaction_feedback + + reaction_feedback.record_sent_message( + platform="telegram", + chat_id=chat_id, + message_id=message_id, + thread_id=thread_id, + content=content, + metadata=metadata, + ) + except Exception: + logger.debug("[%s] Failed to record reaction feedback target", self.name, exc_info=True) + + @staticmethod + def _reaction_emoji_values(reactions: Any) -> List[str]: + values: List[str] = [] + for reaction in reactions or []: + emoji = getattr(reaction, "emoji", None) + if emoji: + values.append(str(emoji)) + return values + + @staticmethod + def _normalize_reaction_chat_type(chat_type: Any, thread_id: Optional[str]) -> str: + value = getattr(chat_type, "value", chat_type) + normalized = str(value or "").strip().lower() + if normalized == "private": + return "dm" + if normalized == "supergroup" and thread_id is not None: + return "forum" + if normalized == "supergroup": + return "group" + return normalized or "group" + + def _reaction_actor_is_bot(self, user: Any) -> bool: + if getattr(user, "is_bot", False): + return True + user_id = getattr(user, "id", None) + bot_id = getattr(getattr(self, "_bot", None), "id", None) + return user_id is not None and bot_id is not None and str(user_id) == str(bot_id) + + def _reaction_feedback_enabled(self) -> bool: + """Check if inbound Telegram reaction feedback capture is enabled.""" + raw = os.getenv("TELEGRAM_REACTION_FEEDBACK") + if raw is not None and raw.strip() != "": + return raw.strip().lower() in {"true", "1", "yes", "on"} + return self._coerce_bool_extra("reaction_feedback", False) + def _link_preview_kwargs(self) -> Dict[str, Any]: if not getattr(self, "_disable_link_previews", False): return {} @@ -1249,6 +1320,13 @@ async def _try_send_rich( rich_sent_store.record(str(chat_id), str(message_id), content) except Exception: pass + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + content, + metadata, + thread_id=self._metadata_thread_id(metadata), + ) return SendResult( success=True, message_id=str(message_id) if message_id is not None else None, @@ -1259,6 +1337,7 @@ async def _try_edit_rich( chat_id: str, message_id: str, content: str, + metadata: Optional[Dict[str, Any]] = None, ) -> Optional[SendResult]: """Edit an existing message in place as a rich message (Bot API 10.1). @@ -1293,6 +1372,13 @@ async def _try_edit_rich( # rich message; treat as a successful no-op so the caller does # not fall through to a redundant legacy edit. if "not modified" in str(exc).lower(): + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + content, + metadata, + thread_id=self._metadata_thread_id(metadata), + ) return SendResult(success=True, message_id=message_id) logger.debug( "[%s] rich editMessageText rejected (%s) — falling back to MarkdownV2 edit", @@ -1300,6 +1386,13 @@ async def _try_edit_rich( ) return None if "not modified" in str(exc).lower(): + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + content, + metadata, + thread_id=self._metadata_thread_id(metadata), + ) return SendResult(success=True, message_id=message_id) err_str = str(exc).lower() try: @@ -1317,6 +1410,13 @@ async def _try_edit_rich( error=str(exc), retryable=(is_connect_timeout or not is_timeout), ) + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + content, + metadata, + thread_id=self._metadata_thread_id(metadata), + ) return SendResult(success=True, message_id=message_id) def _should_attempt_rich_draft(self, content: str) -> bool: @@ -2098,6 +2198,18 @@ def _env_float(name: str, default: float) -> float: )) # Handle inline keyboard button callbacks (update prompts) self._app.add_handler(CallbackQueryHandler(self._handle_callback_query)) + # Optional inbound reaction feedback. This is separate from the + # bot-set processing lifecycle reactions above; it captures user + # reactions to Hermes messages as reviewable feedback events. + if self._reaction_feedback_enabled(): + if MessageReactionHandler is not None: + self._app.add_handler(MessageReactionHandler(self._handle_message_reaction)) + else: + logger.warning( + "[%s] TELEGRAM_REACTION_FEEDBACK is enabled, but this " + "python-telegram-bot version has no MessageReactionHandler", + self.name, + ) # Start polling — retry initialize() for transient TLS resets try: @@ -2567,7 +2679,17 @@ async def send( await asyncio.sleep(wait) continue raise - message_ids.append(str(msg.message_id)) + sent_message_id = str(msg.message_id) + message_ids.append(sent_message_id) + self._record_reaction_feedback_target( + str(chat_id), + sent_message_id, + chunk, + metadata, + thread_id=str(thread_id) if thread_id is not None else ( + str(effective_thread_id) if effective_thread_id is not None else None + ), + ) # Re-trigger typing indicator after sending a message. # Telegram clears the typing state when a new message is delivered, @@ -2678,7 +2800,12 @@ async def edit_message( # chunks. Falls back to the legacy edit path (overflow split included) # on capability/permanent rejection. if finalize and self._rich_eligible(content): - rich_result = await self._try_edit_rich(chat_id, message_id, content) + rich_result = await self._try_edit_rich( + chat_id, + message_id, + content, + metadata=metadata, + ) if rich_result is not None: return rich_result @@ -2689,6 +2816,15 @@ async def edit_message( chat_id, message_id, content, finalize=finalize, metadata=metadata, ) + def _record_feedback_edit_target() -> None: + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + content, + metadata, + thread_id=self._metadata_thread_id(metadata), + ) + try: if not finalize: await self._bot.edit_message_text( @@ -2696,6 +2832,7 @@ async def edit_message( message_id=int(message_id), text=content, ) + _record_feedback_edit_target() return SendResult(success=True, message_id=message_id) formatted = self.format_message(content) @@ -2709,6 +2846,7 @@ async def edit_message( except Exception as fmt_err: # "Message is not modified" is a no-op, not an error if "not modified" in str(fmt_err).lower(): + _record_feedback_edit_target() return SendResult(success=True, message_id=message_id) # Fallback: strip MarkdownV2 escapes and retry as clean plain text logger.warning( @@ -2722,11 +2860,13 @@ async def edit_message( message_id=int(message_id), text=_plain, ) + _record_feedback_edit_target() return SendResult(success=True, message_id=message_id) except Exception as e: err_str = str(e).lower() # "Message is not modified" — content identical, treat as success if "not modified" in err_str: + _record_feedback_edit_target() return SendResult(success=True, message_id=message_id) # Reactive split-and-deliver: parse_mode formatting can inflate # the payload past the limit even when the raw text was under @@ -2758,6 +2898,7 @@ async def edit_message( message_id=int(message_id), text=content, ) + _record_feedback_edit_target() return SendResult(success=True, message_id=message_id) except Exception as retry_err: logger.error( @@ -2832,6 +2973,8 @@ async def _edit_overflow_split( # if truncate_message returned a single chunk just edit normally. chunks = [content] + thread_id = self._metadata_thread_id(metadata) + # Step 1 — edit the existing message with the first chunk. first_chunk = chunks[0] try: @@ -2877,6 +3020,14 @@ async def _edit_overflow_split( ) return SendResult(success=False, error=str(e)) + self._record_reaction_feedback_target( + str(chat_id), + str(message_id), + first_chunk, + metadata, + thread_id=str(thread_id) if thread_id is not None else None, + ) + # Step 2 — send each remaining chunk as a continuation message, # threaded as a reply to the previous so the user sees them as a # contiguous block. We call self._bot.send_message directly so the @@ -2886,7 +3037,6 @@ async def _edit_overflow_split( continuation_ids: list[str] = [] delivered_chunks = [first_chunk] prev_id = message_id - thread_id = self._metadata_thread_id(metadata) for chunk in chunks[1:]: sent_msg = None reply_to_id = int(prev_id) if prev_id else None @@ -2984,6 +3134,13 @@ async def _edit_overflow_split( continuation_message_ids=tuple(continuation_ids), ) new_id = str(getattr(sent_msg, "message_id", "")) or prev_id + self._record_reaction_feedback_target( + str(chat_id), + str(new_id), + chunk, + metadata, + thread_id=str(thread_id) if thread_id is not None else None, + ) continuation_ids.append(new_id) delivered_chunks.append(chunk) prev_id = new_id @@ -6745,6 +6902,82 @@ def _build_message_event( timestamp=message.date, ) + # ── Inbound user reaction feedback ─────────────────────────────────── + + async def _handle_message_reaction( + self, + update: "Update", + context: "ContextTypes.DEFAULT_TYPE", + ) -> None: + """Capture authorized user reactions as normalized feedback events. + + This is intentionally write-only local telemetry. It never sends a reply + and never mutates memory/skills/prompts directly. + """ + if not self._reaction_feedback_enabled(): + return + + reaction_update = getattr(update, "message_reaction", None) + if reaction_update is None: + return + + user = getattr(reaction_update, "user", None) + if user is None or self._reaction_actor_is_bot(user): + return + + chat = getattr(reaction_update, "chat", None) + chat_id = getattr(chat, "id", None) + message_id = getattr(reaction_update, "message_id", None) + if chat_id is None or message_id is None: + return + + try: + from gateway import reaction_feedback + except Exception: + logger.debug("[%s] reaction_feedback module unavailable", self.name, exc_info=True) + return + + target = reaction_feedback.lookup_sent_message("telegram", chat_id, message_id) + if target is None: + return + target_thread_id = target.get("thread_id") + chat_type = self._normalize_reaction_chat_type( + getattr(chat, "type", None), + str(target_thread_id) if target_thread_id is not None else None, + ) + user_id = getattr(user, "id", None) + if user_id is None: + return + + if not self._is_callback_user_authorized( + str(user_id), + chat_id=str(chat_id), + chat_type=chat_type, + thread_id=str(target_thread_id) if target_thread_id is not None else None, + user_name=getattr(user, "first_name", None), + ): + logger.debug("[%s] Ignoring reaction feedback from unauthorized user %s", self.name, user_id) + return + + new_emojis = self._reaction_emoji_values(getattr(reaction_update, "new_reaction", [])) + old_emojis = self._reaction_emoji_values(getattr(reaction_update, "old_reaction", [])) + if not new_emojis and not old_emojis: + return + + try: + reaction_feedback.record_feedback( + platform="telegram", + chat_id=str(chat_id), + message_id=str(message_id), + thread_id=str(target_thread_id) if target_thread_id is not None else None, + actor_user_id=str(user_id), + old_emojis=old_emojis, + new_emojis=new_emojis, + update_id=getattr(update, "update_id", None), + ) + except Exception: + logger.debug("[%s] Failed to record Telegram reaction feedback", self.name, exc_info=True) + # ── Message reactions (processing lifecycle) ────────────────────────── def _reactions_enabled(self) -> bool: diff --git a/gateway/reaction_feedback.py b/gateway/reaction_feedback.py new file mode 100644 index 0000000000000..5d946de30b78e --- /dev/null +++ b/gateway/reaction_feedback.py @@ -0,0 +1,285 @@ +"""Reaction feedback event store for gateway messages. + +This module is deliberately small and dependency-free. It stores two local, +profile-scoped artifacts under ``$HERMES_HOME/state``: + +* ``reaction_feedback_sent_index.json`` — bounded index of Hermes-sent message + ids with route/session metadata and content hashes (not raw text). +* ``reaction_feedback_events.jsonl`` — append-only normalized reaction events. + +The event stream is a feedback sensor only. It does not edit prompts, memory, +skills, or user profile state; downstream learning/proposal systems can read the +JSONL and decide what to propose for approval. +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import json +import os +import time +from pathlib import Path +from typing import Any, Iterable, Optional + +SCHEMA_VERSION = "reaction_feedback.v1" +_SENT_INDEX_VERSION = "sent_index.v1" +_MAX_SENT_INDEX_ENTRIES = 5000 + +_USEFUL = {"👍", "❤️", "❤", "👌", "👏"} +_MISS = {"👎"} +_UNCLEAR = {"🤔", "❓", "?"} +_BAD_TIMING = {"⏰", "😴"} +_TOO_LONG = {"📏"} + + +def _hermes_home() -> Path: + try: + from hermes_constants import get_hermes_home + + return get_hermes_home() + except Exception: + return Path(os.environ.get("HERMES_HOME") or Path.home() / ".hermes") + + +def _state_dir() -> Path: + return _hermes_home() / "state" + + +def sent_index_path() -> Path: + return _state_dir() / "reaction_feedback_sent_index.json" + + +def events_path() -> Path: + return _state_dir() / "reaction_feedback_events.jsonl" + + +def _now_iso() -> str: + return _dt.datetime.now(_dt.timezone.utc).isoformat().replace("+00:00", "Z") + + +def _key(platform: str, chat_id: Any, message_id: Any) -> str: + return f"{platform}:{chat_id}:{message_id}" + + +def _load_index() -> dict[str, Any]: + try: + data = json.loads(sent_index_path().read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return {"schema_version": _SENT_INDEX_VERSION, "messages": {}} + if not isinstance(data, dict): + return {"schema_version": _SENT_INDEX_VERSION, "messages": {}} + messages = data.get("messages") + if not isinstance(messages, dict): + messages = {} + return {"schema_version": data.get("schema_version") or _SENT_INDEX_VERSION, "messages": messages} + + +def _write_index(data: dict[str, Any]) -> None: + path = sent_index_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f".{path.name}.tmp.{os.getpid()}") + tmp.write_text(json.dumps(data, ensure_ascii=False, sort_keys=True), encoding="utf-8") + os.replace(tmp, path) + + +def _trim_messages(messages: dict[str, Any]) -> None: + overflow = len(messages) - _MAX_SENT_INDEX_ENTRIES + if overflow <= 0: + return + oldest = sorted( + messages.items(), + key=lambda item: (item[1] or {}).get("recorded_ts", 0), + )[:overflow] + for key, _ in oldest: + messages.pop(key, None) + + +def _safe_str(value: Any) -> Optional[str]: + if value is None: + return None + return str(value) + + +def _content_hash(content: Optional[str]) -> tuple[Optional[str], Optional[int]]: + if content is None: + return None, None + encoded = content.encode("utf-8") + return hashlib.sha256(encoded).hexdigest(), len(content) + + +def _metadata_value(metadata: Optional[dict[str, Any]], *keys: str) -> Optional[str]: + if not isinstance(metadata, dict): + return None + for key in keys: + value = metadata.get(key) + if value not in {None, ""}: + return str(value) + return None + + +def record_sent_message( + *, + platform: str, + chat_id: Any, + message_id: Any, + thread_id: Any = None, + content: Optional[str] = None, + metadata: Optional[dict[str, Any]] = None, + message_kind: str = "assistant", +) -> None: + """Remember a Hermes-sent message as a possible reaction target. + + Raw message text is never stored. The index keeps route/session metadata and + a SHA-256 hash/length so downstream reports can correlate reactions without + copying private assistant content into the feedback store. + """ + + if chat_id is None or message_id is None: + return + platform_s = str(platform or "").strip().lower() + if not platform_s: + return + + content_sha256, content_chars = _content_hash(content) + thread = _safe_str(thread_id) or _metadata_value( + metadata, + "thread_id", + "message_thread_id", + "direct_messages_topic_id", + "telegram_direct_messages_topic_id", + ) + entry = { + "platform": platform_s, + "chat_id": str(chat_id), + "thread_id": thread, + "message_id": str(message_id), + "message_kind": str(message_kind or "assistant"), + "session_key": _metadata_value(metadata, "session_key", "gateway_session_key"), + "session_id": _metadata_value(metadata, "session_id"), + "content_sha256": content_sha256, + "content_chars": content_chars, + "recorded_at": _now_iso(), + "recorded_ts": time.time(), + } + + try: + data = _load_index() + messages = data.setdefault("messages", {}) + messages[_key(platform_s, chat_id, message_id)] = entry + _trim_messages(messages) + _write_index(data) + except Exception: + # Feedback capture must never break user-facing delivery. + return + + +def lookup_sent_message(platform: str, chat_id: Any, message_id: Any) -> Optional[dict[str, Any]]: + if chat_id is None or message_id is None: + return None + try: + data = _load_index() + entry = data.get("messages", {}).get(_key(str(platform or "").strip().lower(), chat_id, message_id)) + return dict(entry) if isinstance(entry, dict) else None + except Exception: + return None + + +def normalize_feedback(emojis: Iterable[str]) -> dict[str, Any]: + """Map Telegram emoji reactions to the v0 feedback semantics.""" + + values = [str(e) for e in emojis if e] + if not values: + return {"emoji": None, "emojis": [], "semantic": "cleared", "action": "cleared"} + + emoji = values[0] + if emoji in _USEFUL: + semantic = "useful" + elif emoji in _MISS: + semantic = "miss" + elif emoji in _UNCLEAR: + semantic = "unclear" + elif emoji in _BAD_TIMING: + semantic = "bad_timing" + elif emoji in _TOO_LONG: + semantic = "too_long" + else: + semantic = "other" + return {"emoji": emoji, "emojis": values, "semantic": semantic, "action": "set"} + + +def _user_hash(platform: str, user_id: Any) -> Optional[str]: + if user_id in {None, ""}: + return None + return hashlib.sha256(f"{platform}:{user_id}".encode("utf-8")).hexdigest() + + +def _event_target(entry: Optional[dict[str, Any]]) -> dict[str, Any]: + if not entry: + return {"known": False} + return { + "known": True, + "message_kind": entry.get("message_kind"), + "session_key": entry.get("session_key"), + "session_id": entry.get("session_id"), + "content_sha256": entry.get("content_sha256"), + "content_chars": entry.get("content_chars"), + "sent_recorded_at": entry.get("recorded_at"), + } + + +def record_feedback( + *, + platform: str, + chat_id: Any, + message_id: Any, + thread_id: Any = None, + actor_user_id: Any = None, + old_emojis: Optional[Iterable[str]] = None, + new_emojis: Optional[Iterable[str]] = None, + update_id: Any = None, +) -> dict[str, Any]: + """Append a normalized reaction feedback event and return it.""" + + platform_s = str(platform or "").strip().lower() + target_entry = lookup_sent_message(platform_s, chat_id, message_id) + resolved_thread_id = _safe_str(thread_id) or (target_entry or {}).get("thread_id") + feedback = normalize_feedback(new_emojis or []) + old_values = [str(e) for e in (old_emojis or []) if e] + + event = { + "schema_version": SCHEMA_VERSION, + "event_type": "reaction_feedback", + "recorded_at": _now_iso(), + "platform": platform_s, + "route": { + "chat_id": str(chat_id) if chat_id is not None else None, + "thread_id": str(resolved_thread_id) if resolved_thread_id is not None else None, + "message_id": str(message_id) if message_id is not None else None, + }, + "update_id": str(update_id) if update_id is not None else None, + "actor": { + "user_id_hash": _user_hash(platform_s, actor_user_id), + }, + "reaction": { + **feedback, + "old_emojis": old_values, + }, + "target": _event_target(target_entry), + "privacy": { + "raw_text_stored": False, + "actor_user_id_stored": False, + }, + "no_auto_apply": True, + } + + path = events_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n") + except Exception: + # Keep the returned event useful for tests/callers, but never raise into + # the gateway update loop. + return event + return event diff --git a/tests/gateway/test_reaction_feedback_store.py b/tests/gateway/test_reaction_feedback_store.py new file mode 100644 index 0000000000000..25bcef7f702a8 --- /dev/null +++ b/tests/gateway/test_reaction_feedback_store.py @@ -0,0 +1,125 @@ +"""Tests for the local Telegram reaction feedback event store.""" + +from __future__ import annotations + +import hashlib +import json + +from gateway import reaction_feedback + + +def _read_json(path): + return json.loads(path.read_text(encoding="utf-8")) + + +def _read_jsonl(path): + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] + + +def test_record_sent_message_indexes_target_without_raw_text(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="-1001", + thread_id="777", + message_id="42", + content="private assistant content", + metadata={"session_id": "sess-1", "session_key": "telegram:-1001:777"}, + ) + + data = _read_json(reaction_feedback.sent_index_path()) + entry = data["messages"]["telegram:-1001:42"] + + assert entry["chat_id"] == "-1001" + assert entry["thread_id"] == "777" + assert entry["message_id"] == "42" + assert entry["session_id"] == "sess-1" + assert entry["session_key"] == "telegram:-1001:777" + assert entry["content_chars"] == len("private assistant content") + assert entry["content_sha256"] == hashlib.sha256(b"private assistant content").hexdigest() + assert "private assistant content" not in json.dumps(data, ensure_ascii=False) + + +def test_lookup_sent_message_returns_copy(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="123", + message_id="9", + content="hello", + ) + + entry = reaction_feedback.lookup_sent_message("telegram", "123", "9") + assert entry is not None + entry["message_id"] = "mutated" + + assert reaction_feedback.lookup_sent_message("telegram", "123", "9")["message_id"] == "9" + + +def test_normalize_feedback_maps_v0_semantics(): + assert reaction_feedback.normalize_feedback(["👍"])["semantic"] == "useful" + assert reaction_feedback.normalize_feedback(["❤️"])["semantic"] == "useful" + assert reaction_feedback.normalize_feedback(["👎"])["semantic"] == "miss" + assert reaction_feedback.normalize_feedback(["🤔"])["semantic"] == "unclear" + assert reaction_feedback.normalize_feedback(["⏰"])["semantic"] == "bad_timing" + assert reaction_feedback.normalize_feedback(["📏"])["semantic"] == "too_long" + assert reaction_feedback.normalize_feedback(["🧪"])["semantic"] == "other" + assert reaction_feedback.normalize_feedback([])["semantic"] == "cleared" + + +def test_record_feedback_appends_normalized_event_without_raw_actor_or_text(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="-1001", + thread_id="777", + message_id="42", + content="assistant text that must not be copied", + metadata={"session_id": "sess-1", "session_key": "telegram:-1001:777"}, + ) + + event = reaction_feedback.record_feedback( + platform="telegram", + chat_id="-1001", + message_id="42", + actor_user_id="123456", + old_emojis=[], + new_emojis=["👎"], + update_id="u1", + ) + + events = _read_jsonl(reaction_feedback.events_path()) + assert events == [event] + assert event["schema_version"] == reaction_feedback.SCHEMA_VERSION + assert event["event_type"] == "reaction_feedback" + assert event["route"] == {"chat_id": "-1001", "thread_id": "777", "message_id": "42"} + assert event["reaction"]["semantic"] == "miss" + assert event["reaction"]["emoji"] == "👎" + assert event["target"]["known"] is True + assert event["target"]["session_id"] == "sess-1" + assert event["privacy"] == {"raw_text_stored": False, "actor_user_id_stored": False} + assert event["actor"]["user_id_hash"] == hashlib.sha256(b"telegram:123456").hexdigest() + + serialized = json.dumps(event, ensure_ascii=False) + assert "assistant text that must not be copied" not in serialized + assert "123456" not in serialized + assert event["no_auto_apply"] is True + + +def test_record_feedback_cleared_reaction_keeps_old_emoji(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + event = reaction_feedback.record_feedback( + platform="telegram", + chat_id="123", + message_id="5", + actor_user_id="42", + old_emojis=["👍"], + new_emojis=[], + ) + + assert event["reaction"]["semantic"] == "cleared" + assert event["reaction"]["action"] == "cleared" + assert event["reaction"]["old_emojis"] == ["👍"] + assert event["target"]["known"] is False diff --git a/tests/gateway/test_telegram_reactions.py b/tests/gateway/test_telegram_reactions.py index 8b3b0686bb415..c02910e1b478f 100644 --- a/tests/gateway/test_telegram_reactions.py +++ b/tests/gateway/test_telegram_reactions.py @@ -10,13 +10,15 @@ from gateway.session import SessionSource -def _make_adapter(**extra_env): +def _make_adapter(**extra_config): from gateway.platforms.telegram import TelegramAdapter adapter = object.__new__(TelegramAdapter) adapter.platform = Platform.TELEGRAM - adapter.config = PlatformConfig(enabled=True, token="fake-token") + adapter.config = PlatformConfig(enabled=True, token="fake-token", extra=extra_config) + adapter._message_handler = None adapter._bot = AsyncMock() + adapter._bot.id = 999 adapter._bot.set_message_reaction = AsyncMock() return adapter @@ -36,6 +38,33 @@ def _make_event(chat_id: str = "123", message_id: str = "456") -> MessageEvent: ) +def _reaction_obj(emoji: str): + return SimpleNamespace(emoji=emoji) + + +def _make_reaction_update( + *, + user_id: int = 42, + is_bot: bool = False, + chat_id: int = -1001, + chat_type: str = "supergroup", + message_id: int = 456, + old_reaction=None, + new_reaction=None, + update_id: int = 9999, +): + return SimpleNamespace( + update_id=update_id, + message_reaction=SimpleNamespace( + user=SimpleNamespace(id=user_id, is_bot=is_bot, first_name="Tester"), + chat=SimpleNamespace(id=chat_id, type=chat_type), + message_id=message_id, + old_reaction=[_reaction_obj(e) for e in (old_reaction or [])], + new_reaction=[_reaction_obj(e) for e in (new_reaction or [])], + ), + ) + + # ── _reactions_enabled ─────────────────────────────────────────────── @@ -274,6 +303,140 @@ async def test_clear_reactions_returns_false_without_bot(monkeypatch): assert result is False +# ── inbound user reaction feedback ─────────────────────────────────── + + +def test_reaction_feedback_disabled_by_default(monkeypatch): + monkeypatch.delenv("TELEGRAM_REACTION_FEEDBACK", raising=False) + adapter = _make_adapter() + assert adapter._reaction_feedback_enabled() is False + + +def test_reaction_feedback_enabled_when_set_true(monkeypatch): + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "true") + adapter = _make_adapter() + assert adapter._reaction_feedback_enabled() is True + + +def test_reaction_feedback_env_takes_precedence_over_extra(monkeypatch): + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "false") + adapter = _make_adapter(reaction_feedback=True) + assert adapter._reaction_feedback_enabled() is False + + +def test_reaction_feedback_can_be_enabled_from_extra(monkeypatch): + monkeypatch.delenv("TELEGRAM_REACTION_FEEDBACK", raising=False) + adapter = _make_adapter(reaction_feedback=True) + assert adapter._reaction_feedback_enabled() is True + + +@pytest.mark.asyncio +async def test_handle_message_reaction_records_authorized_feedback(monkeypatch, tmp_path): + """Authorized reactions to known Hermes-sent messages become feedback events.""" + import json + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "true") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "42") + monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False) + + from gateway import reaction_feedback + + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="-1001", + thread_id="777", + message_id="456", + content="assistant text", + metadata={"session_id": "sess-1", "session_key": "telegram:-1001:777"}, + ) + adapter = _make_adapter() + + await adapter._handle_message_reaction( + _make_reaction_update(new_reaction=["👍"]), + SimpleNamespace(), + ) + + events = [ + json.loads(line) + for line in reaction_feedback.events_path().read_text(encoding="utf-8").splitlines() + if line + ] + assert len(events) == 1 + assert events[0]["route"] == {"chat_id": "-1001", "thread_id": "777", "message_id": "456"} + assert events[0]["reaction"]["semantic"] == "useful" + assert events[0]["target"]["known"] is True + assert events[0]["target"]["session_id"] == "sess-1" + + +@pytest.mark.asyncio +async def test_handle_message_reaction_ignores_unknown_target(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "true") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "42") + monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False) + + from gateway import reaction_feedback + + adapter = _make_adapter() + + await adapter._handle_message_reaction( + _make_reaction_update(user_id=42, new_reaction=["👎"]), + SimpleNamespace(), + ) + + assert not reaction_feedback.events_path().exists() + + +@pytest.mark.asyncio +async def test_handle_message_reaction_ignores_unauthorized_user(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "true") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "777") + monkeypatch.delenv("GATEWAY_ALLOW_ALL_USERS", raising=False) + + from gateway import reaction_feedback + + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="-1001", + message_id="456", + content="assistant text", + ) + adapter = _make_adapter() + + await adapter._handle_message_reaction( + _make_reaction_update(user_id=42, new_reaction=["👎"]), + SimpleNamespace(), + ) + + assert not reaction_feedback.events_path().exists() + + +@pytest.mark.asyncio +async def test_handle_message_reaction_ignores_bot_actor(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "true") + monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "42") + + from gateway import reaction_feedback + + reaction_feedback.record_sent_message( + platform="telegram", + chat_id="-1001", + message_id="456", + content="assistant text", + ) + adapter = _make_adapter() + + await adapter._handle_message_reaction( + _make_reaction_update(user_id=42, is_bot=True, new_reaction=["👎"]), + SimpleNamespace(), + ) + + assert not reaction_feedback.events_path().exists() + + # ── config.py bridging ─────────────────────────────────────────────── @@ -315,3 +478,41 @@ def test_config_reactions_env_takes_precedence(monkeypatch, tmp_path): import os assert os.getenv("TELEGRAM_REACTIONS") == "false" + + +def test_config_bridges_telegram_reaction_feedback(monkeypatch, tmp_path): + """gateway/config.py bridges telegram.reaction_feedback to TELEGRAM_REACTION_FEEDBACK.""" + import yaml + config_file = tmp_path / "config.yaml" + config_file.write_text(yaml.dump({ + "telegram": { + "reaction_feedback": True, + }, + })) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "") + + from gateway.config import load_gateway_config + load_gateway_config() + + import os + assert os.getenv("TELEGRAM_REACTION_FEEDBACK") == "true" + + +def test_config_reaction_feedback_env_takes_precedence(monkeypatch, tmp_path): + """Env var should take precedence over config.yaml for reaction feedback.""" + import yaml + config_file = tmp_path / "config.yaml" + config_file.write_text(yaml.dump({ + "telegram": { + "reaction_feedback": True, + }, + })) + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("TELEGRAM_REACTION_FEEDBACK", "false") + + from gateway.config import load_gateway_config + load_gateway_config() + + import os + assert os.getenv("TELEGRAM_REACTION_FEEDBACK") == "false"