diff --git a/gateway/life_inbox_store.py b/gateway/life_inbox_store.py new file mode 100644 index 000000000000..085a8345f316 --- /dev/null +++ b/gateway/life_inbox_store.py @@ -0,0 +1,983 @@ +"""Account-scoped life inbox storage for gateway message ingestion. + +Telegram Business/Profile Automation messages are stored in the owner's +account-scoped SQLite DB. Raw private chat text is kept out of gateway logs and +probe tables, but the main Business inbox archive stores plaintext messages so +the passive analyzer can summarize and retrieve them later. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + +PLATFORM_TELEGRAM_BUSINESS = "telegram_business" +DEFAULT_CHAT_RULE_MODE = "full_rag_selected" +SCHEMA_VERSION = 3 +BUSINESS_PAYLOAD_PROBE_LANE = "business_bot_probe" +BUSINESS_PAYLOAD_PROBE_SCENARIOS: tuple[dict[str, str], ...] = ( + { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + }, + { + "scenario_id": "S2_contact_alen_manual_outbound", + "alias": "CONTACT_1", + "expected_direction": "outgoing_from_owner", + }, + { + "scenario_id": "S3_known_noncontact_inbound", + "alias": "KNOWN_NONCONTACT_1", + "expected_direction": "incoming_to_owner", + }, + { + "scenario_id": "S4_known_noncontact_alen_manual_outbound", + "alias": "KNOWN_NONCONTACT_1", + "expected_direction": "outgoing_from_owner", + }, + { + "scenario_id": "S5_new_chat_inbound", + "alias": "NEW_CHAT_1", + "expected_direction": "incoming_to_owner", + }, + { + "scenario_id": "S6_new_chat_alen_manual_outbound", + "alias": "NEW_CHAT_1", + "expected_direction": "outgoing_from_owner", + }, +) + +_MEETING_RE = re.compile( + r"\b(встреча|созвон|звонок|колл|call|zoom|meet|meeting|appointment)\b", + re.IGNORECASE, +) +_TIME_RE = re.compile( + r"(\bзавтра\b|\bсегодня\b|\bпослезавтра\b|\bпонедельник\b|\bвторник\b|" + r"\bсред[ау]\b|\bчетверг\b|\bпятниц[ау]\b|\bсуббот[ау]\b|\bвоскресенье\b|" + r"\bmon(day)?\b|\btue(sday)?\b|\bwed(nesday)?\b|\bthu(rsday)?\b|" + r"\bfri(day)?\b|\bsat(urday)?\b|\bsun(day)?\b|" + r"\b\d{1,2}[:.]\d{2}\b|\bв\s+\d{1,2}\b|\b\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?\b)", + re.IGNORECASE, +) +_DEADLINE_RE = re.compile( + r"\b(deadline|due|дедлайн|до\s+\w+|надо\s+сдать|сдать\s+до|оплати|оплатить)\b", + re.IGNORECASE, +) +_REMINDER_RE = re.compile( + r"\b(напомни|не\s+забудь|не\s+забыть|remind|reminder)\b", + re.IGNORECASE, +) +_FOLLOW_UP_RE = re.compile( + r"\b(follow\s*up|ping|напиши|ответь|ответить|скинь|вернусь|позже|пингани)\b", + re.IGNORECASE, +) +_RESERVATION_RE = re.compile( + r"\b(бронь|брон[ьи]|reservation|ticket|билет|flight|hotel|booking)\b", + re.IGNORECASE, +) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _json_dumps(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _sha256_text(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _json_list(values: Iterable[Any]) -> str: + return _json_dumps([_coerce_text(value) for value in values]) + + +def _direction_for_sender(sender_id: Any, owner_user_chat_id: Any) -> str: + if sender_id is None or owner_user_chat_id is None: + return "unknown" + return "outgoing_from_owner" if str(sender_id) == str(owner_user_chat_id) else "incoming_to_owner" + + +def _coerce_iso(value: Any) -> str | None: + if value is None: + return None + if isinstance(value, datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.isoformat() + return str(value) + + +def _coerce_text(value: Any) -> str | None: + if value is None: + return None + return str(value) + + +def detect_candidate_reasons(text: str | None) -> list[str]: + """Return deterministic life-inbox candidate tags without persisting text. + + The tags are deliberately coarse. They let later jobs/extractors prioritize + likely actionable messages while keeping raw private chat text out of the DB. + """ + + if not text: + return [] + + checks: list[tuple[str, re.Pattern[str]]] = [ + ("meeting", _MEETING_RE), + ("time_reference", _TIME_RE), + ("deadline", _DEADLINE_RE), + ("reminder", _REMINDER_RE), + ("follow_up", _FOLLOW_UP_RE), + ("reservation", _RESERVATION_RE), + ] + return [name for name, pattern in checks if pattern.search(text)] + + +def default_life_home() -> Path: + """Return the life-management root outside Hermes runtime state.""" + + override = os.getenv("HERMES_LIFE_HOME") + if override: + return Path(override).expanduser() + return Path.home() / ".hermes-life" + + +def resolve_life_inbox_db_path(user_chat_id: str | int, *, life_home: Path | str | None = None) -> Path: + """Resolve `life_inbox.sqlite` for a Telegram Business owner account. + + Raises KeyError/FileNotFoundError when the numeric Telegram user is not bound + in `accounts.json`. Gateway call sites should catch and log this rather than + falling back to a shared/global inbox. + """ + + root = Path(life_home).expanduser() if life_home is not None else default_life_home() + registry_path = root / "accounts.json" + data = json.loads(registry_path.read_text()) + key = f"telegram:{user_chat_id}" + account = (data.get("accounts") or {}).get(key) + if not account: + raise KeyError(f"No life account registered for {key}") + + profile_rel = account.get("life_profile") + if not profile_rel: + raise KeyError(f"Life account {key} has no life_profile path") + + profile_path = Path(profile_rel) + if not profile_path.is_absolute(): + profile_path = root / profile_path + return profile_path.parent / "life_inbox.sqlite" + + +def _iter_account_db_paths(*, life_home: Path | str | None = None) -> Iterable[tuple[str, Path]]: + root = Path(life_home).expanduser() if life_home is not None else default_life_home() + data = json.loads((root / "accounts.json").read_text()) + for key, account in (data.get("accounts") or {}).items(): + if not str(key).startswith("telegram:"): + continue + profile_rel = account.get("life_profile") if isinstance(account, dict) else None + if not profile_rel: + continue + profile_path = Path(profile_rel) + if not profile_path.is_absolute(): + profile_path = root / profile_path + yield key, profile_path.parent / "life_inbox.sqlite" + + +def resolve_business_connection_user_chat_id( + connection_id: str | None, + *, + life_home: Path | str | None = None, +) -> str | None: + """Find the owner user_chat_id for a previously stored Business connection. + + This lets gateway restarts continue storing business messages without using + a process-global/shared inbox. The lookup only scans account-scoped DBs + declared in `~/.hermes-life/accounts.json`. + """ + + if not connection_id: + return None + for _account_key, db_path in _iter_account_db_paths(life_home=life_home): + if not db_path.exists(): + continue + try: + with sqlite3.connect(db_path) as conn: + row = conn.execute( + "SELECT user_chat_id FROM business_connections WHERE connection_id = ?", + (str(connection_id),), + ).fetchone() + except (OSError, sqlite3.DatabaseError): + continue + if row and row[0]: + return str(row[0]) + return None + + +class LifeInboxStore: + """Small SQLite store for selected personal/work message ingestion.""" + + def __init__(self, db_path: Path | str): + self.db_path = Path(db_path).expanduser() + self.db_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + self._secure_storage_permissions() + self._init_db() + self._secure_storage_permissions() + + def _secure_storage_permissions(self) -> None: + """Best-effort private permissions for local life-inbox metadata.""" + if os.name == "nt": + return + try: + self.db_path.parent.chmod(0o700) + except OSError: + pass + for path in (self.db_path, Path(f"{self.db_path}-wal"), Path(f"{self.db_path}-shm")): + try: + if path.exists(): + path.chmod(0o600) + except OSError: + pass + + @classmethod + def for_telegram_user_chat_id( + cls, + user_chat_id: str | int, + *, + life_home: Path | str | None = None, + ) -> "LifeInboxStore": + return cls(resolve_life_inbox_db_path(user_chat_id, life_home=life_home)) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + return conn + + def _init_db(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + PRAGMA journal_mode = WAL; + + CREATE TABLE IF NOT EXISTS schema_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS source_chats ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + chat_id TEXT NOT NULL, + chat_name TEXT, + chat_type TEXT, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE(platform, chat_id) + ); + + CREATE TABLE IF NOT EXISTS chat_rules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + chat_id TEXT NOT NULL, + rule_mode TEXT NOT NULL DEFAULT 'metadata_only', + priority TEXT NOT NULL DEFAULT 'normal', + notes TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(platform, chat_id) + ); + + CREATE TABLE IF NOT EXISTS business_connections ( + connection_id TEXT PRIMARY KEY, + update_id INTEGER, + is_enabled INTEGER, + user_chat_id TEXT, + user_id TEXT, + username TEXT, + full_name TEXT, + rights_json TEXT, + first_seen_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS business_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL DEFAULT 'telegram_business', + update_id INTEGER, + update_type TEXT NOT NULL, + connection_id TEXT, + chat_id TEXT NOT NULL, + chat_type TEXT, + chat_name TEXT, + message_id TEXT NOT NULL, + sender_id TEXT, + sender_name TEXT, + message_date TEXT, + has_text INTEGER NOT NULL DEFAULT 0, + text_len INTEGER NOT NULL DEFAULT 0, + text_sha256 TEXT, + text_preview TEXT, + raw_text_stored INTEGER NOT NULL DEFAULT 0, + candidate_reasons_json TEXT NOT NULL DEFAULT '[]', + first_seen_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(connection_id, chat_id, message_id) + ); + + CREATE TABLE IF NOT EXISTS business_message_text ( + business_message_id INTEGER PRIMARY KEY, + text TEXT NOT NULL, + text_sha256 TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(business_message_id) REFERENCES business_messages(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS business_users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + user_id TEXT NOT NULL, + username TEXT, + full_name TEXT, + is_bot INTEGER, + language_code TEXT, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE(platform, user_id) + ); + + CREATE TABLE IF NOT EXISTS chat_participants ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + platform TEXT NOT NULL, + chat_id TEXT NOT NULL, + user_id TEXT NOT NULL, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE(platform, chat_id, user_id) + ); + + CREATE INDEX IF NOT EXISTS idx_business_messages_chat_date + ON business_messages(chat_id, message_date); + CREATE INDEX IF NOT EXISTS idx_business_messages_updated + ON business_messages(updated_at); + CREATE INDEX IF NOT EXISTS idx_business_users_username + ON business_users(platform, username); + CREATE INDEX IF NOT EXISTS idx_chat_participants_user + ON chat_participants(platform, user_id); + + CREATE TABLE IF NOT EXISTS business_payload_probe_scenarios ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_lane TEXT NOT NULL, + scenario_id TEXT NOT NULL, + alias TEXT, + expected_direction TEXT, + probe_text_len INTEGER NOT NULL, + probe_text_sha256 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + matched_event_id INTEGER, + created_at TEXT NOT NULL, + matched_at TEXT, + notes TEXT, + UNIQUE(source_lane, scenario_id), + UNIQUE(source_lane, probe_text_sha256, probe_text_len) + ); + + CREATE INDEX IF NOT EXISTS idx_business_payload_probe_scenarios_status + ON business_payload_probe_scenarios(source_lane, status); + + CREATE TABLE IF NOT EXISTS business_payload_probe_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_lane TEXT NOT NULL, + scenario_id TEXT, + update_id INTEGER, + update_type TEXT NOT NULL, + connection_id TEXT, + chat_id TEXT, + message_id TEXT, + sender_id TEXT, + direction TEXT NOT NULL DEFAULT 'unknown', + message_date TEXT, + has_text INTEGER NOT NULL DEFAULT 0, + text_len INTEGER NOT NULL DEFAULT 0, + text_sha256 TEXT, + raw_text_stored INTEGER NOT NULL DEFAULT 0, + field_availability_json TEXT NOT NULL DEFAULT '{}', + payload_shape_json TEXT NOT NULL DEFAULT '{}', + media_json TEXT NOT NULL DEFAULT '{}', + reply_context_json TEXT NOT NULL DEFAULT '{}', + deleted_message_ids_json TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_business_payload_probe_events_scenario + ON business_payload_probe_events(source_lane, scenario_id, created_at); + CREATE INDEX IF NOT EXISTS idx_business_payload_probe_events_message + ON business_payload_probe_events(source_lane, connection_id, chat_id, message_id); + + CREATE TABLE IF NOT EXISTS deleted_business_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + update_id INTEGER, + connection_id TEXT, + chat_id TEXT, + message_ids_json TEXT NOT NULL, + created_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS life_signals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message_id INTEGER, + signal_type TEXT NOT NULL, + title TEXT, + confidence REAL, + status TEXT NOT NULL DEFAULT 'pending', + extracted_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(message_id) REFERENCES business_messages(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + action TEXT NOT NULL, + actor TEXT NOT NULL, + source TEXT, + payload_hash TEXT, + created_at TEXT NOT NULL + ); + """ + ) + conn.execute( + "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", + ("schema_version", str(SCHEMA_VERSION)), + ) + + def _ensure_chat( + self, + conn: sqlite3.Connection, + *, + platform: str, + chat_id: str, + chat_name: str | None, + chat_type: str | None, + now: str, + ) -> None: + conn.execute( + """ + INSERT INTO source_chats(platform, chat_id, chat_name, chat_type, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(platform, chat_id) DO UPDATE SET + chat_name = excluded.chat_name, + chat_type = excluded.chat_type, + last_seen_at = excluded.last_seen_at + """, + (platform, chat_id, chat_name, chat_type, now, now), + ) + conn.execute( + """ + INSERT OR IGNORE INTO chat_rules(platform, chat_id, rule_mode, priority, created_at, updated_at) + VALUES (?, ?, ?, 'normal', ?, ?) + """, + (platform, chat_id, DEFAULT_CHAT_RULE_MODE, now, now), + ) + + def _upsert_business_user( + self, + conn: sqlite3.Connection, + *, + user_id: str | None, + username: str | None, + full_name: str | None, + is_bot: bool | None, + language_code: str | None, + now: str, + ) -> None: + if not user_id: + return + conn.execute( + """ + INSERT INTO business_users( + platform, user_id, username, full_name, is_bot, language_code, + first_seen_at, last_seen_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(platform, user_id) DO UPDATE SET + username = COALESCE(excluded.username, business_users.username), + full_name = COALESCE(excluded.full_name, business_users.full_name), + is_bot = COALESCE(excluded.is_bot, business_users.is_bot), + language_code = COALESCE(excluded.language_code, business_users.language_code), + last_seen_at = excluded.last_seen_at + """, + ( + PLATFORM_TELEGRAM_BUSINESS, + user_id, + username, + full_name, + None if is_bot is None else int(bool(is_bot)), + language_code, + now, + now, + ), + ) + + def _ensure_chat_participant( + self, + conn: sqlite3.Connection, + *, + chat_id: str, + user_id: str | None, + now: str, + ) -> None: + if not user_id: + return + conn.execute( + """ + INSERT INTO chat_participants(platform, chat_id, user_id, first_seen_at, last_seen_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(platform, chat_id, user_id) DO UPDATE SET + last_seen_at = excluded.last_seen_at + """, + (PLATFORM_TELEGRAM_BUSINESS, chat_id, user_id, now, now), + ) + + def prepare_business_payload_probe_scenarios( + self, + scenarios: Iterable[dict[str, Any]], + *, + source_lane: str = BUSINESS_PAYLOAD_PROBE_LANE, + ) -> None: + """Register strict live-probe scenario codes without storing raw text. + + `probe_text` is accepted only long enough to calculate SHA-256 + length. + The plaintext code is intentionally not persisted, so later gateway + ingestion can match probe messages by hash while keeping DB/logs free of + raw chat content. + """ + + now = _utc_now_iso() + with self._connect() as conn: + for scenario in scenarios: + scenario_id = str(scenario["scenario_id"]) + probe_text = str(scenario["probe_text"]) + conn.execute( + """ + INSERT INTO business_payload_probe_scenarios( + source_lane, scenario_id, alias, expected_direction, + probe_text_len, probe_text_sha256, status, + matched_event_id, created_at, matched_at, notes + ) + VALUES (?, ?, ?, ?, ?, ?, 'pending', NULL, ?, NULL, ?) + ON CONFLICT(source_lane, scenario_id) DO UPDATE SET + alias = excluded.alias, + expected_direction = excluded.expected_direction, + probe_text_len = excluded.probe_text_len, + probe_text_sha256 = excluded.probe_text_sha256, + status = 'pending', + matched_event_id = NULL, + matched_at = NULL, + notes = excluded.notes + """, + ( + source_lane, + scenario_id, + _coerce_text(scenario.get("alias")), + _coerce_text(scenario.get("expected_direction")), + len(probe_text), + _sha256_text(probe_text), + now, + _coerce_text(scenario.get("notes")), + ), + ) + + def _match_probe_scenario_by_text( + self, + conn: sqlite3.Connection, + *, + source_lane: str, + text_sha256: str | None, + text_len: int, + ) -> sqlite3.Row | None: + if not text_sha256: + return None + return conn.execute( + """ + SELECT id, scenario_id + FROM business_payload_probe_scenarios + WHERE source_lane = ? AND probe_text_sha256 = ? AND probe_text_len = ? + ORDER BY CASE status WHEN 'pending' THEN 0 WHEN 'matched' THEN 1 ELSE 2 END, id + LIMIT 1 + """, + (source_lane, text_sha256, text_len), + ).fetchone() + + def _match_probe_scenario_by_message_identity( + self, + conn: sqlite3.Connection, + *, + source_lane: str, + connection_id: str | None, + chat_id: str | None, + message_id: str | None, + deleted_message_ids: Iterable[Any] | None = None, + ) -> str | None: + candidate_ids = [] + if message_id: + candidate_ids.append(str(message_id)) + if deleted_message_ids: + candidate_ids.extend(str(value) for value in deleted_message_ids if value is not None) + if not candidate_ids: + return None + placeholders = ",".join("?" for _ in candidate_ids) + params: list[Any] = [source_lane, connection_id, chat_id, *candidate_ids] + row = conn.execute( + f""" + SELECT scenario_id + FROM business_payload_probe_events + WHERE source_lane = ? + AND connection_id IS ? + AND chat_id IS ? + AND message_id IN ({placeholders}) + AND scenario_id IS NOT NULL + ORDER BY id DESC + LIMIT 1 + """, + params, + ).fetchone() + return str(row["scenario_id"]) if row and row["scenario_id"] else None + + def record_business_payload_probe_event( + self, + *, + update_id: int | None, + update_type: str, + connection_id: str | None, + owner_user_chat_id: str | int | None, + chat_id: str | int | None, + message_id: str | int | None = None, + sender_id: str | int | None = None, + text: str | None = None, + message_date: Any = None, + field_availability: Any | None = None, + payload_shape: Any | None = None, + media: Any | None = None, + reply_context: Any | None = None, + deleted_message_ids: Iterable[Any] | None = None, + source_lane: str = BUSINESS_PAYLOAD_PROBE_LANE, + capture_all: bool = False, + ) -> int | None: + """Record a sanitized Telegram Business payload-probe event. + + By default the store records only events that match a prepared scenario + code by SHA-256/length, or follow-up edit/delete events for an already + matched message. Set `capture_all=True` for temporary shape-only capture + of all Business payloads. Raw text is never stored. + """ + + now = _utc_now_iso() + text_value = text or "" + text_len = len(text_value) + text_sha256 = _sha256_text(text_value) if text_value else None + connection_key = _coerce_text(connection_id) + chat_id_text = _coerce_text(chat_id) + message_id_text = _coerce_text(message_id) + sender_id_text = _coerce_text(sender_id) + deleted_ids = [value for value in (deleted_message_ids or [])] + + with self._connect() as conn: + scenario_row = self._match_probe_scenario_by_text( + conn, + source_lane=source_lane, + text_sha256=text_sha256, + text_len=text_len, + ) + scenario_id = str(scenario_row["scenario_id"]) if scenario_row else None + if scenario_id is None: + scenario_id = self._match_probe_scenario_by_message_identity( + conn, + source_lane=source_lane, + connection_id=connection_key, + chat_id=chat_id_text, + message_id=message_id_text, + deleted_message_ids=deleted_ids, + ) + if scenario_id is None and not capture_all: + return None + + cur = conn.execute( + """ + INSERT INTO business_payload_probe_events( + source_lane, scenario_id, update_id, update_type, connection_id, + chat_id, message_id, sender_id, direction, message_date, + has_text, text_len, text_sha256, raw_text_stored, + field_availability_json, payload_shape_json, media_json, + reply_context_json, deleted_message_ids_json, created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?) + """, + ( + source_lane, + scenario_id, + update_id, + update_type, + connection_key, + chat_id_text, + message_id_text, + sender_id_text, + _direction_for_sender(sender_id_text, owner_user_chat_id), + _coerce_iso(message_date), + int(bool(text_value)), + text_len, + text_sha256, + _json_dumps(field_availability or {}), + _json_dumps(payload_shape or {}), + _json_dumps(media or {}), + _json_dumps(reply_context or {}), + _json_list(deleted_ids), + now, + ), + ) + if cur.lastrowid is None: + raise RuntimeError("payload probe event insert succeeded but row id is unavailable") + event_id = int(cur.lastrowid) + if scenario_row is not None: + conn.execute( + """ + UPDATE business_payload_probe_scenarios + SET status = 'matched', matched_event_id = ?, matched_at = ? + WHERE id = ? + """, + (event_id, now, int(scenario_row["id"])), + ) + elif scenario_id is not None: + conn.execute( + """ + UPDATE business_payload_probe_scenarios + SET status = 'matched', matched_event_id = COALESCE(matched_event_id, ?), + matched_at = COALESCE(matched_at, ?) + WHERE source_lane = ? AND scenario_id = ? + """, + (event_id, now, source_lane, scenario_id), + ) + return event_id + + def record_business_connection( + self, + *, + update_id: int | None, + connection_id: str, + is_enabled: bool | None, + user_chat_id: str | int | None, + user_id: str | int | None, + username: str | None, + full_name: str | None, + rights: Any, + ) -> None: + now = _utc_now_iso() + with self._connect() as conn: + conn.execute( + """ + INSERT INTO business_connections( + connection_id, update_id, is_enabled, user_chat_id, user_id, + username, full_name, rights_json, first_seen_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(connection_id) DO UPDATE SET + update_id = excluded.update_id, + is_enabled = excluded.is_enabled, + user_chat_id = excluded.user_chat_id, + user_id = excluded.user_id, + username = excluded.username, + full_name = excluded.full_name, + rights_json = excluded.rights_json, + updated_at = excluded.updated_at + """, + ( + connection_id, + update_id, + None if is_enabled is None else int(bool(is_enabled)), + _coerce_text(user_chat_id), + _coerce_text(user_id), + username, + full_name, + _json_dumps(rights), + now, + now, + ), + ) + + def record_business_message( + self, + *, + update_id: int | None, + update_type: str, + connection_id: str | None, + chat_id: str | int | None, + chat_type: str | None, + chat_name: str | None, + chat_username: str | None = None, + message_id: str | int | None, + sender_id: str | int | None, + sender_name: str | None, + sender_username: str | None = None, + sender_is_bot: bool | None = None, + sender_language_code: str | None = None, + text: str | None, + message_date: Any, + ) -> int: + now = _utc_now_iso() + connection_key = _coerce_text(connection_id) + chat_id_text = _coerce_text(chat_id) + message_id_text = _coerce_text(message_id) + sender_id_text = _coerce_text(sender_id) + chat_username_text = _coerce_text(chat_username) + sender_username_text = _coerce_text(sender_username) + sender_language_code_text = _coerce_text(sender_language_code) + if not connection_key: + raise ValueError("connection_id is required for Telegram Business messages") + if not chat_id_text: + raise ValueError("chat_id is required for Telegram Business messages") + if not message_id_text: + raise ValueError("message_id is required for Telegram Business messages") + text_value = text or "" + text_sha256 = _sha256_text(text_value) if text_value else None + candidate_reasons = detect_candidate_reasons(text_value) + + with self._connect() as conn: + self._ensure_chat( + conn, + platform=PLATFORM_TELEGRAM_BUSINESS, + chat_id=chat_id_text, + chat_name=chat_name, + chat_type=chat_type, + now=now, + ) + if str(chat_type or "").lower() == "private": + self._upsert_business_user( + conn, + user_id=chat_id_text, + username=chat_username_text, + full_name=chat_name, + is_bot=None, + language_code=None, + now=now, + ) + self._ensure_chat_participant(conn, chat_id=chat_id_text, user_id=chat_id_text, now=now) + self._upsert_business_user( + conn, + user_id=sender_id_text, + username=sender_username_text, + full_name=sender_name, + is_bot=sender_is_bot, + language_code=sender_language_code_text, + now=now, + ) + self._ensure_chat_participant(conn, chat_id=chat_id_text, user_id=sender_id_text, now=now) + conn.execute( + """ + INSERT INTO business_messages( + platform, update_id, update_type, connection_id, chat_id, chat_type, chat_name, + message_id, sender_id, sender_name, message_date, has_text, text_len, + text_sha256, text_preview, raw_text_stored, candidate_reasons_json, + first_seen_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?) + ON CONFLICT(connection_id, chat_id, message_id) DO UPDATE SET + update_id = excluded.update_id, + update_type = excluded.update_type, + chat_type = excluded.chat_type, + chat_name = excluded.chat_name, + sender_id = excluded.sender_id, + sender_name = excluded.sender_name, + message_date = excluded.message_date, + has_text = excluded.has_text, + text_len = excluded.text_len, + text_sha256 = excluded.text_sha256, + text_preview = NULL, + raw_text_stored = excluded.raw_text_stored, + candidate_reasons_json = excluded.candidate_reasons_json, + updated_at = excluded.updated_at + """, + ( + PLATFORM_TELEGRAM_BUSINESS, + update_id, + update_type, + connection_key, + chat_id_text, + chat_type, + chat_name, + message_id_text, + sender_id_text, + sender_name, + _coerce_iso(message_date), + int(bool(text_value)), + len(text_value), + text_sha256, + int(bool(text_value)), + _json_dumps(candidate_reasons), + now, + now, + ), + ) + row = conn.execute( + """ + SELECT id FROM business_messages + WHERE connection_id IS ? AND chat_id = ? AND message_id = ? + """, + (connection_key, chat_id_text, message_id_text), + ).fetchone() + if row is None: + raise RuntimeError("business message insert succeeded but row lookup failed") + business_message_id = int(row["id"]) + if text_value: + conn.execute( + """ + INSERT INTO business_message_text( + business_message_id, text, text_sha256, created_at, updated_at + ) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(business_message_id) DO UPDATE SET + text = excluded.text, + text_sha256 = excluded.text_sha256, + updated_at = excluded.updated_at + """, + (business_message_id, text_value, text_sha256, now, now), + ) + else: + conn.execute( + "DELETE FROM business_message_text WHERE business_message_id = ?", + (business_message_id,), + ) + return business_message_id + + def record_deleted_business_messages( + self, + *, + update_id: int | None, + connection_id: str | None, + chat_id: str | int | None, + message_ids: Iterable[Any], + ) -> None: + now = _utc_now_iso() + with self._connect() as conn: + conn.execute( + """ + INSERT INTO deleted_business_messages(update_id, connection_id, chat_id, message_ids_json, created_at) + VALUES (?, ?, ?, ?, ?) + """, + ( + update_id, + connection_id, + _coerce_text(chat_id), + _json_dumps([_coerce_text(message_id) for message_id in message_ids]), + now, + ), + ) diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index 799a836df735..227c79c8a5fe 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -31,6 +31,7 @@ CommandHandler, CallbackQueryHandler, MessageHandler as TelegramMessageHandler, + TypeHandler, ContextTypes, filters, ) @@ -49,6 +50,7 @@ CommandHandler = Any CallbackQueryHandler = Any TelegramMessageHandler = Any + TypeHandler = None HTTPXRequest = Any filters = None ParseMode = None @@ -118,7 +120,7 @@ def check_telegram_requirements() -> bool: """ global TELEGRAM_AVAILABLE, Update, Bot, Message, InlineKeyboardButton global InlineKeyboardMarkup, LinkPreviewOptions, Application - global CommandHandler, CallbackQueryHandler, TelegramMessageHandler + global CommandHandler, CallbackQueryHandler, TelegramMessageHandler, TypeHandler global ContextTypes, filters, ParseMode, ChatType, HTTPXRequest if TELEGRAM_AVAILABLE: return True @@ -138,6 +140,7 @@ def check_telegram_requirements() -> bool: Application as _App, CommandHandler as _CH, CallbackQueryHandler as _CQH, MessageHandler as _MH, + TypeHandler as _TH, ContextTypes as _CT, filters as _filters, ) from telegram.constants import ParseMode as _PM, ChatType as _CtT @@ -154,6 +157,7 @@ def check_telegram_requirements() -> bool: CommandHandler = _CH CallbackQueryHandler = _CQH TelegramMessageHandler = _MH + TypeHandler = _TH ContextTypes = _CT filters = _filters ParseMode = _PM @@ -425,6 +429,10 @@ def __init__(self, config: PlatformConfig): ) self._pending_text_batches: Dict[str, MessageEvent] = {} self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {} + # Telegram Business/Profile Automation updates are passive inbox data, + # not chat prompts. Keep connection-owner mappings so message updates + # can be stored in the correct per-account life inbox. + self._business_connection_user_chat_ids: Dict[str, str] = {} self._polling_error_task: Optional[asyncio.Task] = None self._polling_conflict_count: int = 0 self._polling_network_error_count: int = 0 @@ -1429,7 +1437,12 @@ def _env_float(name: str, default: float) -> float: self._app = builder.build() self._bot = self._app.bot - # Register handlers + # Register handlers. Business/Profile Automation updates are + # observed in a separate pre-processing group so connection + # metadata/deletions are captured, while regular message handlers + # still run for non-business updates. + if TypeHandler is not None: + self._app.add_handler(TypeHandler(Update, self._handle_business_update), group=-1) self._app.add_handler(TelegramMessageHandler( filters.TEXT & ~filters.COMMAND, self._handle_text_message @@ -1679,6 +1692,18 @@ async def send( """Send a message to a Telegram chat.""" if not self._bot: return SendResult(success=False, error="Not connected") + + if self._is_own_bot_id(chat_id): + logger.warning( + "[%s] Refusing to send Telegram message to own bot chat_id=%s", + self.name, + chat_id, + ) + # Treat as delivered from the gateway's perspective: this is a + # permanent routing bug/stale session, not a transient delivery + # failure. Returning success prevents retry/fallback loops that + # would keep hitting Telegram with an impossible bot-to-bot DM. + return SendResult(success=True, message_id=None) # Skip whitespace-only text to prevent Telegram 400 empty-text errors. if not content or not content.strip(): @@ -4651,6 +4676,10 @@ def _should_process_message(self, message: Message, *, is_command: bool = False) mentioning the bot (``@botname /command``), both of which are recognised as mentions by :meth:`_message_mentions_bot`. """ + if self._message_from_own_bot(message): + logger.info("[%s] Ignoring Telegram message from own bot", self.name) + return False + if not self._is_group_chat(message): return True @@ -4742,6 +4771,498 @@ def _effective_update_message(self, update: Update) -> Optional[Message]: """ return getattr(update, "effective_message", None) or getattr(update, "message", None) + @staticmethod + def _to_plain_dict(value: Any) -> Any: + if value is None: + return {} + to_dict = getattr(value, "to_dict", None) + if callable(to_dict): + try: + return to_dict() + except Exception: + return {} + if isinstance(value, (dict, list, str, int, float, bool)): + return value + return str(value) + + @classmethod + def _payload_shape_only(cls, value: Any, *, _depth: int = 0, _max_depth: int = 5) -> dict[str, Any]: + """Return payload key/type shape without storing primitive values. + + Telegram Business live probes need field availability and object shape, + not private text, names, file ids, or other raw values. This helper is + deliberately value-blind: every primitive becomes only its type name. + """ + + if value is None or cls._is_unset_mock(value): + return {"type": "null"} + if _depth >= _max_depth: + return {"type": type(value).__name__, "truncated": True} + + if isinstance(value, dict): + return { + "type": "object", + "keys": { + str(key): cls._payload_shape_only(item, _depth=_depth + 1, _max_depth=_max_depth) + for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) + }, + } + if isinstance(value, (list, tuple)): + first_shape = ( + cls._payload_shape_only(value[0], _depth=_depth + 1, _max_depth=_max_depth) + if value + else None + ) + result: dict[str, Any] = {"type": "list", "length": len(value)} + if first_shape is not None: + result["first_item"] = first_shape + return result + if isinstance(value, (str, int, float, bool)): + return {"type": type(value).__name__} + + to_dict = getattr(value, "to_dict", None) + if callable(to_dict): + try: + return cls._payload_shape_only(to_dict(), _depth=_depth, _max_depth=_max_depth) + except Exception: + return {"type": type(value).__name__, "to_dict_error": True} + if hasattr(value, "__dict__"): + public_attrs = { + key: item + for key, item in vars(value).items() + if not key.startswith("_") and not callable(item) + } + return cls._payload_shape_only(public_attrs, _depth=_depth, _max_depth=_max_depth) + return {"type": type(value).__name__} + + @classmethod + def _field_present(cls, obj: Any, attr: str) -> bool: + if obj is None or cls._is_unset_mock(obj): + return False + value = getattr(obj, attr, None) + if value is None or cls._is_unset_mock(value): + return False + if isinstance(value, str): + return bool(value) + if isinstance(value, (list, tuple, set, dict)): + return bool(value) + return True + + @classmethod + def _text_present(cls, obj: Any, attr: str) -> bool: + if obj is None or cls._is_unset_mock(obj): + return False + value = getattr(obj, attr, None) + return bool(value) and not cls._is_unset_mock(value) + + @classmethod + def _business_payload_field_availability(cls, update: Any, message: Any) -> dict[str, Any]: + sender = getattr(message, "from_user", None) or getattr(message, "sender_chat", None) + chat = getattr(message, "chat", None) + effective_message = getattr(update, "effective_message", None) or getattr(update, "message", None) + media_fields = ( + "animation", + "audio", + "contact", + "dice", + "document", + "game", + "location", + "paid_media", + "photo", + "poll", + "sticker", + "story", + "venue", + "video", + "video_note", + "voice", + ) + message_fields = { + "has_text": cls._text_present(message, "text"), + "has_caption": cls._text_present(message, "caption"), + "has_entities": cls._field_present(message, "entities"), + "has_caption_entities": cls._field_present(message, "caption_entities"), + "has_reply_to_message": cls._field_present(message, "reply_to_message"), + "has_external_reply": cls._field_present(message, "external_reply"), + "has_quote": cls._field_present(message, "quote"), + "has_forward_origin": cls._field_present(message, "forward_origin"), + "has_media_group_id": cls._field_present(message, "media_group_id"), + "has_sender_business_bot": cls._field_present(message, "sender_business_bot"), + "has_is_from_offline": cls._field_present(message, "is_from_offline"), + "has_edit_date": cls._field_present(message, "edit_date"), + "has_business_connection_id": cls._field_present(message, "business_connection_id"), + } + for field in media_fields: + message_fields[f"has_{field}"] = cls._field_present(message, field) + return { + "update": { + "has_business_connection": cls._field_present(update, "business_connection"), + "has_business_message": cls._field_present(update, "business_message"), + "has_edited_business_message": cls._field_present(update, "edited_business_message"), + "has_deleted_business_messages": cls._field_present(update, "deleted_business_messages"), + "has_regular_message": cls._field_present(update, "message"), + "has_effective_message": effective_message is not None and not cls._is_unset_mock(effective_message), + "regular_message_has_business_connection_id": bool( + effective_message and cls._business_connection_id_from_message(effective_message) + ), + }, + "chat": { + "chat_type": str(getattr(chat, "type", "")) or None, + "has_username": cls._field_present(chat, "username"), + "has_title": cls._field_present(chat, "title"), + "has_full_name": cls._field_present(chat, "full_name"), + }, + "sender": { + "has_from_user": sender is not None and not cls._is_unset_mock(sender), + "has_username": cls._field_present(sender, "username"), + "has_is_bot": cls._field_present(sender, "is_bot"), + "has_language_code": cls._field_present(sender, "language_code"), + "has_is_premium": cls._field_present(sender, "is_premium"), + }, + "message": message_fields, + } + + @classmethod + def _business_payload_media_summary(cls, message: Any) -> dict[str, Any]: + media_fields = ( + "animation", + "audio", + "contact", + "document", + "location", + "paid_media", + "photo", + "poll", + "sticker", + "venue", + "video", + "video_note", + "voice", + ) + summary: dict[str, Any] = {} + for field in media_fields: + value = getattr(message, field, None) + if value is None or cls._is_unset_mock(value): + continue + if isinstance(value, (list, tuple, set, dict)) and not value: + continue + summary[field] = {"present": True, "count": len(value) if isinstance(value, (list, tuple, set, dict)) else 1} + return summary + + @classmethod + def _business_payload_reply_summary(cls, message: Any) -> dict[str, Any]: + reply = getattr(message, "reply_to_message", None) + if reply is None or cls._is_unset_mock(reply): + return {} + return { + "present": True, + "message_id": str(getattr(reply, "message_id", "")) or None, + "has_text": cls._text_present(reply, "text"), + "has_caption": cls._text_present(reply, "caption"), + } + + def _business_payload_probe_capture_all(self) -> bool: + configured = None + if getattr(self.config, "extra", None): + configured = self.config.extra.get("business_payload_probe_capture_all") + if configured is None: + configured = os.getenv("HERMES_TELEGRAM_BUSINESS_PAYLOAD_PROBE", "") + if isinstance(configured, str): + return configured.strip().lower() in {"1", "true", "yes", "on"} + return bool(configured) + + @staticmethod + def _business_connection_id_from_message(message: Any) -> Optional[str]: + connection_id = getattr(message, "business_connection_id", None) + if connection_id is None: + return None + connection_id = str(connection_id).strip() + return connection_id or None + + @staticmethod + def _chat_display_name(chat: Any) -> Optional[str]: + for attr in ("title", "full_name", "username", "first_name", "name"): + value = getattr(chat, attr, None) + if value: + return str(value) + return None + + @staticmethod + def _user_display_name(user: Any) -> Optional[str]: + for attr in ("full_name", "username", "first_name", "name"): + value = getattr(user, attr, None) + if value: + return str(value) + return None + + def _own_bot_id(self) -> Optional[str]: + if not self._bot: + return None + bot_id = getattr(self._bot, "id", None) + if bot_id is None: + return None + return str(bot_id) + + def _is_own_bot_id(self, value: Any) -> bool: + own_bot_id = self._own_bot_id() + return bool(own_bot_id and value is not None and str(value) == own_bot_id) + + def _message_from_own_bot(self, message: Any) -> bool: + sender = getattr(message, "from_user", None) + return self._is_own_bot_id(getattr(sender, "id", None)) + + @staticmethod + def _is_unset_mock(value: Any) -> bool: + # Unit-test MagicMock objects fabricate arbitrary attributes on access. + # Treat those synthetic placeholders as absent so normal Telegram + # messages are not mistaken for Business updates. + return type(value).__module__.startswith("unittest.mock") + + def _is_business_update(self, update: Any) -> bool: + for attr in ( + "business_connection", + "business_message", + "edited_business_message", + "deleted_business_messages", + ): + value = getattr(update, attr, None) + if value is not None and not self._is_unset_mock(value): + return True + msg = self._effective_update_message(update) + return bool(msg and not self._is_unset_mock(msg) and self._business_connection_id_from_message(msg)) + + def _iter_business_message_payloads(self, update: Any): + seen_object_ids: set[int] = set() + seen_message_keys: set[tuple[str, str | None, str | None, str | None]] = set() + + def should_yield(update_type: str, message: Any) -> bool: + if message is None or self._is_unset_mock(message): + return False + object_id = id(message) + if object_id in seen_object_ids: + return False + + connection_id = self._business_connection_id_from_message(message) + chat = getattr(message, "chat", None) + chat_id = None if chat is None or self._is_unset_mock(chat) else getattr(chat, "id", None) + message_id = getattr(message, "message_id", None) + message_key = ( + update_type, + str(connection_id) if connection_id is not None else None, + str(chat_id) if chat_id is not None else None, + str(message_id) if message_id is not None else None, + ) + has_semantic_key = any(part is not None for part in message_key[1:]) + if has_semantic_key and message_key in seen_message_keys: + return False + + seen_object_ids.add(object_id) + if has_semantic_key: + seen_message_keys.add(message_key) + return True + + for attr, update_type in ( + ("business_message", "business_message"), + ("edited_business_message", "edited_business_message"), + ): + message = getattr(update, attr, None) + if should_yield(update_type, message): + yield update_type, message + + # Some Bot API/PTB paths expose Business/Profile Automation messages as + # the ordinary update.message/effective_message while preserving + # Message.business_connection_id. Treat those as business inbox data too; + # otherwise owner-sent replies look like authorized prompts and launch + # the LLM, while counterpart messages are rejected as unauthorized. + for message in (getattr(update, "message", None), self._effective_update_message(update)): + if message is None or self._is_unset_mock(message): + continue + if not self._business_connection_id_from_message(message): + continue + if should_yield("business_message", message): + yield "business_message", message + + def _record_business_connection_update(self, update: Any, business_connection: Any) -> None: + connection_id = getattr(business_connection, "id", None) + user_chat_id = getattr(business_connection, "user_chat_id", None) + if not connection_id or user_chat_id is None: + logger.warning("[Telegram Business Inbox] skipping connection store: missing connection or owner id") + return + + user = getattr(business_connection, "user", None) + try: + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore.for_telegram_user_chat_id(user_chat_id) + store.record_business_connection( + update_id=getattr(update, "update_id", None), + connection_id=str(connection_id), + is_enabled=getattr(business_connection, "is_enabled", None), + user_chat_id=user_chat_id, + user_id=getattr(user, "id", None), + username=getattr(user, "username", None), + full_name=self._user_display_name(user), + rights=self._to_plain_dict(getattr(business_connection, "rights", None)), + ) + self._business_connection_user_chat_ids[str(connection_id)] = str(user_chat_id) + except Exception as exc: + logger.warning("[Telegram Business Inbox] skipping connection store: %s", exc) + + def _business_owner_for_connection(self, connection_id: Optional[str]) -> Optional[str]: + if not connection_id: + return None + cached = self._business_connection_user_chat_ids.get(connection_id) + if cached: + return cached + try: + from gateway.life_inbox_store import resolve_business_connection_user_chat_id + + owner = resolve_business_connection_user_chat_id(connection_id) + except Exception: + owner = None + if owner: + self._business_connection_user_chat_ids[connection_id] = str(owner) + return str(owner) + return None + + def _record_business_message_update(self, update: Any, update_type: str, message: Any) -> None: + connection_id = self._business_connection_id_from_message(message) + owner_user_chat_id = self._business_owner_for_connection(connection_id) + if not owner_user_chat_id: + logger.warning("[Telegram Business Inbox] skipping message store: owner mapping unavailable") + return + + chat = getattr(message, "chat", None) + sender = getattr(message, "from_user", None) or getattr(message, "sender_chat", None) + chat_id = getattr(chat, "id", None) + sender_id = getattr(sender, "id", None) + if self._is_own_bot_id(chat_id) or self._is_own_bot_id(sender_id): + # Telegram Business/Profile Automation mirrors the owner's DM with + # this very bot as business_message updates too. Those copies are + # not life inbox data and would pollute storage with Hermes' own + # prompts/responses. + logger.debug("[Telegram Business Inbox] skipping own-bot business copy") + return + text = getattr(message, "text", None) or getattr(message, "caption", None) + try: + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore.for_telegram_user_chat_id(owner_user_chat_id) + store.record_business_message( + update_id=getattr(update, "update_id", None), + update_type=update_type, + connection_id=connection_id, + chat_id=chat_id, + chat_type=str(getattr(chat, "type", "")) or None, + chat_name=self._chat_display_name(chat), + chat_username=getattr(chat, "username", None), + message_id=getattr(message, "message_id", None), + sender_id=sender_id, + sender_name=self._user_display_name(sender), + sender_username=getattr(sender, "username", None), + sender_is_bot=getattr(sender, "is_bot", None), + sender_language_code=getattr(sender, "language_code", None), + text=text, + message_date=getattr(message, "date", None), + ) + store.record_business_payload_probe_event( + update_id=getattr(update, "update_id", None), + update_type=update_type, + connection_id=connection_id, + owner_user_chat_id=owner_user_chat_id, + chat_id=chat_id, + message_id=getattr(message, "message_id", None), + sender_id=sender_id, + text=text, + message_date=getattr(message, "date", None), + field_availability=self._business_payload_field_availability(update, message), + payload_shape={ + "update": self._payload_shape_only(update), + "message": self._payload_shape_only(message), + }, + media=self._business_payload_media_summary(message), + reply_context=self._business_payload_reply_summary(message), + capture_all=self._business_payload_probe_capture_all(), + ) + except Exception as exc: + logger.warning("[Telegram Business Inbox] skipping message store: %s", exc) + + def _record_deleted_business_messages_update(self, update: Any, deleted: Any) -> None: + connection_id = getattr(deleted, "business_connection_id", None) + connection_id = str(connection_id).strip() if connection_id is not None else None + owner_user_chat_id = self._business_owner_for_connection(connection_id) + if not owner_user_chat_id: + logger.warning("[Telegram Business Inbox] skipping delete store: owner mapping unavailable") + return + chat = getattr(deleted, "chat", None) + message_ids = getattr(deleted, "message_ids", None) or [] + try: + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore.for_telegram_user_chat_id(owner_user_chat_id) + store.record_deleted_business_messages( + update_id=getattr(update, "update_id", None), + connection_id=connection_id, + chat_id=getattr(chat, "id", None), + message_ids=message_ids, + ) + store.record_business_payload_probe_event( + update_id=getattr(update, "update_id", None), + update_type="deleted_business_messages", + connection_id=connection_id, + owner_user_chat_id=owner_user_chat_id, + chat_id=getattr(chat, "id", None), + field_availability={ + "update": { + "has_deleted_business_messages": True, + "has_business_message": self._field_present(update, "business_message"), + "has_edited_business_message": self._field_present(update, "edited_business_message"), + }, + "deleted_business_messages": { + "has_chat": chat is not None and not self._is_unset_mock(chat), + "message_ids_count": len(message_ids), + }, + }, + payload_shape={ + "update": self._payload_shape_only(update), + "deleted_business_messages": self._payload_shape_only(deleted), + }, + deleted_message_ids=message_ids, + capture_all=self._business_payload_probe_capture_all(), + ) + except Exception as exc: + logger.warning("[Telegram Business Inbox] skipping delete store: %s", exc) + + async def _handle_business_update(self, update: Any, context: Any) -> bool: + """Persist Telegram Business/Profile Automation updates without replying. + + These updates represent Alen's personal inbox (both incoming messages + and Alen's own replies in third-party chats). They are data for the life + inbox, not direct prompts to the assistant. Returning True tells normal + message handlers to stop before auth/LLM routing. + """ + if not self._is_business_update(update): + return False + if getattr(update, "_hermes_business_handled", False): + return True + try: + setattr(update, "_hermes_business_handled", True) + except Exception: + pass + + business_connection = getattr(update, "business_connection", None) + if business_connection is not None: + self._record_business_connection_update(update, business_connection) + + for update_type, message in self._iter_business_message_payloads(update): + self._record_business_message_update(update, update_type, message) + + deleted = getattr(update, "deleted_business_messages", None) + if deleted is not None: + self._record_deleted_business_messages_update(update, deleted) + return True + async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming text messages. @@ -4749,6 +5270,8 @@ async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAU rapid successive text messages from the same user/chat and aggregate them into a single MessageEvent before dispatching. """ + if await self._handle_business_update(update, context): + return msg = self._effective_update_message(update) if not msg or not msg.text: return @@ -4765,6 +5288,8 @@ async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAU async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming command messages.""" + if await self._handle_business_update(update, context): + return msg = self._effective_update_message(update) if not msg or not msg.text: return @@ -4779,6 +5304,8 @@ async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TY async def _handle_location_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming location/venue pin messages.""" + if await self._handle_business_update(update, context): + return msg = self._effective_update_message(update) if not msg: return @@ -4956,6 +5483,8 @@ def _enqueue_photo_event(self, batch_key: str, event: MessageEvent) -> None: async def _handle_media_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming media messages, downloading images to local cache.""" + if await self._handle_business_update(update, context): + return if not update.message: return if not self._should_process_message(update.message): diff --git a/scripts/telegram_business_payload_probe.py b/scripts/telegram_business_payload_probe.py new file mode 100644 index 000000000000..8fd4d4aafef7 --- /dev/null +++ b/scripts/telegram_business_payload_probe.py @@ -0,0 +1,340 @@ +#!/usr/bin/env python3 +"""Prepare and inspect sanitized Telegram Business payload probes. + +This script is intentionally shape/metadata oriented: it prints the temporary +probe codes that humans should send, but it stores only SHA-256 + length in the +account-scoped life inbox DB. Status output hides raw chat/sender/message ids by +default so it can be pasted into checkpoints without leaking private dialogs. +""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +from collections.abc import Iterable, Mapping +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from gateway.life_inbox_store import ( + BUSINESS_PAYLOAD_PROBE_LANE, + BUSINESS_PAYLOAD_PROBE_SCENARIOS, + LifeInboxStore, + resolve_life_inbox_db_path, +) + +DEFAULT_OWNER_TELEGRAM_ID = "602562" + +_INSTRUCTIONS = { + "S1_contact_inbound": "CONTACT_1 sends this exact code to Alen in Telegram.", + "S2_contact_alen_manual_outbound": "Alen manually sends this exact code to CONTACT_1 in Telegram.", + "S3_known_noncontact_inbound": "KNOWN_NONCONTACT_1 sends this exact code to Alen in Telegram.", + "S4_known_noncontact_alen_manual_outbound": "Alen manually sends this exact code to KNOWN_NONCONTACT_1 in Telegram.", + "S5_new_chat_inbound": "NEW_CHAT_1 sends this exact code to Alen in Telegram for the first observed chat.", + "S6_new_chat_alen_manual_outbound": "Alen manually sends this exact code to NEW_CHAT_1 in Telegram.", +} + + +def _utc_run_id() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + +def _resolve_db_path(args: argparse.Namespace) -> Path: + if args.db: + return Path(args.db).expanduser() + return resolve_life_inbox_db_path(args.owner_telegram_id) + + +def _json_loads(value: str | None, default: Any) -> Any: + if not value: + return default + try: + return json.loads(value) + except json.JSONDecodeError: + return default + + +def _merge_availability(base: Any, incoming: Any) -> Any: + """Merge field-availability JSON by OR-ing booleans and recursing maps.""" + + if isinstance(base, Mapping) and isinstance(incoming, Mapping): + merged: dict[str, Any] = dict(base) + for key, value in incoming.items(): + if key in merged: + merged[key] = _merge_availability(merged[key], value) + else: + merged[key] = value + return merged + if isinstance(base, bool) or isinstance(incoming, bool): + return bool(base) or bool(incoming) + return incoming if incoming not in (None, "", [], {}) else base + + +def _unique_preserving_order(values: Iterable[Any]) -> list[Any]: + seen: set[str] = set() + result: list[Any] = [] + for value in values: + key = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) + if key in seen: + continue + seen.add(key) + result.append(value) + return result + + +def _scenario_payload(run_id: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for scenario in BUSINESS_PAYLOAD_PROBE_SCENARIOS: + scenario_id = scenario["scenario_id"] + rows.append( + { + "scenario_id": scenario_id, + "alias": scenario.get("alias"), + "expected_direction": scenario.get("expected_direction"), + "instruction": _INSTRUCTIONS.get(scenario_id, "Send this exact code in Telegram."), + "probe_text": f"TBP-{run_id}-{scenario_id}", + } + ) + return rows + + +def _cmd_prepare(args: argparse.Namespace) -> int: + db_path = _resolve_db_path(args) + run_id = args.run_id or _utc_run_id() + scenarios = _scenario_payload(run_id) + notes = "Strict Telegram Business Bot API live payload probe; plaintext code not stored in DB." + + store = LifeInboxStore(db_path) + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": row["scenario_id"], + "alias": row["alias"], + "expected_direction": row["expected_direction"], + "probe_text": row["probe_text"], + "notes": notes, + } + for row in scenarios + ], + source_lane=args.source_lane, + ) + + payload = { + "db_path": str(db_path), + "source_lane": args.source_lane, + "run_id": run_id, + "scenarios": scenarios, + "safety": { + "stored_in_db": "sha256+length only; plaintext probe codes are not persisted", + "source_chat_replies": "disabled/proposal-only", + "raw_text_in_logs": "do not log raw private text", + }, + } + _print_payload(payload, args.format) + return 0 + + +def _load_status_rows(db_path: Path, source_lane: str, include_identifiers: bool) -> dict[str, Any]: + if not db_path.exists(): + return {"db_path": str(db_path), "source_lane": source_lane, "exists": False, "scenarios": []} + + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + scenario_rows = conn.execute( + """ + SELECT id, scenario_id, alias, expected_direction, probe_text_len, + probe_text_sha256, status, matched_event_id, created_at, + matched_at, notes + FROM business_payload_probe_scenarios + WHERE source_lane = ? + ORDER BY id + """, + (source_lane,), + ).fetchall() + event_rows = conn.execute( + """ + SELECT id, scenario_id, update_type, direction, message_date, + text_len, raw_text_stored, field_availability_json, + payload_shape_json, media_json, reply_context_json, + deleted_message_ids_json, created_at, + connection_id, chat_id, message_id, sender_id + FROM business_payload_probe_events + WHERE source_lane = ? + ORDER BY id + """, + (source_lane,), + ).fetchall() + + events_by_scenario: dict[str | None, list[sqlite3.Row]] = {} + for event in event_rows: + events_by_scenario.setdefault(event["scenario_id"], []).append(event) + + scenarios: list[dict[str, Any]] = [] + for scenario in scenario_rows: + scenario_events = events_by_scenario.get(scenario["scenario_id"], []) + availability: dict[str, Any] = {} + media_keys: list[str] = [] + reply_present = False + deleted_message_ids_count = 0 + for event in scenario_events: + availability = _merge_availability( + availability, + _json_loads(event["field_availability_json"], {}), + ) + media = _json_loads(event["media_json"], {}) + if isinstance(media, Mapping): + media_keys.extend(str(key) for key in media.keys()) + reply_context = _json_loads(event["reply_context_json"], {}) + if isinstance(reply_context, Mapping) and reply_context.get("present"): + reply_present = True + deleted_ids = _json_loads(event["deleted_message_ids_json"], []) + if isinstance(deleted_ids, list): + deleted_message_ids_count += len(deleted_ids) + + row = { + "scenario_id": scenario["scenario_id"], + "alias": scenario["alias"], + "expected_direction": scenario["expected_direction"], + "status": scenario["status"], + "probe_text_len": scenario["probe_text_len"], + "probe_text_sha256_prefix": str(scenario["probe_text_sha256"] or "")[:16], + "matched": bool(scenario["matched_event_id"]), + "matched_at": scenario["matched_at"], + "event_count": len(scenario_events), + "event_update_types": _unique_preserving_order(event["update_type"] for event in scenario_events), + "event_directions": _unique_preserving_order(event["direction"] for event in scenario_events), + "first_event_at": scenario_events[0]["created_at"] if scenario_events else None, + "last_event_at": scenario_events[-1]["created_at"] if scenario_events else None, + "field_availability": availability, + "media_keys": sorted(set(media_keys)), + "reply_present": reply_present, + "deleted_message_ids_count": deleted_message_ids_count, + "raw_text_stored_count": sum(int(event["raw_text_stored"] or 0) for event in scenario_events), + } + if include_identifiers: + row["event_identifiers"] = [ + { + "id": event["id"], + "connection_id": event["connection_id"], + "chat_id": event["chat_id"], + "message_id": event["message_id"], + "sender_id": event["sender_id"], + "message_date": event["message_date"], + } + for event in scenario_events + ] + scenarios.append(row) + + unmatched_events = events_by_scenario.get(None, []) + payload: dict[str, Any] = { + "db_path": str(db_path), + "source_lane": source_lane, + "exists": True, + "scenario_count": len(scenarios), + "matched_count": sum(1 for row in scenarios if row["matched"]), + "pending_count": sum(1 for row in scenarios if row["status"] == "pending"), + "unmatched_capture_all_event_count": len(unmatched_events), + "scenarios": scenarios, + } + if include_identifiers and unmatched_events: + payload["unmatched_capture_all_events"] = [ + { + "id": event["id"], + "update_type": event["update_type"], + "direction": event["direction"], + "chat_id": event["chat_id"], + "message_id": event["message_id"], + "sender_id": event["sender_id"], + "created_at": event["created_at"], + } + for event in unmatched_events + ] + return payload + + +def _cmd_status(args: argparse.Namespace) -> int: + db_path = _resolve_db_path(args) + payload = _load_status_rows(db_path, args.source_lane, args.include_identifiers) + _print_payload(payload, args.format) + return 0 + + +def _print_payload(payload: dict[str, Any], output_format: str) -> None: + if output_format == "json": + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return + + if "run_id" in payload: + print(f"Telegram Business payload probe run: {payload['run_id']}") + print(f"DB: {payload['db_path']}") + for row in payload["scenarios"]: + print(f"- {row['scenario_id']} ({row['alias']}): {row['instruction']}") + print(f" code: {row['probe_text']}") + print("Safety: DB stores hash+length only; no source-chat auto-replies.") + return + + print(f"Telegram Business payload probe status: {payload['source_lane']}") + print(f"DB: {payload['db_path']} exists={payload.get('exists')}") + print(f"matched={payload.get('matched_count', 0)} pending={payload.get('pending_count', 0)}") + for row in payload.get("scenarios", []): + updates = ",".join(row["event_update_types"]) or "-" + directions = ",".join(row["event_directions"]) or "-" + print( + f"- {row['scenario_id']}: {row['status']} " + f"events={row['event_count']} updates={updates} directions={directions}" + ) + + +def _add_common_args(parser: argparse.ArgumentParser, *, suppress_defaults: bool = False) -> None: + default: Any = argparse.SUPPRESS if suppress_defaults else None + parser.add_argument( + "--db", + default=default, + help="Path to account-scoped life_inbox.sqlite", + ) + parser.add_argument( + "--owner-telegram-id", + default=argparse.SUPPRESS if suppress_defaults else DEFAULT_OWNER_TELEGRAM_ID, + help="Owner numeric Telegram ID used when --db is omitted (default: 602562)", + ) + parser.add_argument( + "--source-lane", + default=argparse.SUPPRESS if suppress_defaults else BUSINESS_PAYLOAD_PROBE_LANE, + ) + parser.add_argument( + "--format", + choices=("text", "json"), + default=argparse.SUPPRESS if suppress_defaults else "text", + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + _add_common_args(parser) + + sub = parser.add_subparsers(dest="command", required=True) + prepare = sub.add_parser("prepare", help="Create a six-scenario run sheet and register hashes") + _add_common_args(prepare, suppress_defaults=True) + prepare.add_argument("--run-id", help="Stable run id for code generation, e.g. 20260520T090000Z") + prepare.set_defaults(func=_cmd_prepare) + + status = sub.add_parser("status", help="Summarize matched payload-probe events") + _add_common_args(status, suppress_defaults=True) + status.add_argument( + "--include-identifiers", + action="store_true", + help="Include private chat/sender/message ids in status output (off by default)", + ) + status.set_defaults(func=_cmd_status) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + return int(args.func(args)) + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/gateway/test_life_inbox_store.py b/tests/gateway/test_life_inbox_store.py new file mode 100644 index 000000000000..c42852846496 --- /dev/null +++ b/tests/gateway/test_life_inbox_store.py @@ -0,0 +1,464 @@ +import json +import os +import sqlite3 +import stat +from pathlib import Path + +import pytest + +from gateway.life_inbox_store import ( + BUSINESS_PAYLOAD_PROBE_LANE, + BUSINESS_PAYLOAD_PROBE_SCENARIOS, + LifeInboxStore, + detect_candidate_reasons, + resolve_life_inbox_db_path, +) + + +def _write_accounts_registry(life_home: Path) -> None: + profile_rel = "accounts/telegram-602562/profile.json" + profile_path = life_home / profile_rel + profile_path.parent.mkdir(parents=True, exist_ok=True) + profile_path.write_text("{}\n") + (life_home / "accounts.json").write_text( + json.dumps( + { + "version": 1, + "accounts": { + "telegram:602562": { + "display_name": "Alen", + "telegram_user_id": "602562", + "life_profile": profile_rel, + } + }, + } + ) + + "\n" + ) + + +def test_resolve_life_inbox_db_path_uses_account_scoped_profile_dir(tmp_path): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + + db_path = resolve_life_inbox_db_path("602562", life_home=life_home) + + assert db_path == life_home / "accounts/telegram-602562/life_inbox.sqlite" + + +def test_record_business_connection_upserts_without_raw_message_storage(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + + store.record_business_connection( + update_id=101, + connection_id="conn-1", + is_enabled=True, + user_chat_id="602562", + user_id="602562", + username="oldman", + full_name="Alen", + rights={"can_read_messages": True}, + ) + store.record_business_connection( + update_id=102, + connection_id="conn-1", + is_enabled=False, + user_chat_id="602562", + user_id="602562", + username="oldman", + full_name="Alen Updated", + rights={"can_read_messages": False}, + ) + + with sqlite3.connect(store.db_path) as conn: + row = conn.execute( + "SELECT connection_id, is_enabled, user_chat_id, full_name, rights_json FROM business_connections" + ).fetchone() + + assert row[0] == "conn-1" + assert row[1] == 0 + assert row[2] == "602562" + assert row[3] == "Alen Updated" + assert json.loads(row[4]) == {"can_read_messages": False} + + +def test_record_business_message_dedupes_metadata_and_archives_raw_text(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + + message_pk_1 = store.record_business_message( + update_id=201, + update_type="business_message", + connection_id="conn-1", + chat_id="1566649385", + chat_type="private", + chat_name="BaliRadar", + message_id="1408996", + sender_id="1566649385", + sender_name="BaliRadar", + text="завтра в 15 созвон по Cockpit", + message_date="2026-05-19T22:09:49+00:00", + ) + message_pk_2 = store.record_business_message( + update_id=202, + update_type="edited_business_message", + connection_id="conn-1", + chat_id="1566649385", + chat_type="private", + chat_name="BaliRadar", + message_id="1408996", + sender_id="1566649385", + sender_name="BaliRadar", + text="завтра в 15 созвон по Cockpit", + message_date="2026-05-19T22:10:00+00:00", + ) + + assert message_pk_1 == message_pk_2 + + with sqlite3.connect(store.db_path) as conn: + row = conn.execute( + """ + SELECT update_id, update_type, chat_id, chat_name, text_len, text_sha256, + text_preview, raw_text_stored, candidate_reasons_json + FROM business_messages + """ + ).fetchone() + chat_rule = conn.execute( + "SELECT platform, chat_id, rule_mode FROM chat_rules" + ).fetchone() + text_row = conn.execute( + "SELECT text FROM business_message_text WHERE business_message_id = ?", + (message_pk_1,), + ).fetchone() + + assert row[0] == 202 + assert row[1] == "edited_business_message" + assert row[2] == "1566649385" + assert row[3] == "BaliRadar" + assert row[4] == len("завтра в 15 созвон по Cockpit") + assert len(row[5]) == 64 + assert row[6] is None + assert row[7] == 1 + assert set(json.loads(row[8])) >= {"meeting", "time_reference"} + assert chat_rule == ("telegram_business", "1566649385", "full_rag_selected") + assert text_row == ("завтра в 15 созвон по Cockpit",) + + if os.name != "nt": + assert stat.S_IMODE(store.db_path.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(store.db_path.stat().st_mode) == 0o600 + + +def test_record_business_message_archives_raw_text_and_business_users(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + + message_pk = store.record_business_message( + update_id=301, + update_type="business_message", + connection_id="conn-archive", + chat_id="1566649385", + chat_type="private", + chat_name="BaliRadar", + chat_username="RadarAdmin", + message_id="1409001", + sender_id="1566649385", + sender_name="BaliRadar", + sender_username="RadarAdmin", + sender_is_bot=False, + sender_language_code="en", + text="private archive text for analyzer", + message_date="2026-05-20T09:10:00+00:00", + ) + + # Outgoing messages should still retain the private-chat counterpart as a user, + # not only Alen as the sender. + store.record_business_message( + update_id=302, + update_type="business_message", + connection_id="conn-archive", + chat_id="1566649385", + chat_type="private", + chat_name="BaliRadar", + chat_username="RadarAdmin", + message_id="1409002", + sender_id="602562", + sender_name="Alen", + sender_username="oldman", + sender_is_bot=False, + sender_language_code="ru", + text="alen outgoing archive text", + message_date="2026-05-20T09:11:00+00:00", + ) + + with sqlite3.connect(store.db_path) as conn: + message_row = conn.execute( + """ + SELECT raw_text_stored, text_len + FROM business_messages + WHERE id = ? + """, + (message_pk,), + ).fetchone() + text_row = conn.execute( + """ + SELECT text + FROM business_message_text + WHERE business_message_id = ? + """, + (message_pk,), + ).fetchone() + users = conn.execute( + """ + SELECT user_id, username, full_name, is_bot, language_code + FROM business_users + ORDER BY user_id + """ + ).fetchall() + participants = conn.execute( + """ + SELECT chat_id, user_id + FROM chat_participants + ORDER BY user_id + """ + ).fetchall() + + assert message_row == (1, len("private archive text for analyzer")) + assert text_row == ("private archive text for analyzer",) + assert users == [ + ("1566649385", "RadarAdmin", "BaliRadar", 0, "en"), + ("602562", "oldman", "Alen", 0, "ru"), + ] + assert participants == [ + ("1566649385", "1566649385"), + ("1566649385", "602562"), + ] + + +def test_record_business_message_rejects_missing_identity_fields(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + + with pytest.raises(ValueError): + store.record_business_message( + update_id=201, + update_type="business_message", + connection_id="conn-1", + chat_id=None, + chat_type="private", + chat_name="BaliRadar", + message_id="1408996", + sender_id="1566649385", + sender_name="BaliRadar", + text="test", + message_date="2026-05-19T22:09:49+00:00", + ) + + with pytest.raises(ValueError): + store.record_business_message( + update_id=201, + update_type="business_message", + connection_id="conn-1", + chat_id="1566649385", + chat_type="private", + chat_name="BaliRadar", + message_id=None, + sender_id="1566649385", + sender_name="BaliRadar", + text="test", + message_date="2026-05-19T22:09:49+00:00", + ) + + +def test_detect_candidate_reasons_covers_life_inbox_keywords(): + assert set(detect_candidate_reasons("завтра в 15 созвон, напомни follow up")) >= { + "meeting", + "time_reference", + "reminder", + "follow_up", + } + assert detect_candidate_reasons("просто болтаем ни о чём") == [] + + +def test_prepare_payload_probe_scenarios_stores_hashes_not_probe_text(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + probe_text = "TBP-20260520-S1-contact-inbound" + + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + "probe_text": probe_text, + } + ] + ) + + with sqlite3.connect(store.db_path) as conn: + row = conn.execute( + """ + SELECT source_lane, scenario_id, alias, expected_direction, + probe_text_len, probe_text_sha256, status + FROM business_payload_probe_scenarios + """ + ).fetchone() + + assert row[0] == BUSINESS_PAYLOAD_PROBE_LANE + assert row[1] == "S1_contact_inbound" + assert row[2] == "CONTACT_1" + assert row[3] == "incoming_to_owner" + assert row[4] == len(probe_text) + assert len(row[5]) == 64 + assert row[6] == "pending" + assert probe_text not in store.db_path.read_bytes().decode("utf-8", errors="ignore") + + +def test_record_payload_probe_event_matches_scenario_and_keeps_shape_only(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + probe_text = "TBP-20260520-S2-owner-outbound" + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S2_contact_alen_manual_outbound", + "alias": "CONTACT_1", + "expected_direction": "outgoing_from_owner", + "probe_text": probe_text, + } + ] + ) + + event_id = store.record_business_payload_probe_event( + update_id=701, + update_type="business_message", + connection_id="conn-1", + owner_user_chat_id="602562", + chat_id="1566649385", + message_id="1409010", + sender_id="602562", + text=probe_text, + message_date="2026-05-20T08:40:00+00:00", + field_availability={"message": {"has_text": True, "has_reply_to_message": False}}, + payload_shape={"message": {"text": {"type": "str"}, "from": {"id": {"type": "int"}}}}, + ) + + assert event_id is not None + with sqlite3.connect(store.db_path) as conn: + event_row = conn.execute( + """ + SELECT source_lane, scenario_id, direction, text_len, raw_text_stored, + field_availability_json, payload_shape_json + FROM business_payload_probe_events + """ + ).fetchone() + scenario_row = conn.execute( + """ + SELECT status, matched_event_id, matched_at + FROM business_payload_probe_scenarios + WHERE scenario_id = 'S2_contact_alen_manual_outbound' + """ + ).fetchone() + + assert event_row[0] == BUSINESS_PAYLOAD_PROBE_LANE + assert event_row[1] == "S2_contact_alen_manual_outbound" + assert event_row[2] == "outgoing_from_owner" + assert event_row[3] == len(probe_text) + assert event_row[4] == 0 + assert json.loads(event_row[5])["message"]["has_text"] is True + assert json.loads(event_row[6])["message"]["text"]["type"] == "str" + assert scenario_row[0] == "matched" + assert scenario_row[1] == event_id + assert scenario_row[2] is not None + assert probe_text not in store.db_path.read_bytes().decode("utf-8", errors="ignore") + + +def test_payload_probe_event_can_match_followup_edit_by_message_identity(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + probe_text = "TBP-20260520-S1-edit-followup" + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + "probe_text": probe_text, + } + ] + ) + first_event_id = store.record_business_payload_probe_event( + update_id=801, + update_type="business_message", + connection_id="conn-1", + owner_user_chat_id="602562", + chat_id="1566649385", + message_id="1409011", + sender_id="1566649385", + text=probe_text, + message_date="2026-05-20T08:45:00+00:00", + field_availability={}, + payload_shape={}, + ) + + edited_event_id = store.record_business_payload_probe_event( + update_id=802, + update_type="edited_business_message", + connection_id="conn-1", + owner_user_chat_id="602562", + chat_id="1566649385", + message_id="1409011", + sender_id="1566649385", + text="edited private text not registered as a probe code", + message_date="2026-05-20T08:46:00+00:00", + field_availability={"message": {"has_edit_date": True}}, + payload_shape={"message": {"edit_date": {"type": "datetime"}}}, + ) + + assert first_event_id is not None + assert edited_event_id is not None + with sqlite3.connect(store.db_path) as conn: + rows = conn.execute( + "SELECT update_type, scenario_id FROM business_payload_probe_events ORDER BY id" + ).fetchall() + + assert rows == [ + ("business_message", "S1_contact_inbound"), + ("edited_business_message", "S1_contact_inbound"), + ] + assert "private text" not in store.db_path.read_bytes().decode("utf-8", errors="ignore") + + +def test_payload_probe_event_without_scenario_is_skipped_unless_capture_all(tmp_path): + store = LifeInboxStore(tmp_path / "life_inbox.sqlite") + + skipped = store.record_business_payload_probe_event( + update_id=901, + update_type="business_message", + connection_id="conn-1", + owner_user_chat_id="602562", + chat_id="1566649385", + message_id="1409012", + sender_id="1566649385", + text="private unmatched text", + message_date="2026-05-20T08:50:00+00:00", + field_availability={}, + payload_shape={}, + ) + captured = store.record_business_payload_probe_event( + update_id=902, + update_type="business_message", + connection_id="conn-1", + owner_user_chat_id="602562", + chat_id="1566649385", + message_id="1409013", + sender_id="1566649385", + text="private unmatched text", + message_date="2026-05-20T08:50:10+00:00", + field_availability={"message": {"has_photo": True}}, + payload_shape={"message": {"photo": {"type": "list", "length": 1}}}, + capture_all=True, + ) + + assert skipped is None + assert captured is not None + with sqlite3.connect(store.db_path) as conn: + rows = conn.execute("SELECT scenario_id, text_len FROM business_payload_probe_events").fetchall() + + assert rows == [(None, len("private unmatched text"))] + assert "private unmatched text" not in store.db_path.read_bytes().decode("utf-8", errors="ignore") + assert len(BUSINESS_PAYLOAD_PROBE_SCENARIOS) == 6 diff --git a/tests/gateway/test_telegram_business_inbox.py b/tests/gateway/test_telegram_business_inbox.py new file mode 100644 index 000000000000..459aca977b3f --- /dev/null +++ b/tests/gateway/test_telegram_business_inbox.py @@ -0,0 +1,682 @@ +import json +import logging +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from gateway.config import PlatformConfig +from gateway.platforms.telegram import TelegramAdapter + + +def _write_accounts_registry(life_home: Path) -> None: + profile_rel = "accounts/telegram-602562/profile.json" + profile_path = life_home / profile_rel + profile_path.parent.mkdir(parents=True, exist_ok=True) + profile_path.write_text("{}\n") + (life_home / "accounts.json").write_text( + json.dumps( + { + "version": 1, + "accounts": { + "telegram:602562": { + "display_name": "Alen", + "telegram_user_id": "602562", + "life_profile": profile_rel, + } + }, + } + ) + + "\n" + ) + + +@pytest.mark.asyncio +async def test_business_update_handler_persists_metadata_without_auto_reply(tmp_path, monkeypatch, caplog): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + caplog.set_level(logging.WARNING, logger="gateway.platforms.telegram") + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + handled_messages = [] + + async def fake_handle_message(event): + handled_messages.append(event) + + adapter.handle_message = fake_handle_message + + rights = SimpleNamespace(to_dict=lambda: {"can_read_messages": True}) + user = SimpleNamespace(id=602562, username="oldman", full_name="Alen") + business_connection = SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=user, + rights=rights, + ) + await adapter._handle_business_update( + SimpleNamespace( + update_id=301, + business_connection=business_connection, + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + chat = SimpleNamespace(id=1566649385, type="private", title=None, full_name="BaliRadar") + sender = SimpleNamespace(id=1566649385, full_name="BaliRadar") + business_message = SimpleNamespace( + business_connection_id="conn-1", + chat=chat, + from_user=sender, + text="завтра в 15 созвон", + caption=None, + message_id=1408996, + date=datetime(2026, 5, 19, 22, 9, 49, tzinfo=timezone.utc), + ) + await adapter._handle_business_update( + SimpleNamespace( + update_id=302, + business_connection=None, + business_message=business_message, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + assert handled_messages == [] + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + connection_row = conn.execute( + "SELECT connection_id, user_chat_id, username FROM business_connections" + ).fetchone() + message_row = conn.execute( + """ + SELECT chat_id, message_id, text_len, text_preview, raw_text_stored, candidate_reasons_json + FROM business_messages + """ + ).fetchone() + + text_row = conn.execute( + """ + SELECT text + FROM business_message_text + JOIN business_messages ON business_messages.id = business_message_text.business_message_id + WHERE business_messages.chat_id = '1566649385' + """ + ).fetchone() + user_row = conn.execute( + "SELECT user_id, full_name FROM business_users WHERE user_id = '1566649385'" + ).fetchone() + + assert connection_row == ("conn-1", "602562", "oldman") + assert message_row[0] == "1566649385" + assert message_row[1] == "1408996" + assert message_row[2] == len("завтра в 15 созвон") + assert message_row[3] is None + assert message_row[4] == 1 + assert set(json.loads(message_row[5])) >= {"meeting", "time_reference"} + assert text_row == ("завтра в 15 созвон",) + assert user_row == ("1566649385", "BaliRadar") + + warning_text = "\n".join(record.getMessage() for record in caplog.records) + assert "chat_id=1566649385" not in warning_text + assert "text_sha256" not in warning_text + assert "BaliRadar" not in warning_text + + +@pytest.mark.asyncio +async def test_business_message_storage_survives_adapter_restart(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + first_adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + rights = SimpleNamespace(to_dict=lambda: {"can_read_messages": True}) + user = SimpleNamespace(id=602562, username="oldman", full_name="Alen") + await first_adapter._handle_business_update( + SimpleNamespace( + update_id=401, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=user, + rights=rights, + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + restarted_adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + assert restarted_adapter._business_connection_user_chat_ids == {} + await restarted_adapter._handle_business_update( + SimpleNamespace( + update_id=402, + business_connection=None, + business_message=SimpleNamespace( + business_connection_id="conn-1", + chat=SimpleNamespace(id=1566649385, type="private", title=None, full_name="BaliRadar"), + from_user=SimpleNamespace(id=1566649385, full_name="BaliRadar"), + text="завтра в 15 созвон", + caption=None, + message_id=1408997, + date=datetime(2026, 5, 19, 22, 10, 49, tzinfo=timezone.utc), + ), + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + count = conn.execute("SELECT COUNT(*) FROM business_messages").fetchone()[0] + + assert count == 1 + assert restarted_adapter._business_connection_user_chat_ids == {"conn-1": "602562"} + + +@pytest.mark.asyncio +async def test_text_handler_routes_business_connection_messages_to_inbox_not_agent(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + handled_messages = [] + enqueued_messages = [] + + async def fake_handle_message(event): + handled_messages.append(event) + + def fake_enqueue_event(event): + enqueued_messages.append(event) + + adapter.handle_message = fake_handle_message + adapter._enqueue_text_event = fake_enqueue_event + + await adapter._handle_business_update( + SimpleNamespace( + update_id=501, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + rights=SimpleNamespace(to_dict=lambda: {"can_read_messages": True}), + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + # PTB/Bot API can expose Business/Profile Automation traffic through the + # regular text MessageHandler path while preserving Message.business_connection_id. + # Owner-sent replies must be stored passively, not treated as authorized prompts. + chat = SimpleNamespace(id=1566649385, type="private", title=None, full_name="BaliRadar") + owner_sender = SimpleNamespace(id=602562, username="oldman", full_name="Alen") + message = SimpleNamespace( + business_connection_id="conn-1", + chat=chat, + from_user=owner_sender, + text="testmsgtoAdmin", + caption=None, + message_id=1409001, + date=datetime(2026, 5, 20, 7, 13, 40, tzinfo=timezone.utc), + ) + + await adapter._handle_text_message( + SimpleNamespace( + update_id=502, + message=message, + effective_message=message, + business_connection=None, + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + assert handled_messages == [] + assert enqueued_messages == [] + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + row = conn.execute( + "SELECT chat_id, sender_id, message_id, text_len FROM business_messages WHERE message_id = '1409001'" + ).fetchone() + + assert row == ("1566649385", "602562", "1409001", len("testmsgtoAdmin")) + + +@pytest.mark.asyncio +async def test_business_inbox_skips_copies_of_the_hermes_bot_chat(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + adapter._bot = SimpleNamespace(id=796330107, username="alenrbot") + + await adapter._handle_business_update( + SimpleNamespace( + update_id=601, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + rights=SimpleNamespace(to_dict=lambda: {"can_read_messages": True}), + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + # Business/Profile Automation mirrors the owner's DM with this same bot as + # business_message updates. Those should not enter the life inbox. + bot_chat = SimpleNamespace(id=796330107, type="private", full_name="Птолемей | Ассистент Алена") + owner_message_to_bot = SimpleNamespace( + business_connection_id="conn-1", + chat=bot_chat, + from_user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + text="проверяй", + caption=None, + message_id=1409673, + date=datetime(2026, 5, 20, 7, 42, 13, tzinfo=timezone.utc), + ) + bot_reply_copy = SimpleNamespace( + business_connection_id="conn-1", + chat=bot_chat, + from_user=SimpleNamespace(id=796330107, username="alenrbot", full_name="Птолемей | Ассистент Алена"), + text="assistant response", + caption=None, + message_id=1409674, + date=datetime(2026, 5, 20, 7, 42, 20, tzinfo=timezone.utc), + ) + + for update_id, message in ((602, owner_message_to_bot), (603, bot_reply_copy)): + assert await adapter._handle_business_update( + SimpleNamespace( + update_id=update_id, + business_connection=None, + business_message=message, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) is True + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + count = conn.execute("SELECT COUNT(*) FROM business_messages").fetchone()[0] + + assert count == 0 + + +@pytest.mark.asyncio +async def test_send_refuses_own_bot_chat_without_calling_telegram(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + send_calls = [] + + async def fake_send_message(**kwargs): + send_calls.append(kwargs) + raise AssertionError("send_message must not be called for own bot chat") + + adapter._bot = SimpleNamespace(id=796330107, username="alenrbot", send_message=fake_send_message) + + result = await adapter.send(chat_id="796330107", content="should not send") + + assert result.success is True + assert result.message_id is None + assert send_calls == [] + + +@pytest.mark.asyncio +async def test_business_update_handler_records_payload_probe_shape_without_raw_text(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore(life_home / "accounts/telegram-602562/life_inbox.sqlite") + probe_text = "TBP-20260520-S1-contact-inbound" + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + "probe_text": probe_text, + } + ] + ) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + handled_messages = [] + + async def fake_handle_message(event): + handled_messages.append(event) + + adapter.handle_message = fake_handle_message + + await adapter._handle_business_update( + SimpleNamespace( + update_id=701, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + rights=SimpleNamespace(to_dict=lambda: {"can_read_messages": True}), + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + reply_to_message = SimpleNamespace(message_id=1409000) + chat = SimpleNamespace(id=1566649385, type="private", title=None, full_name="BaliRadar") + sender = SimpleNamespace(id=1566649385, full_name="BaliRadar", is_bot=False) + business_message = SimpleNamespace( + business_connection_id="conn-1", + chat=chat, + from_user=sender, + text=probe_text, + caption=None, + message_id=1409010, + date=datetime(2026, 5, 20, 8, 40, 0, tzinfo=timezone.utc), + reply_to_message=reply_to_message, + photo=[SimpleNamespace(file_id="photo-file-id")], + to_dict=lambda: { + "message_id": 1409010, + "date": 1779266400, + "chat": {"id": 1566649385, "type": "private", "first_name": "PrivateName"}, + "from": {"id": 1566649385, "first_name": "PrivateName", "is_bot": False}, + "text": probe_text, + "reply_to_message": {"message_id": 1409000, "text": "do not store me"}, + "photo": [{"file_id": "photo-file-id", "file_unique_id": "unique"}], + "business_connection_id": "conn-1", + }, + ) + + await adapter._handle_business_update( + SimpleNamespace( + update_id=702, + business_connection=None, + business_message=business_message, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + assert handled_messages == [] + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT scenario_id, direction, raw_text_stored, + field_availability_json, payload_shape_json + FROM business_payload_probe_events + """ + ).fetchone() + + assert row[0] == "S1_contact_inbound" + assert row[1] == "incoming_to_owner" + assert row[2] == 0 + fields = json.loads(row[3]) + assert fields["message"]["has_text"] is True + assert fields["message"]["has_reply_to_message"] is True + assert fields["message"]["has_photo"] is True + shape = json.loads(row[4]) + assert shape["message"]["keys"]["text"]["type"] == "str" + raw_db = db_path.read_bytes().decode("utf-8", errors="ignore") + assert probe_text not in raw_db + assert "PrivateName" not in raw_db + assert "do not store me" not in raw_db + + +@pytest.mark.asyncio +async def test_business_update_deduplicates_business_message_and_effective_message_probe_events(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore(life_home / "accounts/telegram-602562/life_inbox.sqlite") + probe_text = "TBP-20260520-S1-dedupe" + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + "probe_text": probe_text, + } + ] + ) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + await adapter._handle_business_update( + SimpleNamespace( + update_id=901, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + rights=SimpleNamespace(to_dict=lambda: {"can_read_messages": True}), + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + chat = SimpleNamespace(id=1566649385, type="private", title=None, full_name="BaliRadar") + sender = SimpleNamespace(id=1566649385, full_name="BaliRadar", is_bot=False) + + def make_message(): + return SimpleNamespace( + business_connection_id="conn-1", + chat=chat, + from_user=sender, + text=probe_text, + caption=None, + message_id=1409099, + date=datetime(2026, 5, 20, 8, 55, 0, tzinfo=timezone.utc), + ) + + await adapter._handle_business_update( + SimpleNamespace( + update_id=902, + business_connection=None, + business_message=make_message(), + # PTB can expose the same Business message through effective_message + # as a distinct Python object. It must still be handled once. + effective_message=make_message(), + message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + probe_count = conn.execute("SELECT COUNT(*) FROM business_payload_probe_events").fetchone()[0] + message_count = conn.execute("SELECT COUNT(*) FROM business_messages").fetchone()[0] + + assert probe_count == 1 + assert message_count == 1 + + +@pytest.mark.asyncio +async def test_business_deleted_update_records_probe_event_for_prior_scenario(tmp_path, monkeypatch): + life_home = tmp_path / ".hermes-life" + _write_accounts_registry(life_home) + monkeypatch.setenv("HERMES_LIFE_HOME", str(life_home)) + + from gateway.life_inbox_store import LifeInboxStore + + store = LifeInboxStore(life_home / "accounts/telegram-602562/life_inbox.sqlite") + probe_text = "TBP-20260520-S5-new-chat-delete" + store.prepare_business_payload_probe_scenarios( + [ + { + "scenario_id": "S5_new_chat_inbound", + "alias": "NEW_CHAT_1", + "expected_direction": "incoming_to_owner", + "probe_text": probe_text, + } + ] + ) + + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + await adapter._handle_business_update( + SimpleNamespace( + update_id=801, + business_connection=SimpleNamespace( + id="conn-1", + is_enabled=True, + user_chat_id=602562, + user=SimpleNamespace(id=602562, username="oldman", full_name="Alen"), + rights=SimpleNamespace(to_dict=lambda: {"can_read_messages": True}), + ), + business_message=None, + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + await adapter._handle_business_update( + SimpleNamespace( + update_id=802, + business_connection=None, + business_message=SimpleNamespace( + business_connection_id="conn-1", + chat=SimpleNamespace(id=999001, type="private", full_name="New Chat"), + from_user=SimpleNamespace(id=999001, full_name="New Chat"), + text=probe_text, + caption=None, + message_id=1409020, + date=datetime(2026, 5, 20, 8, 45, 0, tzinfo=timezone.utc), + ), + edited_business_message=None, + deleted_business_messages=None, + ), + None, + ) + await adapter._handle_business_update( + SimpleNamespace( + update_id=803, + business_connection=None, + business_message=None, + edited_business_message=None, + deleted_business_messages=SimpleNamespace( + business_connection_id="conn-1", + chat=SimpleNamespace(id=999001, type="private"), + message_ids=[1409020], + ), + ), + None, + ) + + db_path = life_home / "accounts/telegram-602562/life_inbox.sqlite" + with sqlite3.connect(db_path) as conn: + rows = conn.execute( + """ + SELECT update_type, scenario_id, deleted_message_ids_json + FROM business_payload_probe_events ORDER BY id + """ + ).fetchall() + + assert rows == [ + ("business_message", "S5_new_chat_inbound", "[]"), + ("deleted_business_messages", "S5_new_chat_inbound", '["1409020"]'), + ] + + +def test_payload_shape_only_sanitizes_ptb_to_dict_values(): + message = SimpleNamespace( + to_dict=lambda: { + "message_id": 1, + "text": "private raw text", + "from": {"id": 1566649385, "first_name": "PrivateName"}, + "entities": [{"type": "url", "offset": 0, "length": 10}], + } + ) + + shape = TelegramAdapter._payload_shape_only(message) + + assert shape["keys"]["text"] == {"type": "str"} + assert shape["keys"]["from"]["keys"]["first_name"] == {"type": "str"} + assert "private raw text" not in json.dumps(shape) + assert "PrivateName" not in json.dumps(shape) + + +def test_business_payload_field_availability_treats_empty_ptb_collections_as_absent(): + message = SimpleNamespace( + chat=SimpleNamespace(type="private", username=None, title=None, full_name="Name"), + from_user=SimpleNamespace(id=1, username=None, is_bot=False), + text="probe", + caption=None, + business_connection_id="conn-1", + entities=(), + caption_entities=(), + photo=(), + document=None, + reply_to_message=None, + ) + update = SimpleNamespace( + business_message=message, + edited_business_message=None, + deleted_business_messages=None, + message=None, + effective_message=message, + business_connection=None, + ) + + fields = TelegramAdapter._business_payload_field_availability(update, message) + media = TelegramAdapter._business_payload_media_summary(message) + + assert fields["message"]["has_text"] is True + assert fields["message"]["has_entities"] is False + assert fields["message"]["has_caption_entities"] is False + assert fields["message"]["has_photo"] is False + assert media == {} + + +def test_should_process_message_ignores_messages_from_own_bot(): + adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token")) + adapter._bot = SimpleNamespace(id=796330107, username="alenrbot") + message = SimpleNamespace( + chat=SimpleNamespace(id=796330107, type="private"), + from_user=SimpleNamespace(id=796330107, is_bot=True, username="alenrbot"), + text="loop bait", + caption=None, + ) + + assert adapter._should_process_message(message) is False diff --git a/tests/gateway/test_telegram_business_payload_probe_script.py b/tests/gateway/test_telegram_business_payload_probe_script.py new file mode 100644 index 000000000000..9a58a34cdc4d --- /dev/null +++ b/tests/gateway/test_telegram_business_payload_probe_script.py @@ -0,0 +1,92 @@ +import json +import sqlite3 + +from gateway.life_inbox_store import LifeInboxStore +from scripts import telegram_business_payload_probe as probe + + +def test_prepare_run_sheet_registers_probe_hashes_without_plaintext(tmp_path, capsys): + db_path = tmp_path / "life_inbox.sqlite" + run_id = "20260520T090000Z" + + exit_code = probe.main([ + "prepare", + "--db", + str(db_path), + "--run-id", + run_id, + "--format", + "json", + ]) + + assert exit_code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["source_lane"] == "business_bot_probe" + assert payload["run_id"] == run_id + assert len(payload["scenarios"]) == 6 + assert payload["scenarios"][0] == { + "scenario_id": "S1_contact_inbound", + "alias": "CONTACT_1", + "expected_direction": "incoming_to_owner", + "instruction": "CONTACT_1 sends this exact code to Alen in Telegram.", + "probe_text": "TBP-20260520T090000Z-S1_contact_inbound", + } + + with sqlite3.connect(db_path) as conn: + rows = conn.execute( + """ + SELECT scenario_id, probe_text_len, probe_text_sha256, status + FROM business_payload_probe_scenarios + ORDER BY id + """ + ).fetchall() + + assert len(rows) == 6 + assert rows[0][0] == "S1_contact_inbound" + assert rows[0][1] == len("TBP-20260520T090000Z-S1_contact_inbound") + assert len(rows[0][2]) == 64 + assert rows[0][3] == "pending" + assert f"TBP-{run_id}" not in db_path.read_bytes().decode("utf-8", errors="ignore") + + +def test_status_summarizes_probe_events_without_chat_or_sender_ids_by_default(tmp_path, capsys): + db_path = tmp_path / "life_inbox.sqlite" + run_id = "20260520T091500Z" + probe_text = "TBP-20260520T091500Z-S2_contact_alen_manual_outbound" + + probe.main(["prepare", "--db", str(db_path), "--run-id", run_id, "--format", "json"]) + capsys.readouterr() + + store = LifeInboxStore(db_path) + store.record_business_payload_probe_event( + update_id=42, + update_type="business_message", + connection_id="private-connection-id", + owner_user_chat_id="602562", + chat_id="private-chat-id", + message_id="private-message-id", + sender_id="602562", + text=probe_text, + message_date="2026-05-20T09:15:00+00:00", + field_availability={"message": {"has_text": True, "has_reply_to_message": False}}, + payload_shape={"message": {"keys": {"text": {"type": "str"}}}}, + media={}, + reply_context={}, + ) + + exit_code = probe.main(["status", "--db", str(db_path), "--format", "json"]) + + assert exit_code == 0 + output = capsys.readouterr().out + assert "private-chat-id" not in output + assert "private-message-id" not in output + assert "private-connection-id" not in output + payload = json.loads(output) + scenario = next(row for row in payload["scenarios"] if row["scenario_id"] == "S2_contact_alen_manual_outbound") + assert scenario["status"] == "matched" + assert scenario["event_count"] == 1 + assert scenario["event_update_types"] == ["business_message"] + assert scenario["event_directions"] == ["outgoing_from_owner"] + assert scenario["field_availability"]["message"]["has_text"] is True + assert scenario["media_keys"] == [] + assert scenario["reply_present"] is False