"""Machine-readable Telegram inline-button contracts. Only callback actions that resolve to one explicit handler contract may be rendered. Transport readiness is evaluated separately by TelegramGateway; this registry prevents an enabled transport from reviving legacy ghost buttons. """ from __future__ import annotations import re from dataclasses import asdict, dataclass @dataclass(frozen=True) class TelegramButtonActionContract: action: str callback_format: str handler: str risk: str executor: str verifier: str state: str = "active" _STATIC_ACTIONS: dict[str, TelegramButtonActionContract] = { "detail": TelegramButtonActionContract( "detail", "info", "TelegramGateway._send_incident_detail", "low", "telegram_info_query", "telegram_callback_reply_receipt", ), "history": TelegramButtonActionContract( "history", "info", "TelegramGateway._send_incident_history", "low", "telegram_info_query", "telegram_callback_reply_receipt", ), "reanalyze": TelegramButtonActionContract( "reanalyze", "info", "TelegramGateway._send_reanalyze_result", "low", "telegram_reanalysis_query", "telegram_callback_reply_receipt", ), "drift_view_page": TelegramButtonActionContract( "drift_view_page", "info", "TelegramGateway._send_drift_diff_detail", "low", "telegram_info_query", "telegram_callback_reply_receipt", ), "approve": TelegramButtonActionContract( "approve", "nonce", "TelegramGateway.handle_callback", "high", "approval_control_plane", "approval_and_execution_receipt", state="blocked_missing_controlled_apply_contract", ), "reject": TelegramButtonActionContract( "reject", "nonce", "TelegramGateway.handle_callback", "medium", "approval_control_plane", "approval_state_receipt", state="blocked_missing_controlled_apply_contract", ), "silence": TelegramButtonActionContract( "silence", "nonce", "TelegramGateway._handle_silence", "medium", "telegram_silence_store", "telegram_callback_reply_receipt", state="blocked_missing_controlled_apply_contract", ), "snooze": TelegramButtonActionContract( "snooze", "nonce", "TelegramGateway._handle_snooze", "low", "telegram_snooze_store", "telegram_callback_reply_receipt", state="blocked_missing_controlled_apply_contract", ), "tune": TelegramButtonActionContract( "tune", "nonce", "TelegramGateway._handle_auto_tuning", "low", "shadow_candidate_recorder", "candidate_record_receipt", state="shadow_candidate_only", ), "log_manual_fix": TelegramButtonActionContract( "log_manual_fix", "nonce", "TelegramGateway.handle_callback:log_manual_fix", "low", "incident_timeline", "timeline_write_receipt", state="blocked_missing_controlled_apply_contract", ), "drift_view": TelegramButtonActionContract( "drift_view", "nonce", "TelegramGateway._handle_drift_action", "low", "drift_readback", "telegram_callback_reply_receipt", state="blocked_missing_controlled_apply_contract", ), "drift_adopt": TelegramButtonActionContract( "drift_adopt", "nonce", "TelegramGateway._handle_drift_action", "high", "drift_control_plane", "drift_post_verifier", state="blocked_missing_controlled_apply_contract", ), "drift_revert": TelegramButtonActionContract( "drift_revert", "nonce", "TelegramGateway._handle_drift_action", "high", "drift_control_plane", "drift_post_verifier", state="blocked_missing_controlled_apply_contract", ), "la": TelegramButtonActionContract( "la", "llm_short_id", "TelegramGateway._handle_llm_action_callback", "dynamic", "callback_dispatcher", "callback_dispatch_receipt", state="blocked_missing_controlled_apply_contract", ), } _AI_ADVISORY_ACTIONS = { "ai_advisory_handled", "ai_advisory_snooze", "ai_advisory_view", "ai_advisory_produce_cmd", } _LLM_SHORT_ID_RE = re.compile(r"^[0-9a-f]{16}$") def _callback_action(callback_data: str) -> str: return str(callback_data or "").split(":", 1)[0].strip() def _registry_action_contract(action: str) -> TelegramButtonActionContract | None: contract = _STATIC_ACTIONS.get(action) if contract is not None: return contract if contract.state == "active" else None if action in _AI_ADVISORY_ACTIONS: return None # The YAML registry is the existing source of truth for category actions. # Import lazily so registry validation never creates a module cycle. from src.services.callback_dispatcher import get_action_spec spec = get_action_spec(action) if spec is None or spec.callback_format != "info" or spec.risk != "low": return None return TelegramButtonActionContract( action=spec.name, callback_format=spec.callback_format, handler="callback_dispatcher.dispatch_action", risk=spec.risk, executor=f"mcp:{spec.mcp_provider}/{spec.mcp_tool}", verifier="callback_dispatch_receipt", ) def resolve_callback_action_contract( callback_data: str, ) -> TelegramButtonActionContract | None: """Resolve and validate a callback payload without persisting its nonce.""" raw = str(callback_data or "").strip() if not raw or len(raw.encode("utf-8")) > 64: return None parts = raw.split(":") action = _callback_action(raw) contract = _registry_action_contract(action) if contract is None: return None if contract.callback_format == "info": return contract if len(parts) == 2 and bool(parts[1]) else None if contract.callback_format == "nonce": return contract if len(parts) in {4, 5} and all(parts[1:4]) else None if contract.callback_format == "llm_short_id": return ( contract if len(parts) == 2 and _LLM_SHORT_ID_RE.fullmatch(parts[1]) else None ) if contract.callback_format == "advisory": return contract if len(parts) == 3 and all(parts[1:]) else None return None def callback_action_is_registered(callback_data: str) -> bool: return resolve_callback_action_contract(callback_data) is not None def registered_info_callback_actions() -> frozenset[str]: """Return every two-part callback action accepted by the dispatcher.""" from src.services.callback_dispatcher import load_action_registry names = { action for action, contract in _STATIC_ACTIONS.items() if contract.callback_format == "info" and contract.state == "active" } names.update( spec.name for spec in load_action_registry().values() if spec.callback_format == "info" and spec.risk == "low" ) return frozenset(names) def telegram_button_registry_snapshot() -> dict[str, object]: """Return a redaction-safe registry readback for health and tests.""" from src.services.callback_dispatcher import load_action_registry contracts = list(_STATIC_ACTIONS.values()) contracts.extend( TelegramButtonActionContract( action=action, callback_format="advisory", handler="TelegramGateway._handle_ai_advisory_action", risk="low", executor="ai_advisory_control_plane", verifier="telegram_callback_reply_receipt", state="blocked_missing_controlled_apply_contract", ) for action in sorted(_AI_ADVISORY_ACTIONS) ) contracts.extend( TelegramButtonActionContract( action=spec.name, callback_format=spec.callback_format, handler="callback_dispatcher.dispatch_action", risk=spec.risk, executor=f"mcp:{spec.mcp_provider}/{spec.mcp_tool}", verifier="callback_dispatch_receipt", state=( "active" if spec.callback_format == "info" and spec.risk == "low" else "break_glass_only" if spec.risk == "critical" else "blocked_missing_controlled_apply_contract" ), ) for spec in load_action_registry().values() if spec.name not in _STATIC_ACTIONS ) rows = sorted((asdict(item) for item in contracts), key=lambda item: item["action"]) return { "schema_version": "telegram_button_action_registry_v1", "declared_action_count": len(rows), "registered_action_count": sum( 1 for item in rows if item["state"] == "active" ), "blocked_action_count": sum( 1 for item in rows if item["state"] != "active" ), "unknown_action_policy": "do_not_render", "write_action_policy": "fail_closed_until_controlled_apply_contract", "actions": rows, } __all__ = [ "TelegramButtonActionContract", "callback_action_is_registered", "registered_info_callback_actions", "resolve_callback_action_contract", "telegram_button_registry_snapshot", ]