Add agentic platform: event bus, webhook API, scheduler, notifications

pyproject.toml

@@ -35,6 +35,9 @@ aiofiles = "^24.1.0"
 aiosqlite = "^0.21.0"
 anthropic = "^0.40.0"
 claude-agent-sdk = "^0.1.30"
+fastapi = "^0.115.0"
+uvicorn = {version = "^0.34.0", extras = ["standard"]}
+apscheduler = "^3.10"
 
 [tool.poetry.scripts]
 claude-telegram-bot = "src.main:run"

src/api/__init__.py

@@ -0,0 +1,5 @@
+"""Webhook API server for receiving external events."""
+
+from .server import create_api_app, run_api_server
+
+__all__ = ["create_api_app", "run_api_server"]

src/api/auth.py

@@ -0,0 +1,61 @@
+"""Webhook signature verification for external providers.
+
+Each provider has its own signing mechanism:
+- GitHub: HMAC-SHA256 with X-Hub-Signature-256 header
+- Generic: shared secret in Authorization header
+"""
+
+import hashlib
+import hmac
+from typing import Optional
+
+import structlog
+
+logger = structlog.get_logger()
+
+
+def verify_github_signature(
+    payload_body: bytes,
+    signature_header: Optional[str],
+    secret: str,
+) -> bool:
+    """Verify GitHub webhook HMAC-SHA256 signature.
+
+    GitHub sends the signature as: sha256=<hex_digest>
+    """
+    if not signature_header:
+        logger.warning("GitHub webhook missing signature header")
+        return False
+
+    if not signature_header.startswith("sha256="):
+        logger.warning("GitHub webhook signature has unexpected format")
+        return False
+
+    expected_signature = (
+        "sha256="
+        + hmac.new(
+            secret.encode("utf-8"),
+            payload_body,
+            hashlib.sha256,
+        ).hexdigest()
+    )
+
+    return hmac.compare_digest(expected_signature, signature_header)
+
+
+def verify_shared_secret(
+    authorization_header: Optional[str],
+    secret: str,
+) -> bool:
+    """Verify a simple shared secret in the Authorization header.
+
+    Expects: Bearer <secret>
+    """
+    if not authorization_header:
+        return False
+
+    if not authorization_header.startswith("Bearer "):
+        return False
+
+    token = authorization_header[7:]
+    return hmac.compare_digest(token, secret)

src/api/server.py

