feat: add Telegram bot integration

Bridges Telegram messages to Timmy via python-telegram-bot (optional
dependency). The bot token can be supplied through the TELEGRAM_TOKEN
env var or at runtime via the new POST /telegram/setup dashboard
endpoint, which (re)starts the bot without a restart.

Changes:
- src/telegram_bot/bot.py — TelegramBot singleton: token persistence
  (telegram_state.json), lifecycle (start/stop), /start command and
  message handler that forwards to Timmy
- src/dashboard/routes/telegram.py — /telegram/setup and /telegram/status
  FastAPI routes
- src/dashboard/app.py — register telegram router; auto-start/stop bot
  in lifespan hook
- src/config.py — TELEGRAM_TOKEN setting (pydantic-settings)
- pyproject.toml — [telegram] optional extra (python-telegram-bot>=21),
  telegram_bot wheel include
- .env.example — TELEGRAM_TOKEN section
- .gitignore — exclude telegram_state.json (contains token)
- tests/conftest.py — stub telegram/telegram.ext for offline test runs
- tests/test_telegram_bot.py — 16 tests covering token helpers,
  lifecycle, and all dashboard routes (370 total, all passing)

https://claude.ai/code/session_01CNBm3ZLobtx3Z1YogHq8ZS

.env.example

@@ -32,3 +32,9 @@
 
 # Lightning backend: "mock" (default) | "lnd"
 # LIGHTNING_BACKEND=mock
+
+# ── Telegram bot ──────────────────────────────────────────────────────────────
+# Bot token from @BotFather on Telegram.
+# Alternatively, configure via the /telegram/setup dashboard endpoint at runtime.
+# Requires: pip install ".[telegram]"
+# TELEGRAM_TOKEN=

.gitignore

@@ -21,6 +21,9 @@ env/
 # SQLite memory — never commit agent memory
 *.db
 
+# Telegram bot state (contains bot token)
+telegram_state.json
+
 # Testing
 .pytest_cache/
 .coverage

pyproject.toml

