From 83f67367f80bcd48aaa734fbb14907f8954b3f85 Mon Sep 17 00:00:00 2001 From: "Andrex Ibiza, MBA" Date: Fri, 28 Aug 2026 08:04:26 -0500 Subject: [PATCH] feat(commands): establish Telegram projection contract Refs #96692 --- hermes_cli/telegram_command_normalization.py | 141 +++++ hermes_cli/telegram_command_projection.py | 527 ++++++++++++++++++ hermes_cli/telegram_menu_reconciliation.py | 195 +++++++ .../test_telegram_command_normalization.py | 140 +++++ .../test_telegram_command_projection.py | 231 ++++++++ .../test_telegram_menu_reconciliation.py | 192 +++++++ 6 files changed, 1426 insertions(+) create mode 100644 hermes_cli/telegram_command_normalization.py create mode 100644 hermes_cli/telegram_command_projection.py create mode 100644 hermes_cli/telegram_menu_reconciliation.py create mode 100644 tests/hermes_cli/test_telegram_command_normalization.py create mode 100644 tests/hermes_cli/test_telegram_command_projection.py create mode 100644 tests/hermes_cli/test_telegram_menu_reconciliation.py diff --git a/hermes_cli/telegram_command_normalization.py b/hermes_cli/telegram_command_normalization.py new file mode 100644 index 000000000000..8b3616e9a389 --- /dev/null +++ b/hermes_cli/telegram_command_normalization.py @@ -0,0 +1,141 @@ +"""Typed Telegram ``/command@bot`` normalization for the command dispatcher.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import Enum + +from hermes_cli.telegram_command_projection import ( + TelegramCommandProjection, + resolve_telegram_command_binding, +) + + +_TELEGRAM_INVOCATION_RE = re.compile( + r"^/(?P[A-Za-z0-9_][A-Za-z0-9_-]*)" + r"(?:@(?P[A-Za-z0-9_]+))?" + r"(?P\s.*)?$", + re.DOTALL, +) + + +class TelegramCommandAttemptStatus(str, Enum): + """Typed classification of one Telegram text input.""" + + NOT_COMMAND = "not_command" + KNOWN_COMMAND = "known_command" + UNKNOWN_COMMAND = "unknown_command" + NOT_FOR_THIS_BOT = "not_for_this_bot" + INVALID_COMMAND = "invalid_command" + + +@dataclass(frozen=True, slots=True) +class TelegramCommandAttempt: + """Normalized Telegram text classification and canonical binding.""" + + status: TelegramCommandAttemptStatus + raw_input: str + entered_name: str | None = None + addressed_bot: str | None = None + raw_arguments: str = "" + command_id: str | None = None + canonical_name: str | None = None + canonical_input: str | None = None + + +def _nonblank_text(value: object) -> str | None: + if not isinstance(value, str): + return None + normalized = value.strip() + return normalized or None + + +def _not_command_slash_text(text: str) -> bool: + token = text.split(None, 1)[0] + if token.startswith("//"): + return True + remainder = token[1:] + return any(character in remainder for character in ("/", "\\", ".", ":")) + + +def normalize_telegram_command_attempt( + text: object, + projection: TelegramCommandProjection, + *, + bot_username: str | None = None, +) -> TelegramCommandAttempt: + """Classify and normalize one Telegram text without prompt fallthrough. + + A matching ``/command@bot args`` and ``/command args`` produce the same + canonical identity, arguments, and dispatcher input. Unknown inputs that + satisfy Telegram's command grammar remain typed ``unknown_command`` + attempts; slash-prefixed paths/code remain ordinary text. + """ + + if not isinstance(text, str): + return TelegramCommandAttempt( + status=TelegramCommandAttemptStatus.NOT_COMMAND, + raw_input="" if text is None else str(text), + ) + + raw_input = text + normalized = text.strip() + if not normalized.startswith("/"): + return TelegramCommandAttempt( + status=TelegramCommandAttemptStatus.NOT_COMMAND, + raw_input=raw_input, + ) + + match = _TELEGRAM_INVOCATION_RE.fullmatch(normalized) + if match is None: + status = ( + TelegramCommandAttemptStatus.NOT_COMMAND + if _not_command_slash_text(normalized) + else TelegramCommandAttemptStatus.INVALID_COMMAND + ) + return TelegramCommandAttempt(status=status, raw_input=raw_input) + + entered_name = match.group("name") + addressed_bot = match.group("bot") + raw_arguments = (match.group("tail") or "").lstrip() + + if addressed_bot is not None: + current_bot = _nonblank_text(bot_username) + current_bot = current_bot.lstrip("@") if current_bot is not None else None + if current_bot is None or addressed_bot.casefold() != current_bot.casefold(): + return TelegramCommandAttempt( + status=TelegramCommandAttemptStatus.NOT_FOR_THIS_BOT, + raw_input=raw_input, + entered_name=entered_name, + addressed_bot=addressed_bot, + raw_arguments=raw_arguments, + ) + + binding = resolve_telegram_command_binding(projection, entered_name) + if binding is None: + canonical_input = f"/{entered_name}" + if raw_arguments: + canonical_input = f"{canonical_input} {raw_arguments}" + return TelegramCommandAttempt( + status=TelegramCommandAttemptStatus.UNKNOWN_COMMAND, + raw_input=raw_input, + entered_name=entered_name, + addressed_bot=addressed_bot, + raw_arguments=raw_arguments, + canonical_input=canonical_input, + ) + + canonical_input = f"/{binding.canonical_name}" + if raw_arguments: + canonical_input = f"{canonical_input} {raw_arguments}" + return TelegramCommandAttempt( + status=TelegramCommandAttemptStatus.KNOWN_COMMAND, + raw_input=raw_input, + entered_name=entered_name, + addressed_bot=addressed_bot, + raw_arguments=raw_arguments, + command_id=binding.command_id, + canonical_name=binding.canonical_name, + canonical_input=canonical_input, + ) diff --git a/hermes_cli/telegram_command_projection.py b/hermes_cli/telegram_command_projection.py new file mode 100644 index 000000000000..897ff75623b2 --- /dev/null +++ b/hermes_cli/telegram_command_projection.py @@ -0,0 +1,527 @@ +"""Immutable Telegram menu projection of the canonical command catalog. + +This is the bounded PR-8 projection seam under #96692. It consumes a catalog +snapshot and owns only Telegram presentation: command-name sanitization, native +visibility, native limits, and one deterministic projection fingerprint. +Command identity, aliases, policy, arguments, availability, and execution remain +catalog/dispatcher concerns. + +Current main does not yet expose the versioned ``CommandCatalog`` ABI, so this +module accepts catalog-shaped objects without importing a second schema. A +non-blank ``command_id`` is authoritative when present; canonical name is the +explicit v1 compatibility fallback until the stable-id slice lands. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from enum import Enum +from typing import Any + + +TELEGRAM_BOT_API_MAX_COMMANDS = 100 +TELEGRAM_COMMAND_NAME_MAX_LENGTH = 32 + +_MISSING = object() +_TELEGRAM_TYPED_NAME_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_-]*$") +_TELEGRAM_NATIVE_INVALID_RE = re.compile(r"[^a-z0-9_]") +_TELEGRAM_MULTI_UNDERSCORE_RE = re.compile(r"_{2,}") + + +class TelegramMenuOmissionReason(str, Enum): + """Why a catalog command is absent from Telegram's native menu.""" + + HIDDEN = "hidden" + NATIVE_NAME_INVALID = "native_name_invalid" + NATIVE_LIMIT = "native_limit" + + +@dataclass(frozen=True, slots=True) +class TelegramCommandBinding: + """Telegram syntax binding for one canonical catalog command.""" + + command_id: str + canonical_name: str + aliases: tuple[str, ...] + description: str + typed_tokens: tuple[str, ...] + native_name: str | None + native_visible: bool + + +@dataclass(frozen=True, slots=True) +class TelegramNativeCommand: + """One exact Telegram ``BotCommand`` projection row.""" + + command_id: str + command: str + description: str + + +@dataclass(frozen=True, slots=True) +class TelegramMenuOmission: + """One catalog command intentionally absent from the native menu.""" + + command_id: str + canonical_name: str + reason: TelegramMenuOmissionReason + + +@dataclass(frozen=True, slots=True) +class TelegramCommandProjection: + """Immutable Telegram projection of one exact catalog snapshot.""" + + catalog_revision: str + projection_fingerprint: str + bindings: tuple[TelegramCommandBinding, ...] + native_commands: tuple[TelegramNativeCommand, ...] + omissions: tuple[TelegramMenuOmission, ...] + + @property + def native_payload(self) -> tuple[tuple[str, str], ...]: + """Return the exact ordered payload used by Telegram's Bot API.""" + + return tuple( + (command.command, command.description) + for command in self.native_commands + ) + + +def _value(source: object, name: str, default: Any = None) -> Any: + if isinstance(source, Mapping): + return source.get(name, default) + return getattr(source, name, default) + + +def _command_value(source: object, name: str, default: Any = None) -> Any: + value = _value(source, name, _MISSING) + if value is not _MISSING: + return value + legacy = _value(source, "legacy", None) + if isinstance(legacy, Mapping): + return legacy.get(name, default) + return default + + +def _nonblank_text(value: object) -> str | None: + if not isinstance(value, str): + return None + normalized = value.strip() + return normalized or None + + +def _string_tuple(value: object) -> tuple[str, ...]: + if value is None: + return () + if isinstance(value, str): + return (value,) + if not isinstance(value, Iterable): + return () + result: list[str] = [] + for item in value: + text = _nonblank_text(item) + if text is not None: + result.append(text) + return tuple(result) + + +def _normalized_tokens(value: object) -> frozenset[str]: + return frozenset( + item.casefold().replace("_", "-") for item in _string_tuple(value) + ) + + +def _catalog_commands(catalog: object) -> tuple[object, ...]: + commands = _value(catalog, "commands", None) + if commands is None: + commands = catalog + if isinstance(commands, (str, bytes, Mapping)): + raise TypeError("Telegram projection requires an iterable of catalog commands") + try: + return tuple(commands) # type: ignore[arg-type] + except TypeError as exc: + raise TypeError( + "Telegram projection requires an iterable of catalog commands" + ) from exc + + +def _catalog_revision(catalog: object, explicit_revision: str | None) -> str | None: + if explicit_revision is not None: + revision = _nonblank_text(explicit_revision) + if revision is None: + raise ValueError("catalog revision must not be blank") + return revision + for field in ("revision", "catalog_revision", "fingerprint"): + revision = _nonblank_text(_value(catalog, field, None)) + if revision is not None: + return revision + return None + + +def _canonical_name(command: object) -> str: + name = _nonblank_text(_value(command, "name", None)) + if name is None: + name = _nonblank_text(_value(command, "canonical_name", None)) + if name is None: + raise ValueError("Telegram catalog command name must not be blank") + if not _TELEGRAM_TYPED_NAME_RE.fullmatch(name): + raise ValueError(f"invalid Telegram typed command name: {name}") + return name + + +def _command_id(command: object, canonical_name: str) -> str: + command_id = _nonblank_text(_value(command, "command_id", None)) + return command_id or canonical_name + + +def _aliases(command: object) -> tuple[str, ...]: + aliases: list[str] = [] + for alias in _string_tuple(_value(command, "aliases", ())): + normalized = alias.removeprefix("/") + if _TELEGRAM_TYPED_NAME_RE.fullmatch(normalized): + aliases.append(normalized) + return tuple(aliases) + + +def _telegram_presentation_override(command: object) -> Mapping[str, Any]: + overrides = _value(command, "presentation_overrides", None) + if not isinstance(overrides, Mapping): + return {} + telegram = overrides.get("telegram") + return telegram if isinstance(telegram, Mapping) else {} + + +def _description(command: object, canonical_name: str) -> str: + telegram = _telegram_presentation_override(command) + candidates = ( + telegram.get("description"), + telegram.get("description_fallback"), + telegram.get("label"), + _command_value(command, "telegram_label", None), + _value(command, "description_fallback", None), + _value(command, "description", None), + ) + for candidate in candidates: + text = _nonblank_text(candidate) + if text is not None: + return re.sub(r"\s+", " ", text) + return f"Run /{canonical_name}" + + +def _visibility_tokens(visibility: object) -> frozenset[str]: + if isinstance(visibility, str): + return frozenset({visibility.casefold().replace("_", "-")}) + if isinstance(visibility, Mapping): + return frozenset() + if isinstance(visibility, Sequence): + return frozenset( + str(item).strip().casefold().replace("_", "-") + for item in visibility + if str(item).strip() + ) + return frozenset() + + +def _native_visible(command: object) -> bool: + if bool(_command_value(command, "hidden", False)) or bool( + _command_value(command, "debug", False) + ): + return False + + telegram = _telegram_presentation_override(command) + if bool(telegram.get("hidden")) or bool(telegram.get("debug")): + return False + for key in ("native_menu", "native-menu"): + if key in telegram and not bool(telegram[key]): + return False + + visibility = _command_value(command, "visibility", None) + if isinstance(visibility, Mapping): + if bool(visibility.get("hidden")) or bool(visibility.get("debug")): + return False + for key in ("native_menu", "native-menu"): + if key in visibility: + return bool(visibility[key]) + telegram_visibility = visibility.get("telegram") + if isinstance(telegram_visibility, bool): + return telegram_visibility + if isinstance(telegram_visibility, Mapping): + if bool(telegram_visibility.get("hidden")) or bool( + telegram_visibility.get("debug") + ): + return False + for key in ("native_menu", "native-menu"): + if key in telegram_visibility: + return bool(telegram_visibility[key]) + return True + + tokens = _visibility_tokens(visibility) + if tokens & {"hidden", "debug"}: + return False + presentation_tokens = tokens & {"help", "completion", "native-menu"} + return not presentation_tokens or "native-menu" in presentation_tokens + + +def _availability_mapping_supports_telegram(availability: Mapping[str, Any]) -> bool: + if availability.get("available") is False: + return False + + unsupported = _normalized_tokens( + availability.get("unsupported_platforms") + or availability.get("unsupported_surfaces") + ) + if unsupported & {"telegram", "gateway", "messaging"}: + return False + + supported_value = ( + availability.get("supported_platforms") + or availability.get("supported_surfaces") + ) + supported = _normalized_tokens(supported_value) + if supported and not supported & {"telegram", "gateway", "messaging"}: + return False + return True + + +def _supports_telegram(command: object) -> bool: + if _command_value(command, "available", True) is False: + return False + + unsupported = _normalized_tokens( + _command_value(command, "unsupported_platforms", None) + or _command_value(command, "unsupported_surfaces", None) + ) + if unsupported & {"telegram", "gateway", "messaging"}: + return False + + supported_value = ( + _command_value(command, "supported_platforms", None) + or _command_value(command, "supported_surfaces", None) + ) + supported = _normalized_tokens(supported_value) + if supported and not supported & {"telegram", "gateway", "messaging"}: + return False + + availability = _command_value(command, "availability", None) + if isinstance(availability, Mapping) and not _availability_mapping_supports_telegram( + availability + ): + return False + + # Compatibility for a raw current-v1/PR-1 catalog instead of the future + # context-filtered snapshot. Unresolved CLI-only rows fail closed. + if bool(_command_value(command, "cli_only", False)) and not bool( + _command_value(command, "gateway_only", False) + ): + return False + return True + + +def _sanitize_native_name(raw: str) -> str | None: + name = raw.casefold().replace("-", "_") + name = _TELEGRAM_NATIVE_INVALID_RE.sub("", name) + name = _TELEGRAM_MULTI_UNDERSCORE_RE.sub("_", name).strip("_") + if not name or len(name) > TELEGRAM_COMMAND_NAME_MAX_LENGTH: + return None + return name + + +def _binding_tokens(canonical_name: str, aliases: tuple[str, ...]) -> tuple[str, ...]: + result: list[str] = [] + seen: set[str] = set() + for name in (canonical_name, *aliases): + for candidate in (name, _sanitize_native_name(name)): + if candidate is None: + continue + key = candidate.casefold() + if key not in seen: + seen.add(key) + result.append(key) + return tuple(result) + + +def _stable_digest(payload: object) -> str: + encoded = json.dumps( + payload, + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def build_telegram_command_projection( + catalog: object, + *, + max_commands: int = TELEGRAM_BOT_API_MAX_COMMANDS, + catalog_revision: str | None = None, +) -> TelegramCommandProjection: + """Project one exact catalog snapshot into Telegram syntax and menu rows. + + Catalog order is preserved. Native clipping never removes a command from + typed resolution. Duplicate IDs, aliases, and sanitized Telegram names fail + closed instead of allowing catalog order to select authority. + """ + + if isinstance(max_commands, bool) or not isinstance(max_commands, int): + raise TypeError("max_commands must be an integer") + if not 0 <= max_commands <= TELEGRAM_BOT_API_MAX_COMMANDS: + raise ValueError( + f"max_commands must be between 0 and {TELEGRAM_BOT_API_MAX_COMMANDS}" + ) + + bindings: list[TelegramCommandBinding] = [] + seen_ids: dict[str, str] = {} + seen_tokens: dict[str, str] = {} + seen_native_names: dict[str, str] = {} + + for command in _catalog_commands(catalog): + if not _supports_telegram(command): + continue + + canonical_name = _canonical_name(command) + command_id = _command_id(command, canonical_name) + aliases = _aliases(command) + typed_tokens = _binding_tokens(canonical_name, aliases) + native_name = _sanitize_native_name(canonical_name) + binding = TelegramCommandBinding( + command_id=command_id, + canonical_name=canonical_name, + aliases=aliases, + description=_description(command, canonical_name), + typed_tokens=typed_tokens, + native_name=native_name, + native_visible=_native_visible(command), + ) + + id_key = command_id.casefold() + if id_key in seen_ids: + raise ValueError( + "duplicate Telegram command identity: " + f"{command_id} ({seen_ids[id_key]} and {canonical_name})" + ) + seen_ids[id_key] = canonical_name + + for token in typed_tokens: + owner = seen_tokens.get(token) + if owner is not None and owner != command_id: + raise ValueError( + f"Telegram token collision for {token!r}: {owner} vs {command_id}" + ) + seen_tokens[token] = command_id + + if native_name is not None: + owner = seen_native_names.get(native_name) + if owner is not None and owner != command_id: + raise ValueError( + "Telegram native-name collision for " + f"{native_name!r}: {owner} vs {command_id}" + ) + seen_native_names[native_name] = command_id + + bindings.append(binding) + + semantic_payload = [ + { + "command_id": binding.command_id, + "canonical_name": binding.canonical_name, + "aliases": binding.aliases, + "description": binding.description, + "typed_tokens": binding.typed_tokens, + "native_name": binding.native_name, + "native_visible": binding.native_visible, + } + for binding in bindings + ] + revision = _catalog_revision(catalog, catalog_revision) + if revision is None: + revision = f"compat-v1:{_stable_digest(semantic_payload)}" + + candidates: list[tuple[TelegramCommandBinding, TelegramNativeCommand]] = [] + omissions: list[TelegramMenuOmission] = [] + for binding in bindings: + if not binding.native_visible: + omissions.append( + TelegramMenuOmission( + binding.command_id, + binding.canonical_name, + TelegramMenuOmissionReason.HIDDEN, + ) + ) + elif binding.native_name is None: + omissions.append( + TelegramMenuOmission( + binding.command_id, + binding.canonical_name, + TelegramMenuOmissionReason.NATIVE_NAME_INVALID, + ) + ) + else: + candidates.append( + ( + binding, + TelegramNativeCommand( + binding.command_id, + binding.native_name, + binding.description, + ), + ) + ) + + native_commands = tuple(command for _binding, command in candidates[:max_commands]) + omissions.extend( + TelegramMenuOmission( + binding.command_id, + binding.canonical_name, + TelegramMenuOmissionReason.NATIVE_LIMIT, + ) + for binding, _command in candidates[max_commands:] + ) + + projection_payload = { + "catalog_revision": revision, + "max_commands": max_commands, + "bindings": semantic_payload, + "native_commands": [ + { + "command_id": command.command_id, + "command": command.command, + "description": command.description, + } + for command in native_commands + ], + "omissions": [ + { + "command_id": omission.command_id, + "canonical_name": omission.canonical_name, + "reason": omission.reason.value, + } + for omission in omissions + ], + } + return TelegramCommandProjection( + catalog_revision=revision, + projection_fingerprint=_stable_digest(projection_payload), + bindings=tuple(bindings), + native_commands=native_commands, + omissions=tuple(omissions), + ) + + +def resolve_telegram_command_binding( + projection: TelegramCommandProjection, + token: str, +) -> TelegramCommandBinding | None: + """Resolve one canonical, alias, or sanitized Telegram token.""" + + key = str(token or "").lstrip("/").casefold() + if not key: + return None + for binding in projection.bindings: + if key in binding.typed_tokens: + return binding + return None diff --git a/hermes_cli/telegram_menu_reconciliation.py b/hermes_cli/telegram_menu_reconciliation.py new file mode 100644 index 000000000000..b3af37d4c709 --- /dev/null +++ b/hermes_cli/telegram_menu_reconciliation.py @@ -0,0 +1,195 @@ +"""Revision-aware Telegram native-menu reconciliation and exact settlement.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from enum import Enum + +from hermes_cli.telegram_command_projection import ( + TelegramCommandProjection, + TelegramNativeCommand, +) + + +class TelegramMenuReconciliationAction(str, Enum): + """Required native-registration action for one Telegram scope.""" + + NOOP = "noop" + ADOPT = "adopt" + SET = "set" + + +class TelegramMenuVerificationStatus(str, Enum): + """Terminal result of a reconciliation read-back.""" + + SETTLED = "settled" + MISMATCH = "mismatch" + + +@dataclass(frozen=True, slots=True) +class TelegramMenuSettlement: + """Last exact native-menu object proved for one Telegram scope.""" + + scope: str + catalog_revision: str + projection_fingerprint: str + + +@dataclass(frozen=True, slots=True) +class TelegramMenuReconciliationPlan: + """Deterministic action required to align one Telegram menu scope.""" + + scope: str + action: TelegramMenuReconciliationAction + reason: str + catalog_revision: str + projection_fingerprint: str + desired_commands: tuple[tuple[str, str], ...] + observed_commands: tuple[tuple[str, str], ...] + prior_settlement: TelegramMenuSettlement | None + + @property + def requires_set(self) -> bool: + return self.action is TelegramMenuReconciliationAction.SET + + @property + def requires_read_back(self) -> bool: + return self.action is TelegramMenuReconciliationAction.SET + + +@dataclass(frozen=True, slots=True) +class TelegramMenuVerification: + """Exact post-reconciliation proof; mismatches never advance state.""" + + status: TelegramMenuVerificationStatus + expected_commands: tuple[tuple[str, str], ...] + observed_commands: tuple[tuple[str, str], ...] + settlement: TelegramMenuSettlement | None + + +def _nonblank_text(value: object) -> str | None: + if not isinstance(value, str): + return None + normalized = value.strip() + return normalized or None + + +def _native_command_pair(command: object) -> tuple[str, str]: + if isinstance(command, TelegramNativeCommand): + return command.command, command.description + if isinstance(command, Mapping): + name = command.get("command", command.get("name")) + description = command.get("description") + elif isinstance(command, Sequence) and not isinstance(command, (str, bytes)): + if len(command) < 2: + raise ValueError("Telegram native command tuple requires name and description") + name, description = command[0], command[1] + else: + name = getattr(command, "command", getattr(command, "name", None)) + description = getattr(command, "description", None) + + if not isinstance(name, str) or not isinstance(description, str): + raise ValueError("Telegram native command requires string name and description") + return name, description + + +def normalize_telegram_native_commands( + commands: Iterable[object], +) -> tuple[tuple[str, str], ...]: + """Normalize Bot API objects, mappings, or tuples for exact comparison.""" + + return tuple(_native_command_pair(command) for command in commands) + + +def plan_telegram_menu_reconciliation( + projection: TelegramCommandProjection, + observed_commands: Iterable[object], + *, + scope: str, + prior_settlement: TelegramMenuSettlement | None = None, +) -> TelegramMenuReconciliationPlan: + """Plan exact revision-aware reconciliation for one Telegram menu scope.""" + + normalized_scope = _nonblank_text(scope) + if normalized_scope is None: + raise ValueError("Telegram menu reconciliation scope must not be blank") + + desired = projection.native_payload + observed = normalize_telegram_native_commands(observed_commands) + prior_matches_scope = ( + prior_settlement is not None and prior_settlement.scope == normalized_scope + ) + prior_is_current = ( + prior_matches_scope + and prior_settlement is not None + and prior_settlement.catalog_revision == projection.catalog_revision + and prior_settlement.projection_fingerprint + == projection.projection_fingerprint + ) + + if observed == desired: + action = ( + TelegramMenuReconciliationAction.NOOP + if prior_is_current + else TelegramMenuReconciliationAction.ADOPT + ) + reason = "in_sync" if prior_is_current else "observed_current_projection" + else: + action = TelegramMenuReconciliationAction.SET + if prior_is_current: + reason = "remote_drift" + elif prior_matches_scope: + reason = "revision_changed" + else: + reason = "unsettled" + + return TelegramMenuReconciliationPlan( + scope=normalized_scope, + action=action, + reason=reason, + catalog_revision=projection.catalog_revision, + projection_fingerprint=projection.projection_fingerprint, + desired_commands=desired, + observed_commands=observed, + prior_settlement=prior_settlement, + ) + + +def verify_telegram_menu_reconciliation( + plan: TelegramMenuReconciliationPlan, + read_back_commands: Iterable[object] | None = None, +) -> TelegramMenuVerification: + """Settle a plan only after exact payload read-back. + + ``SET`` plans require a post-write read-back. ``ADOPT`` and ``NOOP`` plans + may settle from the exact preflight observation already captured in the + plan. Any mismatch returns a typed terminal result with no settlement. + """ + + if read_back_commands is None: + if plan.action is TelegramMenuReconciliationAction.SET: + raise ValueError("SET reconciliation requires post-write read-back") + observed = plan.observed_commands + else: + observed = normalize_telegram_native_commands(read_back_commands) + + if observed != plan.desired_commands: + return TelegramMenuVerification( + status=TelegramMenuVerificationStatus.MISMATCH, + expected_commands=plan.desired_commands, + observed_commands=observed, + settlement=None, + ) + + settlement = TelegramMenuSettlement( + scope=plan.scope, + catalog_revision=plan.catalog_revision, + projection_fingerprint=plan.projection_fingerprint, + ) + return TelegramMenuVerification( + status=TelegramMenuVerificationStatus.SETTLED, + expected_commands=plan.desired_commands, + observed_commands=observed, + settlement=settlement, + ) diff --git a/tests/hermes_cli/test_telegram_command_normalization.py b/tests/hermes_cli/test_telegram_command_normalization.py new file mode 100644 index 000000000000..0dae01a42f52 --- /dev/null +++ b/tests/hermes_cli/test_telegram_command_normalization.py @@ -0,0 +1,140 @@ +"""Typed Telegram slash-command normalization characterization.""" + +from types import SimpleNamespace + +import pytest + +from hermes_cli.telegram_command_normalization import ( + TelegramCommandAttemptStatus, + normalize_telegram_command_attempt, +) +from hermes_cli.telegram_command_projection import ( + TelegramMenuOmissionReason, + build_telegram_command_projection, +) + + +def _command(name: str, description: str | None = None, **overrides): + values = { + "name": name, + "description": description or f"Run {name}", + "aliases": (), + "command_id": None, + "visibility": None, + "hidden": False, + "debug": False, + "available": True, + "unsupported_surfaces": (), + "supported_surfaces": (), + "cli_only": False, + "gateway_only": False, + "presentation_overrides": {}, + } + values.update(overrides) + return SimpleNamespace(**values) + +def test_command_addressed_to_current_bot_equals_unaddressed_command(): + projection = build_telegram_command_projection( + [_command("status", command_id="session.status")] + ) + + plain = normalize_telegram_command_attempt( + "/status Mixed CASE --Flag", projection, bot_username="HermesBot" + ) + addressed = normalize_telegram_command_attempt( + "/status@hermesbot Mixed CASE --Flag", + projection, + bot_username="@HermesBot", + ) + + assert plain.status is TelegramCommandAttemptStatus.KNOWN_COMMAND + assert addressed.status is TelegramCommandAttemptStatus.KNOWN_COMMAND + assert ( + addressed.command_id, + addressed.canonical_name, + addressed.raw_arguments, + addressed.canonical_input, + ) == ( + plain.command_id, + plain.canonical_name, + plain.raw_arguments, + plain.canonical_input, + ) + + +def test_foreign_or_unprovable_bot_target_fails_closed(): + projection = build_telegram_command_projection([_command("status")]) + + foreign = normalize_telegram_command_attempt( + "/status@OtherBot", projection, bot_username="HermesBot" + ) + unknown_owner = normalize_telegram_command_attempt( + "/status@HermesBot", projection + ) + + assert foreign.status is TelegramCommandAttemptStatus.NOT_FOR_THIS_BOT + assert unknown_owner.status is TelegramCommandAttemptStatus.NOT_FOR_THIS_BOT + assert foreign.command_id is None + assert unknown_owner.command_id is None + + +def test_unknown_slash_attempt_never_becomes_ordinary_text(): + projection = build_telegram_command_projection([_command("status")]) + + attempt = normalize_telegram_command_attempt( + "/definitely_unknown@HermesBot payload", + projection, + bot_username="hermesbot", + ) + + assert attempt.status is TelegramCommandAttemptStatus.UNKNOWN_COMMAND + assert attempt.canonical_input == "/definitely_unknown payload" + assert attempt.command_id is None + + +def test_ordinary_text_paths_and_code_are_not_commands(): + projection = build_telegram_command_projection([_command("status")]) + + for text in ("hello", "/usr/local/bin/hermes", "// comment", "/module.py"): + assert ( + normalize_telegram_command_attempt(text, projection).status + is TelegramCommandAttemptStatus.NOT_COMMAND + ) + assert ( + normalize_telegram_command_attempt("/", projection).status + is TelegramCommandAttemptStatus.INVALID_COMMAND + ) + + +def test_duplicate_stable_identity_fails_closed(): + with pytest.raises(ValueError, match="duplicate Telegram command identity"): + build_telegram_command_projection( + [ + _command("one", command_id="session.same"), + _command("two", command_id="session.same"), + ] + ) + + +def test_alias_or_sanitized_token_collision_fails_closed(): + with pytest.raises(ValueError, match="Telegram token collision"): + build_telegram_command_projection( + [ + _command("foo-bar", command_id="one"), + _command("foo_bar", command_id="two"), + ] + ) + + +def test_overlong_native_name_is_omitted_but_remains_typed(): + long_name = "a" * 33 + projection = build_telegram_command_projection([_command(long_name)]) + + assert projection.native_payload == () + assert projection.omissions[0].reason is TelegramMenuOmissionReason.NATIVE_NAME_INVALID + assert ( + normalize_telegram_command_attempt(f"/{long_name}", projection).status + is TelegramCommandAttemptStatus.KNOWN_COMMAND + ) + + diff --git a/tests/hermes_cli/test_telegram_command_projection.py b/tests/hermes_cli/test_telegram_command_projection.py new file mode 100644 index 000000000000..4cffb4e17f29 --- /dev/null +++ b/tests/hermes_cli/test_telegram_command_projection.py @@ -0,0 +1,231 @@ +"""Projection characterization for Telegram command catalogs.""" + +from dataclasses import FrozenInstanceError +from types import SimpleNamespace + +import pytest + +from hermes_cli.telegram_command_normalization import ( + TelegramCommandAttemptStatus, + normalize_telegram_command_attempt, +) +from hermes_cli.telegram_command_projection import ( + TELEGRAM_BOT_API_MAX_COMMANDS, + TelegramMenuOmissionReason, + build_telegram_command_projection, +) + + +def _command(name: str, description: str | None = None, **overrides): + values = { + "name": name, + "description": description or f"Run {name}", + "aliases": (), + "command_id": None, + "visibility": None, + "hidden": False, + "debug": False, + "available": True, + "unsupported_surfaces": (), + "supported_surfaces": (), + "cli_only": False, + "gateway_only": False, + "presentation_overrides": {}, + } + values.update(overrides) + return SimpleNamespace(**values) + + +def test_projection_uses_stable_id_and_current_v1_name_fallback(): + projection = build_telegram_command_projection( + [ + _command("new", command_id="session.new"), + _command("status", command_id=" "), + ], + catalog_revision="catalog-7", + ) + + assert projection.catalog_revision == "catalog-7" + assert [binding.command_id for binding in projection.bindings] == [ + "session.new", + "status", + ] + assert projection.native_payload == ( + ("new", "Run new"), + ("status", "Run status"), + ) + + +def test_catalog_object_supplies_revision_and_order(): + catalog = SimpleNamespace( + revision="rev-object", + commands=(_command("beta"), _command("alpha")), + ) + + projection = build_telegram_command_projection(catalog) + + assert projection.catalog_revision == "rev-object" + assert [command.command for command in projection.native_commands] == [ + "beta", + "alpha", + ] + + +def test_pr1_catalog_json_shape_uses_nested_legacy_availability(): + catalog = { + "revision": "pr1-revision", + "commands": [ + { + "command_id": "command.clear", + "name": "clear", + "aliases": [], + "description_fallback": "Clear terminal", + "legacy": {"cli_only": True, "gateway_only": False}, + }, + { + "command_id": "command.status", + "name": "status", + "aliases": [], + "description_fallback": "Show status", + "legacy": {"cli_only": False, "gateway_only": False}, + }, + ], + } + + projection = build_telegram_command_projection(catalog) + + assert projection.catalog_revision == "pr1-revision" + assert [binding.command_id for binding in projection.bindings] == [ + "command.status" + ] + assert projection.native_payload == (("status", "Show status"),) + + +def test_projection_is_immutable_and_fingerprinted_deterministically(): + first = build_telegram_command_projection([_command("new"), _command("status")]) + second = build_telegram_command_projection([_command("new"), _command("status")]) + clipped = build_telegram_command_projection( + [_command("new"), _command("status")], max_commands=1 + ) + + assert isinstance(first.bindings, tuple) + assert first.catalog_revision == second.catalog_revision + assert first.projection_fingerprint == second.projection_fingerprint + assert clipped.catalog_revision == first.catalog_revision + assert clipped.projection_fingerprint != first.projection_fingerprint + with pytest.raises(FrozenInstanceError): + first.catalog_revision = "mutated" # type: ignore[misc] + + +def test_hidden_debug_and_unsupported_commands_do_not_leak_to_native_menu(): + projection = build_telegram_command_projection( + [ + _command("visible"), + _command("hidden", hidden=True), + _command("debugger", visibility=("debug",)), + _command("desktop-only", supported_surfaces=("desktop",)), + ] + ) + + assert [command.command for command in projection.native_commands] == ["visible"] + assert [binding.canonical_name for binding in projection.bindings] == [ + "visible", + "hidden", + "debugger", + ] + omitted = {(item.canonical_name, item.reason) for item in projection.omissions} + assert omitted == { + ("hidden", TelegramMenuOmissionReason.HIDDEN), + ("debugger", TelegramMenuOmissionReason.HIDDEN), + } + + +def test_visibility_mapping_can_explicitly_include_or_exclude_native_menu(): + projection = build_telegram_command_projection( + [ + _command("help-only", visibility={"native_menu": False}), + _command("native", visibility={"native_menu": True}), + _command("completion", visibility=("completion",)), + _command("all", visibility=("help", "completion", "native-menu")), + ] + ) + + assert [command.command for command in projection.native_commands] == [ + "native", + "all", + ] + + +def test_telegram_presentation_override_changes_only_projection_text(): + projection = build_telegram_command_projection( + [ + _command( + "status", + description="Canonical status", + presentation_overrides={ + "telegram": {"description": "Telegram status\nsummary"} + }, + ) + ] + ) + + assert projection.native_payload == (("status", "Telegram status summary"),) + + +def test_native_limit_omission_retains_typed_fallback(): + projection = build_telegram_command_projection( + [_command("alpha"), _command("beta"), _command("gamma")], + max_commands=1, + ) + + assert projection.native_payload == (("alpha", "Run alpha"),) + assert { + omission.canonical_name + for omission in projection.omissions + if omission.reason is TelegramMenuOmissionReason.NATIVE_LIMIT + } == {"beta", "gamma"} + + typed = normalize_telegram_command_attempt("/gamma payload", projection) + assert typed.status is TelegramCommandAttemptStatus.KNOWN_COMMAND + assert typed.canonical_name == "gamma" + assert typed.raw_arguments == "payload" + + +def test_zero_native_slots_still_preserves_every_typed_command(): + projection = build_telegram_command_projection( + [_command("alpha"), _command("beta")], max_commands=0 + ) + + assert projection.native_payload == () + assert normalize_telegram_command_attempt( + "/beta", projection + ).status is TelegramCommandAttemptStatus.KNOWN_COMMAND + + +def test_hyphenated_names_and_aliases_share_one_binding(): + projection = build_telegram_command_projection( + [_command("codex-runtime", aliases=("cr",), command_id="runtime.codex")] + ) + + assert projection.native_payload == (("codex_runtime", "Run codex-runtime"),) + for text in ("/codex-runtime", "/codex_runtime", "/cr"): + attempt = normalize_telegram_command_attempt(text, projection) + assert attempt.status is TelegramCommandAttemptStatus.KNOWN_COMMAND + assert attempt.command_id == "runtime.codex" + assert attempt.canonical_input == "/codex-runtime" + + + +@pytest.mark.parametrize("max_commands", [-1, TELEGRAM_BOT_API_MAX_COMMANDS + 1]) +def test_native_limit_bounds_fail_closed(max_commands): + with pytest.raises(ValueError, match="max_commands must be between"): + build_telegram_command_projection( + [_command("status")], max_commands=max_commands + ) + + +def test_non_integer_native_limit_fails_closed(): + with pytest.raises(TypeError, match="max_commands must be an integer"): + build_telegram_command_projection( + [_command("status")], max_commands=True # type: ignore[arg-type] + ) diff --git a/tests/hermes_cli/test_telegram_menu_reconciliation.py b/tests/hermes_cli/test_telegram_menu_reconciliation.py new file mode 100644 index 000000000000..0440522406fe --- /dev/null +++ b/tests/hermes_cli/test_telegram_menu_reconciliation.py @@ -0,0 +1,192 @@ +"""Revision-aware Telegram native-menu reconciliation characterization.""" + +from types import SimpleNamespace + +import pytest + +from hermes_cli.telegram_command_projection import ( + TELEGRAM_BOT_API_MAX_COMMANDS, + build_telegram_command_projection, +) +from hermes_cli.telegram_menu_reconciliation import ( + TelegramMenuReconciliationAction, + TelegramMenuSettlement, + TelegramMenuVerificationStatus, + normalize_telegram_native_commands, + plan_telegram_menu_reconciliation, + verify_telegram_menu_reconciliation, +) + + +def _command(name: str, description: str | None = None, **overrides): + values = { + "name": name, + "description": description or f"Run {name}", + "aliases": (), + "command_id": None, + "visibility": None, + "hidden": False, + "debug": False, + "available": True, + "unsupported_surfaces": (), + "supported_surfaces": (), + "cli_only": False, + "gateway_only": False, + "presentation_overrides": {}, + } + values.update(overrides) + return SimpleNamespace(**values) + +def test_native_command_normalization_accepts_bot_api_shapes(): + bot_command = SimpleNamespace(command="status", description="Show status") + + assert normalize_telegram_native_commands( + [ + bot_command, + {"command": "new", "description": "New session"}, + ("stop", "Stop work"), + ] + ) == ( + ("status", "Show status"), + ("new", "New session"), + ("stop", "Stop work"), + ) + + +def test_exact_observed_projection_is_adopted_then_becomes_noop(): + projection = build_telegram_command_projection( + [_command("status")], catalog_revision="rev-1" + ) + observed = [SimpleNamespace(command="status", description="Run status")] + + adopt = plan_telegram_menu_reconciliation( + projection, observed, scope="default" + ) + assert adopt.action is TelegramMenuReconciliationAction.ADOPT + assert adopt.reason == "observed_current_projection" + adopted = verify_telegram_menu_reconciliation(adopt) + assert adopted.status is TelegramMenuVerificationStatus.SETTLED + assert adopted.settlement is not None + + noop = plan_telegram_menu_reconciliation( + projection, + observed, + scope="default", + prior_settlement=adopted.settlement, + ) + assert noop.action is TelegramMenuReconciliationAction.NOOP + assert noop.reason == "in_sync" + assert noop.requires_set is False + assert noop.requires_read_back is False + + +def test_remote_drift_requires_set_and_exact_read_back(): + projection = build_telegram_command_projection( + [_command("status")], catalog_revision="rev-1" + ) + settlement = TelegramMenuSettlement( + scope="default", + catalog_revision=projection.catalog_revision, + projection_fingerprint=projection.projection_fingerprint, + ) + + plan = plan_telegram_menu_reconciliation( + projection, + [("old", "Stale")], + scope="default", + prior_settlement=settlement, + ) + + assert plan.action is TelegramMenuReconciliationAction.SET + assert plan.reason == "remote_drift" + assert plan.requires_set is True + assert plan.requires_read_back is True + + mismatch = verify_telegram_menu_reconciliation(plan, [("status", "Wrong")]) + assert mismatch.status is TelegramMenuVerificationStatus.MISMATCH + assert mismatch.settlement is None + + settled = verify_telegram_menu_reconciliation(plan, projection.native_payload) + assert settled.status is TelegramMenuVerificationStatus.SETTLED + assert settled.settlement == settlement + + +def test_revision_change_reconciles_and_advances_only_after_read_back(): + old_projection = build_telegram_command_projection( + [_command("status", description="Old")], catalog_revision="rev-1" + ) + old_settlement = TelegramMenuSettlement( + scope="default", + catalog_revision=old_projection.catalog_revision, + projection_fingerprint=old_projection.projection_fingerprint, + ) + new_projection = build_telegram_command_projection( + [_command("status", description="New")], catalog_revision="rev-2" + ) + + plan = plan_telegram_menu_reconciliation( + new_projection, + old_projection.native_payload, + scope="default", + prior_settlement=old_settlement, + ) + + assert plan.action is TelegramMenuReconciliationAction.SET + assert plan.reason == "revision_changed" + with pytest.raises(ValueError, match="requires post-write read-back"): + verify_telegram_menu_reconciliation(plan) + + verification = verify_telegram_menu_reconciliation( + plan, new_projection.native_payload + ) + assert verification.status is TelegramMenuVerificationStatus.SETTLED + assert verification.settlement is not None + assert verification.settlement.catalog_revision == "rev-2" + assert verification.settlement != old_settlement + + +def test_same_payload_new_revision_can_be_adopted_without_rewrite(): + old = build_telegram_command_projection( + [_command("status")], catalog_revision="rev-1" + ) + current = build_telegram_command_projection( + [_command("status")], catalog_revision="rev-2" + ) + old_settlement = TelegramMenuSettlement( + scope="default", + catalog_revision=old.catalog_revision, + projection_fingerprint=old.projection_fingerprint, + ) + + plan = plan_telegram_menu_reconciliation( + current, + current.native_payload, + scope="default", + prior_settlement=old_settlement, + ) + + assert plan.action is TelegramMenuReconciliationAction.ADOPT + verification = verify_telegram_menu_reconciliation(plan) + assert verification.settlement is not None + assert verification.settlement.catalog_revision == "rev-2" + + +def test_settlement_from_another_scope_never_authorizes_noop(): + projection = build_telegram_command_projection([_command("status")]) + other_scope = TelegramMenuSettlement( + scope="all_private_chats", + catalog_revision=projection.catalog_revision, + projection_fingerprint=projection.projection_fingerprint, + ) + + plan = plan_telegram_menu_reconciliation( + projection, + [("stale", "Stale")], + scope="default", + prior_settlement=other_scope, + ) + + assert plan.action is TelegramMenuReconciliationAction.SET + assert plan.reason == "unsettled" + +