@@ -0,0 +1,192 @@
+"""FastAPI webhook server.
+
+Runs in the same process as the bot, sharing the event loop.
+Receives external webhooks and publishes them as events on the bus.
+"""
+
+import uuid
+from typing import Any, Dict, Optional
+
+import structlog
+from fastapi import FastAPI, Header, HTTPException, Request
+
+from ..config.settings import Settings
+from ..events.bus import EventBus
+from ..events.types import WebhookEvent
+from ..storage.database import DatabaseManager
+from .auth import verify_github_signature, verify_shared_secret
+
+logger = structlog.get_logger()
+
+
+def create_api_app(
+    event_bus: EventBus,
+    settings: Settings,
+    db_manager: Optional[DatabaseManager] = None,
+) -> FastAPI:
+    """Create the FastAPI application."""
+
+    app = FastAPI(
+        title="Claude Code Telegram - Webhook API",
+        version="0.1.0",
+        docs_url="/docs" if settings.development_mode else None,
+        redoc_url=None,
+    )
+
+    @app.get("/health")
+    async def health_check() -> Dict[str, str]:
+        return {"status": "ok"}
+
+    @app.post("/webhooks/{provider}")
+    async def receive_webhook(
+        provider: str,
+        request: Request,
+        x_hub_signature_256: Optional[str] = Header(None),
+        x_github_event: Optional[str] = Header(None),
+        x_github_delivery: Optional[str] = Header(None),
+        authorization: Optional[str] = Header(None),
+    ) -> Dict[str, str]:
+        """Receive and validate webhook from an external provider."""
+        body = await request.body()
+
+        # Verify signature based on provider
+        if provider == "github":
+            secret = settings.github_webhook_secret
+            if not secret:
+                raise HTTPException(
+                    status_code=500,
+                    detail="GitHub webhook secret not configured",
+                )
+            if not verify_github_signature(body, x_hub_signature_256, secret):
+                logger.warning(
+                    "GitHub webhook signature verification failed",
+                    delivery_id=x_github_delivery,
+                )
+                raise HTTPException(status_code=401, detail="Invalid signature")
+
+            event_type_name = x_github_event or "unknown"
+            delivery_id = x_github_delivery or str(uuid.uuid4())
+        else:
+            # Generic provider — require auth (fail-closed)
+            secret = settings.webhook_api_secret
+            if not secret:
+                raise HTTPException(
+                    status_code=500,
+                    detail=(
+                        "Webhook API secret not configured. "
+                        "Set WEBHOOK_API_SECRET to accept "
+                        "webhooks from this provider."
+                    ),
+                )
+            if not verify_shared_secret(authorization, secret):
+                raise HTTPException(status_code=401, detail="Invalid authorization")
+            event_type_name = request.headers.get("X-Event-Type", "unknown")
+            delivery_id = request.headers.get("X-Delivery-ID", str(uuid.uuid4()))
+
+        # Parse JSON payload
+        try:
+            payload: Dict[str, Any] = await request.json()
+        except Exception:
+            payload = {"raw_body": body.decode("utf-8", errors="replace")[:5000]}
+
+        # Atomic dedupe: attempt INSERT first, only publish if new
+        if db_manager and delivery_id:
+            is_new = await _try_record_webhook(
+                db_manager,
+                event_id=str(uuid.uuid4()),
+                provider=provider,
+                event_type=event_type_name,
+                delivery_id=delivery_id,
+                payload=payload,
+            )
+            if not is_new:
+                logger.info(
+                    "Duplicate webhook delivery ignored",
+                    provider=provider,
+                    delivery_id=delivery_id,
+                )
+                return {
+                    "status": "duplicate",
+                    "delivery_id": delivery_id,
+                }
+
+        # Publish event to the bus
+        event = WebhookEvent(
+            provider=provider,
+            event_type_name=event_type_name,
+            payload=payload,
+            delivery_id=delivery_id,
+        )
+
+        await event_bus.publish(event)
+
+        logger.info(
+            "Webhook received and published",
+            provider=provider,
+            event_type=event_type_name,
+            delivery_id=delivery_id,
+            event_id=event.id,
+        )
+
+        return {"status": "accepted", "event_id": event.id}
+
+    return app
+
+
+async def _try_record_webhook(
+    db_manager: DatabaseManager,
+    event_id: str,
+    provider: str,
+    event_type: str,
+    delivery_id: str,
+    payload: Dict[str, Any],
+) -> bool:
+    """Atomically insert a webhook event, returning whether it was new.
+
+    Uses INSERT OR IGNORE on the unique delivery_id column.
+    If the row already exists the insert is a no-op and changes() == 0.
+    Returns True if the event is new (inserted), False if duplicate.
+    """
+    import json
+
+    async with db_manager.get_connection() as conn:
+        await conn.execute(
+            """
+            INSERT OR IGNORE INTO webhook_events
+            (event_id, provider, event_type, delivery_id, payload,
+             processed)
+            VALUES (?, ?, ?, ?, ?, 1)
+            """,
+            (
+                event_id,
+                provider,
+                event_type,
+                delivery_id,
+                json.dumps(payload),
+            ),
+        )
+        cursor = await conn.execute("SELECT changes()")
+        row = await cursor.fetchone()
+        inserted = row[0] > 0 if row else False
+        await conn.commit()
+        return inserted
+
+
+async def run_api_server(
+    event_bus: EventBus,
+    settings: Settings,
+    db_manager: Optional[DatabaseManager] = None,
+) -> None:
+    """Run the FastAPI server using uvicorn."""
+    import uvicorn
+
+    app = create_api_app(event_bus, settings, db_manager)
+
+    config = uvicorn.Config(
+        app=app,
+        host="0.0.0.0",
+        port=settings.api_server_port,
+        log_level="info" if not settings.debug else "debug",
+    )
+    server = uvicorn.Server(config)
+    await server.serve()

