"""Telegram bot interface for Steward.""" import logging from collections import defaultdict from telegram import Update from telegram.constants import ParseMode from telegram.ext import ( Application, CommandHandler, ContextTypes, MessageHandler, filters, ) from steward.config import Settings from steward.llm.client import LLMClient from steward.memory.thread_store import ThreadMemoryStore, ThreadSummary from steward.proposals.generator import Proposal, ProposalGenerator logger = logging.getLogger(__name__) # Per-user conversation history for non-threaded messages (capped at _MAX_HISTORY turns). _history: dict[int, list[dict[str, str]]] = defaultdict(list) _MAX_HISTORY = 20 # Per-thread conversation history: (chat_id, thread_id) → full history (unbounded). # Messages belonging to a Telegram message thread are kept in their entirety here # until explicitly flushed by the /flush command. _thread_history: dict[tuple[int, int], list[dict[str, str]]] = defaultdict(list) _FLUSH_SYSTEM_PROMPT = ( "You are Steward. The following is a complete Telegram message thread conversation. " "Produce a concise but comprehensive summary that captures:\n" "- The main topics discussed\n" "- Key decisions or conclusions reached\n" "- Any outstanding actions or open questions\n" "- Important context that would help recall this conversation later\n\n" "Be precise. Omit pleasantries." ) def _thread_key(update: Update) -> tuple[int, int] | None: """Return the (chat_id, thread_id) key if the message is part of a thread, else None.""" msg = update.message chat = update.effective_chat if msg is None or chat is None: return None thread_id = msg.message_thread_id if thread_id is None: return None return (chat.id, thread_id) def _is_allowed(user_id: int, settings: Settings) -> bool: """Return True if the user is in the allow-list (or no list is configured).""" if not settings.telegram_allowed_user_ids: return True return user_id in settings.telegram_allowed_user_ids async def _send_long(update: Update, text: str) -> None: """Send text, splitting across messages if it exceeds Telegram's 4096-char limit.""" limit = 4096 for i in range(0, len(text), limit): await update.message.reply_text( # type: ignore[union-attr] text[i : i + limit], parse_mode=ParseMode.MARKDOWN, ) async def start_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /start.""" settings: Settings = context.bot_data["settings"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return await update.message.reply_text( # type: ignore[union-attr] "Hello, I'm *Steward* \U0001f916\n\n" "I'm your AI-assisted personal operations platform.\n" "Talk to me naturally, or use:\n" "/help – show available commands\n" "/clear – reset conversation history\n" "/flush – summarise and archive this thread's memory\n" "/recall – retrieve archived thread summaries\n" "/analyse – run a manual API analysis right now", parse_mode=ParseMode.MARKDOWN, ) async def help_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /help.""" settings: Settings = context.bot_data["settings"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return await update.message.reply_text( # type: ignore[union-attr] "*Steward commands*\n\n" "/start – greeting\n" "/help – this message\n" "/clear – reset conversation history for this context\n" "/flush – summarise the current thread, store the summary, and compress memory\n" " _(only available inside a message thread)_\n" "/recall – show the stored summary for this thread, or list all summaries\n" "/analyse – trigger an immediate API analysis and proposal", parse_mode=ParseMode.MARKDOWN, ) async def clear_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /clear – wipe conversation history for this context. Inside a message thread: clears the thread's unbounded history. Outside a thread: clears the per-user capped history. """ settings: Settings = context.bot_data["settings"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return key = _thread_key(update) if key is not None: _thread_history[key].clear() await update.message.reply_text( # type: ignore[union-attr] "Thread conversation history cleared." ) else: _history[user.id].clear() await update.message.reply_text( # type: ignore[union-attr] "Conversation history cleared." ) async def flush_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /flush – summarise thread memory, persist it, and compress in-memory history. Steps: 1. Verify the command is issued inside a message thread. 2. Summarise the full thread history via the LLM. 3. Persist the summary in the ThreadMemoryStore (keyed by chat_id + thread_id). 4. Replace the in-memory thread history with a single compressed context message so conversation can continue with the summary as background. """ settings: Settings = context.bot_data["settings"] llm: LLMClient = context.bot_data["llm"] store: ThreadMemoryStore = context.bot_data["thread_store"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return key = _thread_key(update) if key is None: await update.message.reply_text( # type: ignore[union-attr] "\u26a0\ufe0f /flush can only be used inside a message thread." ) return chat_id, thread_id = key history = _thread_history[key] if not history: await update.message.reply_text( # type: ignore[union-attr] "This thread has no conversation history to flush." ) return await update.message.reply_text( # type: ignore[union-attr] "\U0001f4be Summarising thread memory\u2026" ) # Build a readable transcript for the LLM to summarise transcript_lines = [] for msg in history: role_label = "User" if msg["role"] == "user" else "Steward" transcript_lines.append(f"{role_label}: {msg['content']}") transcript = "\n".join(transcript_lines) summary_text = await llm.chat( f"Thread transcript:\n\n{transcript}", system_prompt=_FLUSH_SYSTEM_PROMPT, ) message_count = sum(1 for m in history if m["role"] == "user") thread_summary = ThreadSummary( chat_id=chat_id, thread_id=thread_id, summary=summary_text, message_count=message_count, ) store.save(thread_summary) # Compress: replace history with a single system-context entry so the thread # can continue with the summary as background knowledge. _thread_history[key] = [ {"role": "system", "content": f"Summary of earlier conversation:\n{summary_text}"} ] await update.message.reply_text( # type: ignore[union-attr] f"\u2705 Thread memory flushed and stored " f"(thread `{thread_id}`, {message_count} messages summarised).", parse_mode=ParseMode.MARKDOWN, ) async def recall_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /recall – retrieve stored thread summaries. Inside a thread: shows the stored summary for this thread (if any). Outside a thread: lists all stored summaries (newest first). """ settings: Settings = context.bot_data["settings"] store: ThreadMemoryStore = context.bot_data["thread_store"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return key = _thread_key(update) if key is not None: chat_id, thread_id = key stored = store.get(chat_id, thread_id) if stored is None: await update.message.reply_text( # type: ignore[union-attr] "No stored summary for this thread yet. Use /flush to create one." ) else: await _send_long(update, stored.format_for_telegram()) return # Outside a thread: list all stored summaries all_summaries = store.all() if not all_summaries: await update.message.reply_text( # type: ignore[union-attr] "No thread summaries stored yet. Use /flush inside a message thread." ) return lines = ["\U0001f4da *Stored thread summaries*\n"] for s in all_summaries: date = s.flushed_at[:10] first_line = s.summary.split("\n")[0][:80] lines.append(f"\u2022 Thread `{s.thread_id}` ({date}): {first_line}\u2026") await _send_long(update, "\n".join(lines)) async def analyse_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /analyse – run the proposal generator on demand.""" settings: Settings = context.bot_data["settings"] llm: LLMClient = context.bot_data["llm"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return await update.message.reply_text("Running analysis, please wait...") # type: ignore[union-attr] generator = ProposalGenerator(settings, llm) proposal = await generator.run() if proposal is None: await update.message.reply_text( # type: ignore[union-attr] "Analysis could not be completed. " "Check that `ANALYSIS_TARGET_URL` is configured." ) return await _send_long(update, proposal.format_for_telegram()) async def message_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle plain text messages – forward to LLM and reply. Thread messages: history is stored unbounded under the (chat_id, thread_id) key. Non-thread messages: history is capped at _MAX_HISTORY turns per user. """ settings: Settings = context.bot_data["settings"] llm: LLMClient = context.bot_data["llm"] user = update.effective_user if user is None or not _is_allowed(user.id, settings): return text = update.message.text # type: ignore[union-attr] if not text: return key = _thread_key(update) if key is not None: # Thread message: unbounded history history = _thread_history[key] reply = await llm.chat(text, history=history) history.append({"role": "user", "content": text}) history.append({"role": "assistant", "content": reply}) else: # Non-thread message: capped history per user history = _history[user.id] reply = await llm.chat(text, history=history) history.append({"role": "user", "content": text}) history.append({"role": "assistant", "content": reply}) if len(history) > _MAX_HISTORY * 2: _history[user.id] = history[-(_MAX_HISTORY * 2) :] await _send_long(update, reply) def build_application( settings: Settings, llm: LLMClient, thread_store: ThreadMemoryStore | None = None ) -> Application: # type: ignore[type-arg] """Build and return the Telegram Application.""" app = Application.builder().token(settings.telegram_bot_token).build() app.bot_data["settings"] = settings app.bot_data["llm"] = llm app.bot_data["thread_store"] = thread_store or ThreadMemoryStore(settings.thread_memory_path) app.add_handler(CommandHandler("start", start_handler)) app.add_handler(CommandHandler("help", help_handler)) app.add_handler(CommandHandler("clear", clear_handler)) app.add_handler(CommandHandler("flush", flush_handler)) app.add_handler(CommandHandler("recall", recall_handler)) app.add_handler(CommandHandler("analyse", analyse_handler)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, message_handler)) return app async def send_proposal(app: Application, proposal: Proposal, user_ids: list[int]) -> None: # type: ignore[type-arg] """Send a proposal message to all configured user IDs.""" text = proposal.format_for_telegram() for uid in user_ids: try: await app.bot.send_message( chat_id=uid, text=text, parse_mode=ParseMode.MARKDOWN, ) except Exception: logger.exception("Failed to send proposal to user %d", uid)