@@ -47,6 +47,11 @@ swarm = [
 voice = [
     "pyttsx3>=2.90",
 ]
+# Telegram: bridge Telegram messages to Timmy via python-telegram-bot.
+# pip install ".[telegram]"
+telegram = [
+    "python-telegram-bot>=21.0",
+]
 
 [project.scripts]
 timmy = "timmy.cli:main"
@@ -66,6 +71,7 @@ include = [
     "src/voice",
     "src/notifications",
     "src/shortcuts",
+    "src/telegram_bot",
 ]
 
 [tool.pytest.ini_options]

src/config.py

@@ -13,6 +13,9 @@ class Settings(BaseSettings):
     # Set DEBUG=true to enable /docs and /redoc (disabled by default)
     debug: bool = False
 
+    # Telegram bot token — set via TELEGRAM_TOKEN env var or the /telegram/setup endpoint
+    telegram_token: str = ""
+
     # ── AirLLM / backend selection ───────────────────────────────────────────
     # "ollama"  — always use Ollama (default, safe everywhere)
     # "airllm"  — always use AirLLM (requires pip install ".[bigbrain]")

src/dashboard/app.py

@@ -19,6 +19,7 @@ from dashboard.routes.voice_enhanced import router as voice_enhanced_router
 from dashboard.routes.mobile import router as mobile_router
 from dashboard.routes.swarm_ws import router as swarm_ws_router
 from dashboard.routes.briefing import router as briefing_router
+from dashboard.routes.telegram import router as telegram_router
 
 logging.basicConfig(
     level=logging.INFO,
@@ -62,7 +63,14 @@ async def _briefing_scheduler() -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     task = asyncio.create_task(_briefing_scheduler())
+
+    # Auto-start Telegram bot if a token is configured
+    from telegram_bot.bot import telegram_bot
+    await telegram_bot.start()
+
     yield
+
+    await telegram_bot.stop()
     task.cancel()
     try:
         await task
@@ -92,6 +100,7 @@ app.include_router(voice_enhanced_router)
 app.include_router(mobile_router)
 app.include_router(swarm_ws_router)
 app.include_router(briefing_router)
+app.include_router(telegram_router)
 
 
 @app.get("/", response_class=HTMLResponse)

src/dashboard/routes/telegram.py

@@ -0,0 +1,51 @@
+"""Dashboard routes for Telegram bot setup and status."""
+
+from fastapi import APIRouter
+from pydantic import BaseModel
+
+router = APIRouter(prefix="/telegram", tags=["telegram"])
+
+
+class TokenPayload(BaseModel):
+    token: str
+
+
+@router.post("/setup")
+async def setup_telegram(payload: TokenPayload):
+    """Accept a Telegram bot token, save it, and (re)start the bot.
+
+    Send a POST with JSON body: {"token": "<your-bot-token>"}
+    Get the token from @BotFather on Telegram.
+    """
+    from telegram_bot.bot import telegram_bot
+
+    token = payload.token.strip()
+    if not token:
+        return {"ok": False, "error": "Token cannot be empty."}
+
+    telegram_bot.save_token(token)
+
+    if telegram_bot.is_running:
+        await telegram_bot.stop()
+
+    success = await telegram_bot.start(token=token)
+    if success:
+        return {"ok": True, "message": "Telegram bot started successfully."}
+    return {
+        "ok": False,
+        "error": (
+            "Failed to start bot. Check the token is correct and that "
+            'python-telegram-bot is installed: pip install ".[telegram]"'
+        ),
+    }
+
+
+@router.get("/status")
+async def telegram_status():
+    """Return the current state of the Telegram bot."""
+    from telegram_bot.bot import telegram_bot
+
+    return {
+        "running": telegram_bot.is_running,
+        "token_set": telegram_bot.token_set,
+    }

src/telegram_bot/bot.py

@@ -0,0 +1,163 @@
+"""Telegram bot integration for Timmy Time.
+
+Bridges Telegram messages to Timmy (the local AI agent).  The bot token
+is supplied via the dashboard setup endpoint or the TELEGRAM_TOKEN env var.
+
+Optional dependency — install with:
+    pip install ".[telegram]"
+"""
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# State file lives in the project root alongside timmy.db
+_STATE_FILE = Path(__file__).parent.parent.parent / "telegram_state.json"
+
+
+def _load_token_from_file() -> str | None:
+    """Read the saved bot token from the state file."""
+    try:
+        if _STATE_FILE.exists():
+            data = json.loads(_STATE_FILE.read_text())
+            return data.get("token") or None
+    except Exception as exc:
+        logger.debug("Could not read telegram state file: %s", exc)
+    return None
+
+
+def _save_token_to_file(token: str) -> None:
+    """Persist the bot token to the state file."""
+    _STATE_FILE.write_text(json.dumps({"token": token}))
+
+
+class TelegramBot:
+    """Manages the lifecycle of the python-telegram-bot Application.
+
+    Integrates with an existing asyncio event loop (e.g. FastAPI's).
+    """
+
+    def __init__(self) -> None:
+        self._app = None
+        self._token: str | None = None
+        self._running: bool = False
+
+    # ── Token helpers ─────────────────────────────────────────────────────────
+
+    def load_token(self) -> str | None:
+        """Return the token from the state file or TELEGRAM_TOKEN env var."""
+        from_file = _load_token_from_file()
+        if from_file:
+            return from_file
+        try:
+            from config import settings
+            return settings.telegram_token or None
+        except Exception:
+            return None
+
+    def save_token(self, token: str) -> None:
+        """Persist token so it survives restarts."""
+        _save_token_to_file(token)
+
+    # ── Status ────────────────────────────────────────────────────────────────
+
+    @property
+    def is_running(self) -> bool:
+        return self._running
+
+    @property
+    def token_set(self) -> bool:
+        return bool(self._token)
+
+    # ── Lifecycle ─────────────────────────────────────────────────────────────
+
+    async def start(self, token: str | None = None) -> bool:
+        """Start the bot.  Returns True on success, False otherwise."""
+        if self._running:
+            return True
+
+        tok = token or self.load_token()
+        if not tok:
+            logger.warning("Telegram bot: no token configured, skipping start.")
+            return False
+
+        try:
+            from telegram import Update
+            from telegram.ext import (
+                Application,
+                CommandHandler,
+                ContextTypes,
+                MessageHandler,
+                filters,
+            )
+        except ImportError:
+            logger.error(
+                "python-telegram-bot is not installed. "
+                'Run: pip install ".[telegram]"'
+            )
+            return False
+
+        try:
+            self._token = tok
+            self._app = Application.builder().token(tok).build()
+
+            self._app.add_handler(CommandHandler("start", self._cmd_start))
+            self._app.add_handler(
+                MessageHandler(filters.TEXT & ~filters.COMMAND, self._handle_message)
+            )
+
+            await self._app.initialize()
+            await self._app.start()
+            await self._app.updater.start_polling(allowed_updates=Update.ALL_TYPES)
+
+            self._running = True
+            logger.info("Telegram bot started.")
+            return True
+
+        except Exception as exc:
+            logger.error("Telegram bot failed to start: %s", exc)
+            self._running = False
+            self._token = None
+            self._app = None
+            return False
+
+    async def stop(self) -> None:
+        """Gracefully shut down the bot."""
+        if not self._running or self._app is None:
+            return
+        try:
+            await self._app.updater.stop()
+            await self._app.stop()
+            await self._app.shutdown()
+            logger.info("Telegram bot stopped.")
+        except Exception as exc:
+            logger.error("Error stopping Telegram bot: %s", exc)
+        finally:
+            self._running = False
+
+    # ── Handlers ──────────────────────────────────────────────────────────────
+
+    async def _cmd_start(self, update, context) -> None:
+        await update.message.reply_text(
+            "Sir, affirmative. I'm Timmy — your sovereign local AI agent. "
+            "Send me any message and I'll get right on it."
+        )
+
+    async def _handle_message(self, update, context) -> None:
+        user_text = update.message.text
+        try:
+            from timmy.agent import create_timmy
+            agent = create_timmy()
+            run = await asyncio.to_thread(agent.run, user_text, stream=False)
+            response = run.content if hasattr(run, "content") else str(run)
+        except Exception as exc:
+            logger.error("Timmy error in Telegram handler: %s", exc)
+            response = f"Timmy is offline: {exc}"
+        await update.message.reply_text(response)
+
+
+# Module-level singleton
+telegram_bot = TelegramBot()

tests/conftest.py

@@ -17,6 +17,10 @@ for _mod in [
     # AirLLM is optional (bigbrain extra) — stub it so backend tests can
     # import timmy.backends and instantiate TimmyAirLLMAgent without a GPU.
     "airllm",
+    # python-telegram-bot is optional (telegram extra) — stub so tests run
+    # without the package installed.
+    "telegram",
+    "telegram.ext",
 ]:
     sys.modules.setdefault(_mod, MagicMock())
 

tests/test_telegram_bot.py

@@ -0,0 +1,211 @@
+"""Tests for the Telegram bot integration."""
+
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+# ── TelegramBot unit tests ────────────────────────────────────────────────────
+
+
+class TestTelegramBotTokenHelpers:
+    def test_save_and_load_token(self, tmp_path, monkeypatch):
+        """save_token persists to disk; load_token reads it back."""
+        state_file = tmp_path / "telegram_state.json"
+        monkeypatch.setattr("telegram_bot.bot._STATE_FILE", state_file)
+
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+
+        bot.save_token("test-token-123")
+        assert state_file.exists()
+        data = json.loads(state_file.read_text())
+        assert data["token"] == "test-token-123"
+
+        loaded = bot.load_token()
+        assert loaded == "test-token-123"
+
+    def test_load_token_missing_file(self, tmp_path, monkeypatch):
+        """load_token returns None when no state file and no env var."""
+        state_file = tmp_path / "missing_telegram_state.json"
+        monkeypatch.setattr("telegram_bot.bot._STATE_FILE", state_file)
+
+        # Ensure settings.telegram_token is empty
+        mock_settings = MagicMock()
+        mock_settings.telegram_token = ""
+        with patch("telegram_bot.bot._load_token_from_file", return_value=None):
+            with patch("config.settings", mock_settings):
+                from telegram_bot.bot import TelegramBot
+                bot = TelegramBot()
+                result = bot.load_token()
+        assert result is None
+
+    def test_token_set_property(self):
+        """token_set reflects whether a token has been applied."""
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        assert not bot.token_set
+        bot._token = "tok"
+        assert bot.token_set
+
+    def test_is_running_property(self):
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        assert not bot.is_running
+        bot._running = True
+        assert bot.is_running
+
+
+class TestTelegramBotLifecycle:
+    @pytest.mark.asyncio
+    async def test_start_no_token_returns_false(self, tmp_path, monkeypatch):
+        """start() returns False and stays idle when no token is available."""
+        state_file = tmp_path / "telegram_state.json"
+        monkeypatch.setattr("telegram_bot.bot._STATE_FILE", state_file)
+
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        with patch.object(bot, "load_token", return_value=None):
+            result = await bot.start()
+        assert result is False
+        assert not bot.is_running
+
+    @pytest.mark.asyncio
+    async def test_start_already_running_returns_true(self):
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        bot._running = True
+        result = await bot.start(token="any")
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_start_import_error_returns_false(self):
+        """start() returns False gracefully when python-telegram-bot absent."""
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+
+        with patch.object(bot, "load_token", return_value="tok"), \
+             patch.dict("sys.modules", {"telegram": None, "telegram.ext": None}):
+            result = await bot.start(token="tok")
+        assert result is False
+        assert not bot.is_running
+
+    @pytest.mark.asyncio
+    async def test_stop_when_not_running_is_noop(self):
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        # Should not raise
+        await bot.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_calls_shutdown(self):
+        """stop() invokes the Application shutdown sequence."""
+        from telegram_bot.bot import TelegramBot
+        bot = TelegramBot()
+        bot._running = True
+
+        mock_updater = AsyncMock()
+        mock_app = AsyncMock()
+        mock_app.updater = mock_updater
+        bot._app = mock_app
+
+        await bot.stop()
+
+        mock_updater.stop.assert_awaited_once()
+        mock_app.stop.assert_awaited_once()
+        mock_app.shutdown.assert_awaited_once()
+        assert not bot.is_running
+
+
+# ── Dashboard route tests ─────────────────────────────────────────────────────
+
+
+class TestTelegramRoutes:
+    def test_status_not_running(self, client):
+        """GET /telegram/status returns running=False when bot is idle."""
+        from telegram_bot.bot import telegram_bot
+        telegram_bot._running = False
+        telegram_bot._token = None
+
+        resp = client.get("/telegram/status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["running"] is False
+        assert data["token_set"] is False
+
+    def test_status_running(self, client):
+        """GET /telegram/status returns running=True when bot is active."""
+        from telegram_bot.bot import telegram_bot
+        telegram_bot._running = True
+        telegram_bot._token = "tok"
+
+        resp = client.get("/telegram/status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["running"] is True
+        assert data["token_set"] is True
+
+        # Cleanup
+        telegram_bot._running = False
+        telegram_bot._token = None
+
+    def test_setup_empty_token(self, client):
+        """POST /telegram/setup with empty token returns error."""
+        resp = client.post("/telegram/setup", json={"token": ""})
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["ok"] is False
+        assert "empty" in data["error"].lower()
+
+    def test_setup_success(self, client):
+        """POST /telegram/setup with valid token starts bot and returns ok."""
+        from telegram_bot.bot import telegram_bot
+
+        telegram_bot._running = False
+        with patch.object(telegram_bot, "save_token") as mock_save, \
+             patch.object(telegram_bot, "start", new_callable=AsyncMock, return_value=True):
+            resp = client.post("/telegram/setup", json={"token": "bot123:abc"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["ok"] is True
+        mock_save.assert_called_once_with("bot123:abc")
+
+    def test_setup_failure(self, client):
+        """POST /telegram/setup returns error dict when bot fails to start."""
+        from telegram_bot.bot import telegram_bot
+
+        telegram_bot._running = False
+        with patch.object(telegram_bot, "save_token"), \
+             patch.object(telegram_bot, "start", new_callable=AsyncMock, return_value=False):
+            resp = client.post("/telegram/setup", json={"token": "bad-token"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["ok"] is False
+        assert "error" in data
+
+    def test_setup_stops_running_bot_first(self, client):
+        """POST /telegram/setup stops any running bot before starting new one."""
+        from telegram_bot.bot import telegram_bot
+        telegram_bot._running = True
+
+        with patch.object(telegram_bot, "save_token"), \
+             patch.object(telegram_bot, "stop", new_callable=AsyncMock) as mock_stop, \
+             patch.object(telegram_bot, "start", new_callable=AsyncMock, return_value=True):
+            resp = client.post("/telegram/setup", json={"token": "new-token"})
+
+        mock_stop.assert_awaited_once()
+        assert resp.json()["ok"] is True
+        telegram_bot._running = False
+
+
+# ── Module singleton test ─────────────────────────────────────────────────────
+
+
+def test_module_singleton_exists():
+    """telegram_bot module exposes a singleton TelegramBot instance."""
+    from telegram_bot.bot import telegram_bot, TelegramBot
+    assert isinstance(telegram_bot, TelegramBot)