src/bot/core.py

@@ -40,7 +40,10 @@ def __init__(self, settings: Settings, dependencies: Dict[str, Any]):
         self.feature_registry: Optional[FeatureRegistry] = None
 
     async def initialize(self) -> None:
-        """Initialize bot application."""
+        """Initialize bot application. Idempotent — safe to call multiple times."""
+        if self.app is not None:
+            return
+
         logger.info("Initializing Telegram bot")
 
         # Create application

src/bot/handlers/callback.py

@@ -1007,8 +1007,10 @@ async def handle_git_callback(
             else:
                 # Clean up diff output for Telegram
                 # Remove emoji symbols that interfere with markdown parsing
-                clean_diff = diff_output.replace("➕", "+").replace("➖", "-").replace("📍", "@")
-
+                clean_diff = (
+                    diff_output.replace("➕", "+").replace("➖", "-").replace("📍", "@")
+                )
+
                 # Limit diff output
                 max_length = 2000
                 if len(clean_diff) > max_length:
@@ -1159,7 +1161,26 @@ def _format_file_size(size: int) -> str:
 def _escape_markdown(text: str) -> str:
     """Escape special markdown characters in text for Telegram."""
     # Escape characters that have special meaning in Telegram Markdown
-    special_chars = ['_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!']
+    special_chars = [
+        "_",
+        "*",
+        "[",
+        "]",
+        "(",
+        ")",
+        "~",
+        "`",
+        ">",
+        "#",
+        "+",
+        "-",
+        "=",
+        "|",
+        "{",
+        "}",
+        ".",
+        "!",
+    ]
     for char in special_chars:
-        text = text.replace(char, f'\\{char}')
+        text = text.replace(char, f"\\{char}")
     return text

src/bot/handlers/command.py

@@ -230,7 +230,9 @@ async def continue_session(update: Update, context: ContextTypes.DEFAULT_TYPE) -
             from ..utils.formatting import ResponseFormatter
 
             formatter = ResponseFormatter(settings)
-            formatted_messages = formatter.format_claude_response(claude_response.content)
+            formatted_messages = formatter.format_claude_response(
+                claude_response.content
+            )
 
             for msg in formatted_messages:
                 await update.message.reply_text(
@@ -987,7 +989,26 @@ def _format_file_size(size: int) -> str:
 def _escape_markdown(text: str) -> str:
     """Escape special markdown characters in text for Telegram."""
     # Escape characters that have special meaning in Telegram Markdown
-    special_chars = ['_', '*', '[', ']', '(', ')', '~', '`', '>', '#', '+', '-', '=', '|', '{', '}', '.', '!']
+    special_chars = [
+        "_",
+        "*",
+        "[",
+        "]",
+        "(",
+        ")",
+        "~",
+        "`",
+        ">",
+        "#",
+        "+",
+        "-",
+        "=",
+        "|",
+        "{",
+        "}",
+        ".",
+        "!",
+    ]
     for char in special_chars:
-        text = text.replace(char, f'\\{char}')
+        text = text.replace(char, f"\\{char}")
     return text

src/bot/handlers/message.py

@@ -122,7 +122,12 @@ def _format_error_message(error_str: str) -> str:
         # Generic error handling
         # Escape special markdown characters in error message
         # Replace problematic chars that break Telegram markdown
-        safe_error = error_str.replace("_", "\\_").replace("*", "\\*").replace("`", "\\`").replace("[", "\\[")
+        safe_error = (
+            error_str.replace("_", "\\_")
+            .replace("*", "\\*")
+            .replace("`", "\\`")
+            .replace("[", "\\[")
+        )
         # Truncate very long errors
         if len(safe_error) > 200:
             safe_error = safe_error[:200] + "..."