From a4991b032059cfab25871a13ffed326e87ed3ae3 Mon Sep 17 00:00:00 2001 From: Filyus Date: Sun, 12 Apr 2026 06:30:57 +0500 Subject: [PATCH 01/20] feat: add multi-session router with Telegram topic support --- .github/workflows/ci.yml | 6 +- .github/workflows/release.yml | 28 +- Cargo.lock | 21 + Cargo.toml | 12 + README.md | 129 ++++- src/lib.rs | 1 + src/main.rs | 566 ++++++++++++++++++++- src/router/config.rs | 133 +++++ src/router/mailbox.rs | 498 ++++++++++++++++++ src/router/main.rs | 928 ++++++++++++++++++++++++++++++++++ src/router/mod.rs | 9 + src/router/sessions.rs | 263 ++++++++++ src/router/topics.rs | 192 +++++++ src/telegram/api.rs | 118 +++++ src/telegram/handlers.rs | 12 +- src/telegram/permission.rs | 2 +- src/telegram/tools.rs | 23 +- src/telegram/types.rs | 19 + tests/router_ipc_test.rs | 266 ++++++++++ 19 files changed, 3194 insertions(+), 32 deletions(-) create mode 100644 src/router/config.rs create mode 100644 src/router/mailbox.rs create mode 100644 src/router/main.rs create mode 100644 src/router/mod.rs create mode 100644 src/router/sessions.rs create mode 100644 src/router/topics.rs create mode 100644 tests/router_ipc_test.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4d7828d..889fe8f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,19 +87,21 @@ jobs: if: matrix.cross run: cross build --release --target ${{ matrix.target }} - - name: Package binary (unix) + - name: Package binaries (unix) if: runner.os != 'Windows' run: | mkdir -p dist cp target/${{ matrix.target }}/release/hdcd-telegram dist/ + cp target/${{ matrix.target }}/release/hdcd-router dist/ cd dist && tar czf ../hdcd-telegram-${{ matrix.artifact }}.tar.gz * - - name: Package binary (windows) + - name: Package binaries (windows) if: runner.os == 'Windows' shell: pwsh run: | New-Item -ItemType Directory -Force -Path dist Copy-Item target/${{ matrix.target }}/release/hdcd-telegram.exe dist/ + Copy-Item target/${{ matrix.target }}/release/hdcd-router.exe dist/ Compress-Archive -Path dist/* -DestinationPath hdcd-telegram-${{ matrix.artifact }}.zip - name: Upload artifact diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index c4e0db2..075a723 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -70,6 +70,7 @@ jobs: run: | mkdir -p dist cp target/${{ matrix.target }}/release/hdcd-telegram dist/ + cp target/${{ matrix.target }}/release/hdcd-router dist/ cd dist && tar czf ../hdcd-telegram-${{ github.ref_name }}-${{ matrix.artifact }}.tar.gz * - name: Package (windows) @@ -78,6 +79,7 @@ jobs: run: | New-Item -ItemType Directory -Force -Path dist Copy-Item target/${{ matrix.target }}/release/hdcd-telegram.exe dist/ + Copy-Item target/${{ matrix.target }}/release/hdcd-router.exe dist/ Compress-Archive -Path dist/* -DestinationPath hdcd-telegram-${{ github.ref_name }}-${{ matrix.artifact }}.zip - name: Generate SHA256 checksum (unix) @@ -155,7 +157,13 @@ jobs: xattr -d com.apple.quarantine ./hdcd-telegram ``` - ## Quick start + ## Included binaries + + Each archive contains two binaries: + - **hdcd-telegram** — MCP server (1:1 standalone mode, or `--router` for multi-session) + - **hdcd-router** — multi-session router that creates Telegram forum topics per session + + ## Quick start (standalone) ```bash tar xzf hdcd-telegram-${{ github.ref_name }}-.tar.gz @@ -163,4 +171,20 @@ jobs: claude --dangerously-load-development-channels server:telegram ``` - See [README](https://github.com/gohyperdev/hdcd-telegram#quick-start) for full setup. + ## Quick start (router) + + ```bash + # 1. Configure router + mkdir -p ~/.claude/channels/telegram-router + cat > ~/.claude/channels/telegram-router/config.json << 'EOF' + {"bot_token":"YOUR_TOKEN","supergroup_id":"YOUR_GROUP_ID","allowed_users":["YOUR_USER_ID"]} + EOF + + # 2. Start router (once) + ./hdcd-router + + # 3. Launch Claude Code sessions with --router flag + claude --dangerously-load-development-channels server:telegram + ``` + + See [README](https://github.com/gohyperdev/hdcd-telegram#router-mode-multi-session) for full setup. diff --git a/Cargo.lock b/Cargo.lock index 0945100..0fa40cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -179,6 +179,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs4" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" +dependencies = [ + "rustix", + "windows-sys 0.59.0", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -301,6 +311,7 @@ dependencies = [ "anyhow", "chrono", "dirs", + "fs4", "futures-util", "hex", "rand 0.8.5", @@ -314,6 +325,7 @@ dependencies = [ "tracing", "tracing-subscriber", "uuid", + "windows-sys 0.61.2", ] [[package]] @@ -1741,6 +1753,15 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-sys" version = "0.61.2" diff --git a/Cargo.toml b/Cargo.toml index 457de0b..b0d127f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,18 @@ tokio-util = "0.7" regex-lite = "0.1" hex = "0.4" dirs = "6" +fs4 = "0.13" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_System_Threading", "Win32_Foundation"] } + +[[bin]] +name = "hdcd-telegram" +path = "src/main.rs" + +[[bin]] +name = "hdcd-router" +path = "src/router/main.rs" [dev-dependencies] tokio = { version = "1", features = ["test-util"] } diff --git a/README.md b/README.md index 34f9810..a2ef037 100644 --- a/README.md +++ b/README.md @@ -188,6 +188,131 @@ Done. Your next DM reaches Claude. No ports opened. No webhooks. Everything runs locally over stdio + outbound HTTPS to `api.telegram.org`. +## Router mode (multi-session) + +The default standalone mode supports one Claude Code session per bot token. If you run multiple sessions simultaneously (parallel agents, CI workers, different projects), each would need its own bot -- and only one can poll at a time (Telegram returns 409 Conflict otherwise). + +**Router mode** solves this with a two-binary architecture: + +``` +┌─────────────┐ ┌────────────────┐ ┌──────────────────┐ ┌────────────┐ +│ Telegram │────▶│ hdcd-router │────▶│ hdcd-telegram │────▶│ Claude Code│ +│ (forum │◀────│ (single poll) │◀────│ (--router mode) │◀────│ (session) │ +│ topics) │ └────────────────┘ └──────────────────┘ └────────────┘ +└─────────────┘ one process one per session one per session +``` + +- **hdcd-router** holds the single Telegram polling connection, creates a forum topic per session, and routes messages via filesystem IPC (JSONL mailbox files) +- **hdcd-telegram --router** runs in router mode -- no direct Telegram polling, reads from inbox, writes to outbox +- Each session gets its own forum topic in a Telegram supergroup +- Topics start with a placeholder title (project folder + short session ID, e.g. `hdcd-telegram #a1b2c3`) and Claude renames them via the `set_topic_title` MCP tool once the conversation topic becomes clear + +### Router setup + +#### 1. Create a Telegram supergroup with topics + +1. Create a new group in Telegram (any name, e.g. "Claude Sessions") +2. Open group settings > **Group Type** > set to **Public** or **Private** (this converts it to a supergroup) +3. Go to settings > **Topics** > toggle **ON** (this option only appears after the group is a supergroup) +4. Add your bot to the group +5. Promote the bot to admin with **Manage Topics** enabled (required to create, close, and reopen forum topics). Other admin permissions are optional. + +> **Tip:** If you don't see the Topics toggle, make sure you completed step 2 first — Topics are only available in supergroups, not regular groups. + +#### 2. Get your supergroup ID and user ID + +Send any message in the group, then query the bot API: + +```bash +curl -s "https://api.telegram.org/bot/getUpdates" | jq '.result[-1].message' +``` + +From the response: +- **supergroup ID**: `.chat.id` — negative, starts with `-100` (e.g. `-1001234567890`) +- **your user ID**: `.from.id` — positive number (e.g. `123456789`) + +#### 3. Configure the router + +```bash +mkdir -p ~/.claude/channels/telegram-router +cat > ~/.claude/channels/telegram-router/config.json << 'EOF' +{ + "bot_token": "YOUR_BOT_TOKEN", + "supergroup_id": "-100XXXXXXXXXX", + "allowed_users": ["YOUR_TELEGRAM_USER_ID"] +} +EOF +``` + +Optional config fields: + +| Field | Default | Description | +|---|---|---| +| `close_topic_on_disconnect` | `true` | Close forum topic when session disconnects | +| `outbox_poll_interval_ms` | `200` | How often to check outbox files | +| `health_check_interval_s` | `30` | How often to check if session PIDs are alive | +| `auto_shutdown_delay_s` | `60` | Shut down router after this many seconds with no active sessions (0 = stay running) | + +#### 4. Launch Claude Code sessions + +The router starts automatically when needed. `hdcd-telegram --router` checks `router.lock` on startup -- if the router isn't running, it spawns `hdcd-router` as a background process (both binaries must be in the same directory). The router shuts down automatically after 60 seconds with no active sessions. + +To start the router manually instead: `./hdcd-router` + +Each session uses `hdcd-telegram` in router mode. Add to `.mcp.json`: + +```json +{ + "mcpServers": { + "telegram": { + "command": "/path/to/hdcd-telegram", + "args": ["--router"] + } + } +} +``` + +Then launch as usual: + +```bash +claude --dangerously-load-development-channels server:telegram +``` + +Each session automatically registers with the router, gets a forum topic, and starts receiving messages. + +#### Router commands + +Send these in the General topic of your supergroup: + +| Command | Description | +|---|---| +| `/status` | List active sessions with PIDs, topic IDs, and working directories | +| `/kill ` | Close a session's forum topic | +| `/help` | Show available commands | + +### How router mode works + +``` +~/.claude/channels/telegram-router/ + config.json ← router config + sessions.json ← persistent session registry + router.lock ← heartbeat file (PID + timestamp) + register/ ← session registration files + .json + inbox/ ← Telegram → session (router writes, MCP reads) + .jsonl + outbox/ ← session → Telegram (MCP writes, router reads) + .jsonl +``` + +1. `hdcd-telegram --router` writes a registration file to `register/` +2. `hdcd-router` detects it, creates a forum topic, updates `sessions.json` +3. Telegram messages in the topic are written to `inbox/.jsonl` +4. MCP server reads inbox, converts to `notifications/claude/channel` +5. Claude's replies (via `reply` tool) are written to `outbox/.jsonl` +6. Router reads outbox, sends to the correct forum topic +7. On disconnect (stdin EOF or dead PID), topic is closed + ## Troubleshooting ### Authentication: channels require claude.ai OAuth @@ -288,6 +413,7 @@ If whisper or ffmpeg are not installed, voice messages are forwarded as `"(voice | `WHISPER_MODEL` | `small` | Whisper model size (`tiny`, `base`, `small`, `medium`, `large`) | | `WHISPER_LANGUAGE` | auto-detect | Language hint (`Polish`, `English`, etc.) | | `HDCD_ECHO_TRANSCRIPT` | `true` | Send transcript back for user confirmation before delivering to Claude | +| `ROUTER_STATE_DIR` | `~/.claude/channels/telegram-router` | Router state directory (config.json, sessions, mailbox) | | `RUST_LOG` | `hdcd_telegram=info` | Log level filter ([`tracing-subscriber`](https://docs.rs/tracing-subscriber) format) | ## Running alongside the official Telegram plugin @@ -318,11 +444,12 @@ If only `TELEGRAM_BOT_TOKEN` is set, hdcd-telegram uses it as before — fully b ## Features - **All 8 message types**: text, photo, document, voice, audio, video, video note, sticker -- **4 MCP tools**: `reply` (with chunking, threading, file attachments, MarkdownV2), `react`, `edit_message`, `download_attachment` +- **5 MCP tools**: `reply` (with chunking, threading, file attachments, MarkdownV2), `react`, `edit_message`, `download_attachment`, `set_topic_title` (router mode — lets Claude rename its forum topic when the conversation shifts) - **Access control**: pairing flow (6-hex code), allowlist, group policies with @mention gating - **Permission relay**: inline keyboard for remote tool-use approval/denial (`claude/channel/permission`) - **Voice transcription** (optional): automatic speech-to-text via [whisper](https://github.com/openai/whisper) with echo-back confirmation flow - **Bot commands**: `/start`, `/help`, `/status` +- **Router mode**: multi-session support via `hdcd-router` + forum topics (one topic per session, `/status`, `/kill` commands) - **409 Conflict retry** with exponential backoff - **Clean shutdown** on stdin EOF (no zombie polling) diff --git a/src/lib.rs b/src/lib.rs index cc878ee..f83e5e2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,4 +8,5 @@ //! access control, Bot API client, message handlers, polling, tools, //! voice transcription, and shared types. +pub mod router; pub mod telegram; diff --git a/src/main.rs b/src/main.rs index c065a5e..218ede6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,8 +3,15 @@ //! `hdcd-telegram` — standalone MCP server for the Telegram channel. //! -//! Drop-in replacement for the official Bun-based Telegram plugin. -//! Speaks JSON-RPC 2.0 on stdio and long-polls the Telegram Bot API. +//! Two modes of operation: +//! +//! **Standalone (default):** Direct 1:1 bridge between one Claude Code +//! session and the Telegram Bot API. Polls `getUpdates` directly. +//! +//! **Router mode (`--router`):** Works with an `hdcd-router` process. +//! No polling — reads inbound messages from inbox files, writes +//! outbound messages to outbox files. The router handles all Telegram +//! communication. //! //! Configure in `.mcp.json`: //! ```json @@ -29,6 +36,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Stdout}; use tokio::sync::Mutex; use tracing::{debug, error, info, warn}; +use hdcd_telegram::router::{config as router_config, mailbox, sessions}; use hdcd_telegram::telegram::api::BotCommand; use hdcd_telegram::telegram::{api, handlers, permission, polling, tools, transcribe, types}; @@ -164,6 +172,65 @@ async fn write_frame(stdout: &Arc>, frame: &Value) -> Result<()> { Ok(()) } +fn ok_response(id: Value, result: Value) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "result": result, + }) +} + +fn error_response(id: Value, code: i32, message: &str) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "error": { + "code": code, + "message": message, + } + }) +} + +// ========================================================================= +// Mode detection +// ========================================================================= + +/// Parsed CLI arguments. +struct CliArgs { + router_mode: bool, +} + +fn parse_args() -> CliArgs { + let args: Vec = std::env::args().collect(); + let router_mode = args.iter().any(|a| a == "--router"); + CliArgs { router_mode } +} + +/// Generate a session label from the environment. +/// +/// We only prefix non-"cli" entrypoints — the prefix exists to disambiguate +/// environments (e.g. `claude-vscode`) that behave differently. VS Code panel +/// sessions are refused up-front (channels are silently dropped there), so in +/// practice every session running here is CLI and the prefix is just noise. +fn session_label() -> String { + let cwd = std::env::current_dir() + .ok() + .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned())) + .unwrap_or_else(|| "unknown".into()); + + let ep = std::env::var("CLAUDE_CODE_ENTRYPOINT").unwrap_or_default(); + + if ep.is_empty() || ep == "cli" { + cwd + } else { + format!("{ep}: {cwd}") + } +} + +// ========================================================================= +// Entry point +// ========================================================================= + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -174,6 +241,20 @@ async fn main() -> Result<()> { ) .init(); + let cli = parse_args(); + + if cli.router_mode { + run_router_mode().await + } else { + run_standalone_mode().await + } +} + +// ========================================================================= +// Standalone mode (original behavior, unchanged) +// ========================================================================= + +async fn run_standalone_mode() -> Result<()> { let sd = state_dir()?; let token = load_token(&sd)?; let inbox_dir = sd.join("inbox"); @@ -231,7 +312,7 @@ async fn main() -> Result<()> { } }); - // Approval checker (polls approved/ directory for pairing confirmations). + // Approval checker. let approval_api = Arc::clone(&bot_api); let approval_sd = sd.clone(); let approval_cancel = cancel.clone(); @@ -257,7 +338,7 @@ async fn main() -> Result<()> { pending_transcriptions: tokio::sync::Mutex::new(std::collections::HashMap::new()), }); - // Spawn update processor — reads from update_rx, writes notifications to stdout. + // Spawn update processor. let update_stdout = Arc::clone(&stdout); let update_ctx = Arc::clone(&handler_ctx); tokio::spawn(async move { @@ -305,7 +386,6 @@ async fn main() -> Result<()> { let params = msg.get("params").cloned(); if let Some(ref req_id) = id { - // This is a request — needs a response. let resp = match method.as_str() { "initialize" => { info!("client sent initialize"); @@ -362,7 +442,6 @@ async fn main() -> Result<()> { error!(error = %e, "failed to write response"); } } else { - // Notification (no id). match method.as_str() { "notifications/claude/channel/permission_request" => { if let Some(ref p) = params { @@ -380,31 +459,480 @@ async fn main() -> Result<()> { info!("stdin closed; shutting down"); cancel.cancel(); - // Give polling a couple seconds to exit. let _ = tokio::time::timeout(std::time::Duration::from_secs(2), poll_handle).await; Ok(()) } -fn ok_response(id: Value, result: Value) -> Value { - json!({ - "jsonrpc": "2.0", - "id": id, - "result": result, - }) +// ========================================================================= +// Router mode +// ========================================================================= + +async fn run_router_mode() -> Result<()> { + // Refuse VS Code panel (stream-json) sessions: claude.exe silently drops + // `notifications/claude/channel` in that mode, so inbound Telegram would + // never reach the panel and users would see a connected-but-silent MCP. + // Override with HDCD_ALLOW_VSCODE=1 once the upstream fix ships. + let ep = std::env::var("CLAUDE_CODE_ENTRYPOINT").unwrap_or_default(); + if ep == "claude-vscode" && std::env::var_os("HDCD_ALLOW_VSCODE").is_none() { + eprintln!( + "hdcd-telegram: refusing to start in VS Code panel — channels are dropped by \ + claude.exe in stream-json mode. Use Claude Code in a terminal (CLI) for Telegram \ + integration. Set HDCD_ALLOW_VSCODE=1 to override once upstream fixes it." + ); + return Ok(()); + } + + let router_sd = router_config::state_dir()?; + + // Auto-launch hdcd-router if not already running. + ensure_router_running(&router_sd)?; + + let (inbox_dir, outbox_dir, register_dir) = mailbox::ensure_dirs(&router_sd)?; + + let session_id = uuid::Uuid::new_v4().to_string(); + let short_id = &session_id[..6]; + let label = format!("{} #{short_id}", session_label()); + + info!(session_id, label, "starting in router mode"); + + // Write registration file. + let reg = sessions::Registration { + session_id: session_id.clone(), + label: label.clone(), + pid: Some(std::process::id()), + cwd: std::env::current_dir() + .ok() + .map(|p| p.to_string_lossy().into_owned()), + registered_at: chrono::Utc::now().to_rfc3339(), + disconnected: false, + }; + let reg_path = register_dir.join(format!("{session_id}.json")); + let reg_json = serde_json::to_string_pretty(®).context("serialize registration")?; + std::fs::write(®_path, format!("{reg_json}\n")) + .with_context(|| format!("write registration {}", reg_path.display()))?; + info!(path = %reg_path.display(), "registration file written"); + + let inbox_path = inbox_dir.join(format!("{session_id}.jsonl")); + let outbox_path = outbox_dir.join(format!("{session_id}.jsonl")); + let inbox_pos_path = mailbox::pos_path_for(&inbox_path); + + // Shared stdout writer. + let stdout: Arc> = Arc::new(Mutex::new(tokio::io::stdout())); + + // Cancellation token for clean shutdown. + let cancel = tokio_util::sync::CancellationToken::new(); + + // Spawn inbox poller — reads inbox file, emits MCP channel notifications. + let inbox_stdout = Arc::clone(&stdout); + let inbox_poll_path = inbox_path.clone(); + let inbox_poll_pos = inbox_pos_path.clone(); + let inbox_cancel = cancel.clone(); + tokio::spawn(async move { + let mut interval = + tokio::time::interval(std::time::Duration::from_millis(500)); + loop { + tokio::select! { + _ = interval.tick() => {} + _ = inbox_cancel.cancelled() => return, + } + + let messages: Vec = + match mailbox::read_new_lines(&inbox_poll_path, &inbox_poll_pos) { + Ok(m) => m, + Err(e) => { + warn!(error = %e, "failed to read inbox"); + continue; + } + }; + + for msg in messages { + let frame = inbox_to_notification(&msg); + info!( + chat_id = %msg.chat_id, + message_id = msg.message_id, + user = %msg.user, + "delivering channel notification to stdout" + ); + if let Err(e) = write_frame(&inbox_stdout, &frame).await { + error!(error = %e, "failed to write inbox notification"); + } + } + } + }); + + // Main loop: read stdin JSON-RPC messages. + let stdin = tokio::io::stdin(); + let mut reader = BufReader::new(stdin).lines(); + while let Some(line) = reader.next_line().await? { + if line.trim().is_empty() { + continue; + } + debug!(%line, "stdin"); + + let msg: Value = match serde_json::from_str(&line) { + Ok(v) => v, + Err(e) => { + warn!(error = %e, "failed to parse JSON-RPC message"); + continue; + } + }; + + let id = msg.get("id").cloned(); + let method = msg + .get("method") + .and_then(|m| m.as_str()) + .unwrap_or("") + .to_string(); + let params = msg.get("params").cloned(); + + if let Some(ref req_id) = id { + let resp = match method.as_str() { + "initialize" => { + info!("client sent initialize (router mode)"); + ok_response( + req_id.clone(), + json!({ + "protocolVersion": MCP_PROTOCOL_VERSION, + "capabilities": server_capabilities(), + "serverInfo": server_info(), + "instructions": instructions(), + }), + ) + } + "tools/list" => { + ok_response(req_id.clone(), json!({ "tools": tools::tool_schemas() })) + } + "tools/call" => { + let name = params + .as_ref() + .and_then(|p| p.get("name")) + .and_then(|n| n.as_str()) + .unwrap_or(""); + let args = params + .as_ref() + .and_then(|p| p.get("arguments")) + .cloned() + .unwrap_or(Value::Null); + + match router_tool_call(name, &args, &outbox_path).await { + Ok(result) => ok_response(req_id.clone(), result), + Err(e) => ok_response( + req_id.clone(), + json!({ + "content": [{ "type": "text", "text": format!("{name} failed: {e}") }], + "isError": true, + }), + ), + } + } + "shutdown" => { + info!("client sent shutdown"); + ok_response(req_id.clone(), Value::Null) + } + other => { + debug!(method = other, "unimplemented request method"); + error_response( + req_id.clone(), + -32601, + &format!("method not found: {other}"), + ) + } + }; + if let Err(e) = write_frame(&stdout, &resp).await { + error!(error = %e, "failed to write response"); + } + } else { + // Notifications — permission requests are forwarded through outbox + // for the router to relay. For now, log them. + debug!(method = %method, "notification (router mode, no action)"); + } + } + + // stdin closed — write disconnect marker and shut down. + info!("stdin closed; writing disconnect marker"); + let disconnect_reg = sessions::Registration { + session_id: session_id.clone(), + label, + pid: Some(std::process::id()), + cwd: reg.cwd, + registered_at: reg.registered_at, + disconnected: true, + }; + let disc_json = + serde_json::to_string_pretty(&disconnect_reg).unwrap_or_default(); + let _ = std::fs::write(®_path, format!("{disc_json}\n")); + + cancel.cancel(); + info!("router-mode session ended"); + + Ok(()) } -fn error_response(id: Value, code: i32, message: &str) -> Value { +// --------------------------------------------------------------------------- +// Router-mode helpers +// --------------------------------------------------------------------------- + +/// Convert an inbox message to a MCP `notifications/claude/channel` frame. +fn inbox_to_notification(msg: &mailbox::InboxMessage) -> Value { + let mut meta = serde_json::Map::new(); + meta.insert("chat_id".into(), json!(msg.chat_id)); + meta.insert("message_id".into(), json!(msg.message_id.to_string())); + meta.insert("user".into(), json!(msg.user)); + meta.insert("user_id".into(), json!(msg.user_id)); + meta.insert("ts".into(), json!(msg.ts)); + if let Some(ref path) = msg.image_path { + meta.insert("image_path".into(), json!(path)); + } + if let Some(ref file_id) = msg.attachment_file_id { + meta.insert("attachment_file_id".into(), json!(file_id)); + } + if let Some(ref kind) = msg.attachment_kind { + meta.insert("attachment_kind".into(), json!(kind)); + } + if let Some(ref name) = msg.attachment_name { + meta.insert("attachment_name".into(), json!(name)); + } + if let Some(ref mime) = msg.attachment_mime { + meta.insert("attachment_mime".into(), json!(mime)); + } + if let Some(ref size) = msg.attachment_size { + meta.insert("attachment_size".into(), json!(size)); + } + json!({ "jsonrpc": "2.0", - "id": id, - "error": { - "code": code, - "message": message, + "method": "notifications/claude/channel", + "params": { + "content": msg.text, + "meta": Value::Object(meta), } }) } +/// Handle a tool call in router mode — write to outbox instead of +/// calling the Telegram API directly. +async fn router_tool_call( + name: &str, + args: &Value, + outbox_path: &std::path::Path, +) -> Result { + match name { + "reply" => { + let text = args["text"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing text"))?; + let reply_to = args + .get("reply_to") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + let files: Vec = args + .get("files") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(|s| s.to_string())) + .collect() + }) + .unwrap_or_default(); + let format = args + .get("format") + .and_then(|v| v.as_str()) + .unwrap_or("text") + .to_string(); + + let outbox_msg = mailbox::OutboxMessage { + text: text.to_string(), + reply_to, + files, + format, + edit_message_id: None, + react_message_id: None, + react_emoji: None, + rename_to: None, + }; + mailbox::append_line(outbox_path, &outbox_msg)?; + Ok(json!({ "content": [{ "type": "text", "text": "sent (via router)" }] })) + } + "react" => { + let message_id = args["message_id"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing message_id"))? + .parse::() + .map_err(|_| anyhow::anyhow!("invalid message_id"))?; + let emoji = args["emoji"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing emoji"))?; + + let outbox_msg = mailbox::OutboxMessage { + text: String::new(), + reply_to: None, + files: Vec::new(), + format: "text".into(), + edit_message_id: None, + react_message_id: Some(message_id), + react_emoji: Some(emoji.to_string()), + rename_to: None, + }; + mailbox::append_line(outbox_path, &outbox_msg)?; + Ok(json!({ "content": [{ "type": "text", "text": "reacted (via router)" }] })) + } + "edit_message" => { + let message_id = args["message_id"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing message_id"))? + .parse::() + .map_err(|_| anyhow::anyhow!("invalid message_id"))?; + let text = args["text"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing text"))?; + let format = args + .get("format") + .and_then(|v| v.as_str()) + .unwrap_or("text") + .to_string(); + + let outbox_msg = mailbox::OutboxMessage { + text: text.to_string(), + reply_to: None, + files: Vec::new(), + format, + edit_message_id: Some(message_id), + react_message_id: None, + react_emoji: None, + rename_to: None, + }; + mailbox::append_line(outbox_path, &outbox_msg)?; + Ok(json!({ "content": [{ "type": "text", "text": format!("edited (via router, id: {message_id})") }] })) + } + "download_attachment" => { + // In router mode, attachments are downloaded by the router + // and the path is provided in the inbox message. + anyhow::bail!( + "download_attachment is not available in router mode — \ + the router downloads attachments and provides the path \ + in the inbox message" + ) + } + "set_topic_title" => { + let title = args["title"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing title"))? + .trim(); + if title.is_empty() { + anyhow::bail!("title cannot be empty"); + } + if title.chars().count() > 128 { + anyhow::bail!("title too long (max 128 chars)"); + } + + let outbox_msg = mailbox::OutboxMessage { + text: String::new(), + reply_to: None, + files: Vec::new(), + format: "text".into(), + edit_message_id: None, + react_message_id: None, + react_emoji: None, + rename_to: Some(title.to_string()), + }; + mailbox::append_line(outbox_path, &outbox_msg)?; + Ok(json!({ "content": [{ "type": "text", "text": format!("renamed to \"{title}\" (via router)") }] })) + } + _ => anyhow::bail!("unknown tool: {name}"), + } +} + +/// Check if the router is running by reading `router.lock`. If not, +/// spawn `hdcd-router` as a detached background process. +fn ensure_router_running(state_dir: &std::path::Path) -> Result<()> { + // Skip auto-launch if there's no config (e.g. test environment). + if !state_dir.join("config.json").exists() { + debug!("no config.json in state dir, skipping router auto-launch"); + return Ok(()); + } + + let lock_path = state_dir.join("router.lock"); + + // Check if router.lock exists and PID is alive. + if lock_path.exists() { + if let Ok(content) = std::fs::read_to_string(&lock_path) { + if let Ok(v) = serde_json::from_str::(&content) { + if let Some(pid) = v["pid"].as_u64() { + if is_pid_alive(pid as u32) { + info!(pid, "router already running"); + return Ok(()); + } + info!(pid, "router.lock found but PID is dead, launching new router"); + } + } + } + } + + // Find hdcd-router binary next to this binary. + let self_exe = std::env::current_exe().context("cannot determine own executable path")?; + let self_dir = self_exe.parent().context("executable has no parent directory")?; + + let router_name = if cfg!(windows) { + "hdcd-router.exe" + } else { + "hdcd-router" + }; + let router_exe = self_dir.join(router_name); + + if !router_exe.exists() { + warn!( + path = %router_exe.display(), + "hdcd-router not found — start it manually or place it next to hdcd-telegram" + ); + return Ok(()); + } + + info!(path = %router_exe.display(), "launching hdcd-router"); + + let mut cmd = std::process::Command::new(&router_exe); + cmd.stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::inherit()); + + // Detach on Windows so the router survives parent exit. + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x00000200; + const DETACHED_PROCESS: u32 = 0x00000008; + cmd.creation_flags(CREATE_NEW_PROCESS_GROUP | DETACHED_PROCESS); + } + + let child = cmd.spawn().with_context(|| { + format!("failed to spawn {}", router_exe.display()) + })?; + info!(pid = child.id(), "hdcd-router spawned"); + + // Wait briefly for the router to write router.lock. + for _ in 0..20 { + std::thread::sleep(std::time::Duration::from_millis(250)); + if lock_path.exists() { + if let Ok(content) = std::fs::read_to_string(&lock_path) { + if let Ok(v) = serde_json::from_str::(&content) { + if v["pid"].as_u64().is_some() { + info!("hdcd-router is ready"); + return Ok(()); + } + } + } + } + } + + warn!("hdcd-router spawned but router.lock not yet written — proceeding anyway"); + Ok(()) +} + +/// Check if a PID is alive (used for router.lock validation). +fn is_pid_alive(pid: u32) -> bool { + sessions::SessionRegistry::is_pid_alive(pid) +} + /// Check the approved/ directory for pairing confirmations from the /// /telegram:access skill. async fn check_approvals(api: &api::BotApi, approved_dir: &std::path::Path) { @@ -416,7 +944,7 @@ async fn check_approvals(api: &api::BotApi, approved_dir: &std::path::Path) { let sender_id = entry.file_name().to_string_lossy().into_owned(); let path = entry.path(); match api - .send_message(&sender_id, "Paired! Say hi to Claude.", None, None, None) + .send_message(&sender_id, "Paired! Say hi to Claude.", None, None, None, None) .await { Ok(_) => { diff --git a/src/router/config.rs b/src/router/config.rs new file mode 100644 index 0000000..37fdb3c --- /dev/null +++ b/src/router/config.rs @@ -0,0 +1,133 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Router configuration — reads `config.json` from the router state directory. + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::Deserialize; + +/// Router configuration loaded from +/// `~/.claude/channels/telegram-router/config.json`. +#[derive(Debug, Clone, Deserialize)] +pub struct RouterConfig { + pub bot_token: String, + pub supergroup_id: String, + #[serde(default)] + pub allowed_users: Vec, + #[serde(default = "default_session_label_format")] + pub session_label_format: String, + #[serde(default = "default_close_on_disconnect")] + pub close_topic_on_disconnect: bool, + #[serde(default = "default_inbox_poll_ms")] + pub inbox_poll_interval_ms: u64, + #[serde(default = "default_outbox_poll_ms")] + pub outbox_poll_interval_ms: u64, + #[serde(default = "default_health_check_s")] + pub health_check_interval_s: u64, + #[serde(default = "default_auto_shutdown_s")] + pub auto_shutdown_delay_s: u64, +} + +fn default_session_label_format() -> String { + "{cwd_basename}".into() +} +fn default_close_on_disconnect() -> bool { + true +} +fn default_inbox_poll_ms() -> u64 { + 500 +} +fn default_outbox_poll_ms() -> u64 { + 200 +} +fn default_health_check_s() -> u64 { + 30 +} +fn default_auto_shutdown_s() -> u64 { + 60 +} + +/// Resolve the router state directory: `~/.claude/channels/telegram-router/`. +pub fn state_dir() -> Result { + let dir = if let Ok(d) = std::env::var("ROUTER_STATE_DIR") { + PathBuf::from(d) + } else { + dirs::home_dir() + .context("home dir unavailable")? + .join(".claude") + .join("channels") + .join("telegram-router") + }; + std::fs::create_dir_all(&dir) + .with_context(|| format!("create router state dir {}", dir.display()))?; + Ok(dir) +} + +/// Load router config from `config.json` in the given state directory. +pub fn load(state_dir: &Path) -> Result { + let path = state_dir.join("config.json"); + let raw = std::fs::read_to_string(&path) + .with_context(|| format!("read {}", path.display()))?; + let config: RouterConfig = serde_json::from_str(&raw) + .with_context(|| format!("parse {}", path.display()))?; + + if config.bot_token.is_empty() { + anyhow::bail!("bot_token is empty in {}", path.display()); + } + if config.supergroup_id.is_empty() { + anyhow::bail!("supergroup_id is empty in {}", path.display()); + } + + Ok(config) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn minimal_config_parses() { + let json = r#"{ + "bot_token": "123:AAHtest", + "supergroup_id": "-1001234567890" + }"#; + let config: RouterConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.bot_token, "123:AAHtest"); + assert_eq!(config.supergroup_id, "-1001234567890"); + assert!(config.allowed_users.is_empty()); + assert!(config.close_topic_on_disconnect); + assert_eq!(config.outbox_poll_interval_ms, 200); + assert_eq!(config.inbox_poll_interval_ms, 500); + assert_eq!(config.health_check_interval_s, 30); + assert_eq!(config.auto_shutdown_delay_s, 60); + } + + #[test] + fn full_config_parses() { + let json = r#"{ + "bot_token": "123:AAHtest", + "supergroup_id": "-1001234567890", + "allowed_users": ["123456789"], + "session_label_format": "{cwd_basename}", + "close_topic_on_disconnect": false, + "inbox_poll_interval_ms": 1000, + "outbox_poll_interval_ms": 100, + "health_check_interval_s": 60, + "auto_shutdown_delay_s": 0 + }"#; + let config: RouterConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.allowed_users, vec!["123456789"]); + assert!(!config.close_topic_on_disconnect); + assert_eq!(config.inbox_poll_interval_ms, 1000); + assert_eq!(config.auto_shutdown_delay_s, 0); + } + + #[test] + fn empty_token_rejected() { + let dir = tempfile::tempdir().unwrap(); + let json = r#"{"bot_token": "", "supergroup_id": "-100"}"#; + std::fs::write(dir.path().join("config.json"), json).unwrap(); + assert!(load(dir.path()).is_err()); + } +} diff --git a/src/router/mailbox.rs b/src/router/mailbox.rs new file mode 100644 index 0000000..ad52474 --- /dev/null +++ b/src/router/mailbox.rs @@ -0,0 +1,498 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Filesystem-based IPC via JSONL mailbox files. +//! +//! Each session has two files: +//! - `inbox/.jsonl` — Telegram → session (router writes, MCP reads) +//! - `outbox/.jsonl` — session → Telegram (MCP writes, router reads) +//! +//! Position tracking: a companion `.pos` file stores the byte offset of the +//! last-read position so that readers pick up only new lines after restart. + +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use tracing::warn; + +// --------------------------------------------------------------------------- +// Inbox message (Telegram → Session) +// --------------------------------------------------------------------------- + +/// A message written by the router into a session's inbox. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InboxMessage { + pub text: String, + pub user: String, + pub user_id: String, + pub chat_id: String, + pub message_id: i64, + pub ts: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub image_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub attachment_file_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub attachment_kind: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub attachment_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub attachment_mime: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub attachment_size: Option, +} + +// --------------------------------------------------------------------------- +// Outbox message (Session → Telegram) +// --------------------------------------------------------------------------- + +/// A message written by the MCP server into its outbox for the router. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct OutboxMessage { + pub text: String, + #[serde(default)] + pub reply_to: Option, + #[serde(default)] + pub files: Vec, + #[serde(default = "default_format")] + pub format: String, + /// If set, edit this message instead of sending a new one. + #[serde(skip_serializing_if = "Option::is_none")] + pub edit_message_id: Option, + /// If set, add this emoji reaction. + #[serde(skip_serializing_if = "Option::is_none")] + pub react_message_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub react_emoji: Option, + /// If set, rename the session's forum topic. `text` is ignored. + #[serde(skip_serializing_if = "Option::is_none")] + pub rename_to: Option, +} + +fn default_format() -> String { + "text".into() +} + +// --------------------------------------------------------------------------- +// Mailbox directories +// --------------------------------------------------------------------------- + +/// Ensure inbox and outbox directories exist under the state dir. +pub fn ensure_dirs(state_dir: &Path) -> Result<(PathBuf, PathBuf, PathBuf)> { + let inbox = state_dir.join("inbox"); + let outbox = state_dir.join("outbox"); + let register = state_dir.join("register"); + std::fs::create_dir_all(&inbox).context("create inbox dir")?; + std::fs::create_dir_all(&outbox).context("create outbox dir")?; + std::fs::create_dir_all(®ister).context("create register dir")?; + Ok((inbox, outbox, register)) +} + +// --------------------------------------------------------------------------- +// JSONL writer +// --------------------------------------------------------------------------- + +/// Append a single JSON line to a JSONL file. +/// +/// Opens in append mode so multiple writers don't clobber each other +/// (for lines < 4096 bytes, append writes are atomic on most OS/FS combos). +pub fn append_line(path: &Path, value: &impl Serialize) -> Result<()> { + let mut line = serde_json::to_vec(value).context("serialize JSONL line")?; + line.push(b'\n'); + + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .with_context(|| format!("open {} for append", path.display()))?; + + file.write_all(&line) + .with_context(|| format!("write to {}", path.display()))?; + + Ok(()) +} + +// --------------------------------------------------------------------------- +// JSONL reader with position tracking +// --------------------------------------------------------------------------- + +/// Peek new lines since the last recorded position without advancing it. +/// Returns `(end_offset, message)` pairs where `end_offset` is the byte +/// offset right after that line (including its trailing `\n`). +/// +/// Pair with [`commit_pos`] for at-least-once semantics: commit after a +/// message has been successfully processed so that a crash between read +/// and processing re-delivers unprocessed messages on restart. +/// +/// Malformed lines are warned and skipped; their bytes are folded into +/// the next valid message's `end_offset`, so committing that offset also +/// advances past the bad lines. Partial trailing lines (no `\n` yet) are +/// left for the next read. +pub fn peek_new_lines Deserialize<'de>>( + jsonl_path: &Path, + pos_path: &Path, +) -> Result> { + if !jsonl_path.exists() { + return Ok(Vec::new()); + } + + let start = read_pos(pos_path); + + let mut file = std::fs::File::open(jsonl_path) + .with_context(|| format!("open {}", jsonl_path.display()))?; + + let file_len = file.metadata().map(|m| m.len()).unwrap_or(0); + + if start >= file_len { + return Ok(Vec::new()); + } + + file.seek(SeekFrom::Start(start)) + .with_context(|| format!("seek {} to {start}", jsonl_path.display()))?; + + let mut buf = String::new(); + file.read_to_string(&mut buf) + .with_context(|| format!("read {}", jsonl_path.display()))?; + + let mut offset = start; + let mut messages = Vec::new(); + let mut parts = buf.split('\n').peekable(); + let mut line_idx = 0; + while let Some(line) = parts.next() { + // Last segment has no trailing `\n`. If it's non-empty the writer + // is mid-append — leave it for the next tick. + if parts.peek().is_none() { + break; + } + let line_end = offset + line.len() as u64 + 1; // +1 for the `\n` + let trimmed = line.trim(); + if !trimmed.is_empty() { + match serde_json::from_str::(trimmed) { + Ok(msg) => messages.push((line_end, msg)), + Err(e) => warn!( + file = %jsonl_path.display(), + line_offset = line_idx, + error = %e, + "skipping malformed JSONL line" + ), + } + } + offset = line_end; + line_idx += 1; + } + + Ok(messages) +} + +/// Advance the persisted read position. Use with [`peek_new_lines`] for +/// at-least-once semantics — commit after a message is successfully +/// processed so crashes don't silently drop unsent messages. +pub fn commit_pos(pos_path: &Path, pos: u64) { + write_pos(pos_path, pos); +} + +/// Read new lines and advance the position immediately (at-most-once, +/// fire-and-forget). Suitable for readers that don't need ACK semantics. +/// For at-least-once delivery use [`peek_new_lines`] + [`commit_pos`]. +pub fn read_new_lines Deserialize<'de>>( + jsonl_path: &Path, + pos_path: &Path, +) -> Result> { + let peeked = peek_new_lines::(jsonl_path, pos_path)?; + if let Some((last_offset, _)) = peeked.last() { + commit_pos(pos_path, *last_offset); + } + Ok(peeked.into_iter().map(|(_, m)| m).collect()) +} + +/// Path for the `.pos` companion file. +pub fn pos_path_for(jsonl_path: &Path) -> PathBuf { + let mut p = jsonl_path.as_os_str().to_owned(); + p.push(".pos"); + PathBuf::from(p) +} + +fn read_pos(pos_path: &Path) -> u64 { + std::fs::read_to_string(pos_path) + .ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0) +} + +fn write_pos(pos_path: &Path, pos: u64) { + let _ = std::fs::write(pos_path, pos.to_string()); +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn append_and_read_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("test.jsonl"); + let pos = pos_path_for(&jsonl); + + // Write two messages. + let m1 = InboxMessage { + text: "hello".into(), + user: "alice".into(), + user_id: "1".into(), + chat_id: "1".into(), + message_id: 100, + ts: "2026-04-11T15:00:00Z".into(), + image_path: None, + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + let m2 = InboxMessage { + text: "world".into(), + user: "bob".into(), + user_id: "2".into(), + chat_id: "2".into(), + message_id: 101, + ts: "2026-04-11T15:01:00Z".into(), + image_path: None, + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + append_line(&jsonl, &m1).unwrap(); + append_line(&jsonl, &m2).unwrap(); + + // Read all — should get both. + let msgs: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0].text, "hello"); + assert_eq!(msgs[1].text, "world"); + + // Read again — nothing new. + let msgs2: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert!(msgs2.is_empty()); + + // Append a third, read only new. + let m3 = InboxMessage { + text: "third".into(), + user: "charlie".into(), + user_id: "3".into(), + chat_id: "3".into(), + message_id: 102, + ts: "2026-04-11T15:02:00Z".into(), + image_path: None, + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + append_line(&jsonl, &m3).unwrap(); + let msgs3: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(msgs3.len(), 1); + assert_eq!(msgs3[0].text, "third"); + } + + #[test] + fn read_nonexistent_file_returns_empty() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("missing.jsonl"); + let pos = pos_path_for(&jsonl); + let msgs: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert!(msgs.is_empty()); + } + + #[test] + fn malformed_lines_skipped() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("bad.jsonl"); + let pos = pos_path_for(&jsonl); + + // Write valid, invalid, valid. + let m1 = InboxMessage { + text: "good1".into(), + user: "a".into(), + user_id: "1".into(), + chat_id: "1".into(), + message_id: 1, + ts: "t".into(), + image_path: None, + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + append_line(&jsonl, &m1).unwrap(); + + // Append raw garbage. + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&jsonl) + .unwrap(); + f.write_all(b"NOT JSON\n").unwrap(); + + let m2 = InboxMessage { + text: "good2".into(), + user: "b".into(), + user_id: "2".into(), + chat_id: "2".into(), + message_id: 2, + ts: "t".into(), + image_path: None, + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + append_line(&jsonl, &m2).unwrap(); + + let msgs: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0].text, "good1"); + assert_eq!(msgs[1].text, "good2"); + } + + #[test] + fn outbox_message_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("outbox.jsonl"); + let pos = pos_path_for(&jsonl); + + let msg = OutboxMessage { + text: "response".into(), + reply_to: Some(456), + files: vec!["/tmp/img.png".into()], + format: "text".into(), + edit_message_id: None, + react_message_id: None, + react_emoji: None, + rename_to: None, + }; + append_line(&jsonl, &msg).unwrap(); + + let msgs: Vec = read_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0].text, "response"); + assert_eq!(msgs[0].reply_to, Some(456)); + assert_eq!(msgs[0].files, vec!["/tmp/img.png"]); + } + + fn make_outbox(text: &str) -> OutboxMessage { + OutboxMessage { + text: text.into(), + reply_to: None, + files: Vec::new(), + format: "text".into(), + edit_message_id: None, + react_message_id: None, + react_emoji: None, + rename_to: None, + } + } + + #[test] + fn peek_does_not_advance_pos() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("outbox.jsonl"); + let pos = pos_path_for(&jsonl); + + append_line(&jsonl, &make_outbox("one")).unwrap(); + append_line(&jsonl, &make_outbox("two")).unwrap(); + append_line(&jsonl, &make_outbox("three")).unwrap(); + + // First peek returns all three. + let peeked: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked.len(), 3); + assert!(peeked[0].0 < peeked[1].0); + assert!(peeked[1].0 < peeked[2].0); + + // Peek again with no commit — still all three, proving pos + // didn't advance. + let peeked2: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked2.len(), 3); + } + + #[test] + fn commit_per_message_enables_at_least_once() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("outbox.jsonl"); + let pos = pos_path_for(&jsonl); + + append_line(&jsonl, &make_outbox("one")).unwrap(); + append_line(&jsonl, &make_outbox("two")).unwrap(); + append_line(&jsonl, &make_outbox("three")).unwrap(); + + // Process "one", commit. Simulate crash before "two". + let peeked: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked[0].1.text, "one"); + commit_pos(&pos, peeked[0].0); + + // Next tick after "crash" — "two" and "three" still visible. + let after_crash: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(after_crash.len(), 2); + assert_eq!(after_crash[0].1.text, "two"); + assert_eq!(after_crash[1].1.text, "three"); + } + + #[test] + fn peek_skips_malformed_via_next_offset() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("outbox.jsonl"); + let pos = pos_path_for(&jsonl); + + append_line(&jsonl, &make_outbox("first")).unwrap(); + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&jsonl) + .unwrap(); + f.write_all(b"NOT JSON\n").unwrap(); + append_line(&jsonl, &make_outbox("third")).unwrap(); + + let peeked: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked.len(), 2); + assert_eq!(peeked[0].1.text, "first"); + assert_eq!(peeked[1].1.text, "third"); + + // Committing the "third" offset also skips the bad line. + commit_pos(&pos, peeked[1].0); + let after: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert!(after.is_empty()); + } + + #[test] + fn peek_leaves_partial_trailing_line() { + let dir = tempfile::tempdir().unwrap(); + let jsonl = dir.path().join("outbox.jsonl"); + let pos = pos_path_for(&jsonl); + + append_line(&jsonl, &make_outbox("complete")).unwrap(); + // Simulate a writer mid-append: partial JSON, no trailing \n. + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&jsonl) + .unwrap(); + f.write_all(br#"{"text":"partial""#).unwrap(); + + let peeked: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked.len(), 1); + assert_eq!(peeked[0].1.text, "complete"); + + // Commit first message; now finish the partial line and peek again. + commit_pos(&pos, peeked[0].0); + f.write_all(b",\"files\":[],\"format\":\"text\"}\n").unwrap(); + + let peeked2: Vec<(u64, OutboxMessage)> = peek_new_lines(&jsonl, &pos).unwrap(); + assert_eq!(peeked2.len(), 1); + assert_eq!(peeked2[0].1.text, "partial"); + } +} diff --git a/src/router/main.rs b/src/router/main.rs new file mode 100644 index 0000000..635b2c7 --- /dev/null +++ b/src/router/main.rs @@ -0,0 +1,928 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! `hdcd-router` — standalone process that holds the single Telegram +//! polling connection and routes messages between forum topics and +//! Claude Code sessions via filesystem IPC. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use tokio::sync::Mutex; +use tracing::{debug, error, info, warn}; + +use hdcd_telegram::router::{config, mailbox, sessions, topics}; +use hdcd_telegram::telegram::{api, polling, types}; + +/// Short timeout for the shutdown "Router stopped" announcement so a flaky +/// network can't wedge the process during exit. +const SHUTDOWN_SEND_TIMEOUT_SECS: u64 = 5; + +/// Filename of the OS-level exclusive lock guard, kept next to `router.lock`. +/// The OS releases this lock on any process exit — normal, crash, kill, or +/// reboot — so there is no PID-reuse race or stale-heartbeat window. +const LOCK_GUARD_FILE: &str = "router.guard"; + +/// Shared state protected by a mutex for the update handler and outbox poller. +struct RouterState { + registry: sessions::SessionRegistry, + topic_mgr: topics::TopicManager, + inbox_dir: PathBuf, + outbox_dir: PathBuf, + register_dir: PathBuf, + config: config::RouterConfig, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_writer(std::io::stderr) + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "hdcd_router=info,hdcd_telegram=info".into()), + ) + .init(); + + let start_time = std::time::Instant::now(); + + let sd = config::state_dir()?; + + // Refuse to start if another router is already alive. Two routers sharing + // the same bot token both long-poll getUpdates and ping-pong 409 Conflict + // forever, so neither delivers messages. The OS releases the lock on any + // process termination (clean exit, crash, SIGKILL, reboot), so there is + // no stale-lock window to wait out. + let _lock_guard = match acquire_lock_guard(&sd) { + Ok(g) => g, + Err(e) => { + error!( + error = %e, + "another hdcd-router is already running — refusing to start a second instance" + ); + return Ok(()); + } + }; + + let cfg = config::load(&sd).with_context(|| { + format!( + "failed to load config from {}/config.json", + sd.display() + ) + })?; + + info!( + supergroup = %cfg.supergroup_id, + allowed_users = ?cfg.allowed_users, + "router config loaded" + ); + + let (inbox_dir, outbox_dir, register_dir) = mailbox::ensure_dirs(&sd)?; + + let bot_api = Arc::new(api::BotApi::new(&cfg.bot_token)); + + // Verify bot token. + let me = bot_api + .get_me() + .await + .context("getMe failed — check bot_token in config.json")?; + let bot_username = me.username.unwrap_or_default(); + info!(username = %bot_username, "telegram bot identified"); + + let registry = sessions::SessionRegistry::load(&sd); + let topic_mgr = topics::TopicManager::new(Arc::clone(&bot_api), &cfg); + + let state = Arc::new(Mutex::new(RouterState { + registry, + topic_mgr, + inbox_dir, + outbox_dir: outbox_dir.clone(), + register_dir: register_dir.clone(), + config: cfg.clone(), + })); + + // Reconcile stale sessions from a previous run. + reconcile_sessions(&state).await; + + // Cancellation token for clean shutdown. + let cancel = tokio_util::sync::CancellationToken::new(); + + // Handle Ctrl+C. + let shutdown_cancel = cancel.clone(); + tokio::spawn(async move { + match tokio::signal::ctrl_c().await { + Ok(()) => { + info!("received Ctrl+C, shutting down"); + shutdown_cancel.cancel(); + } + Err(e) => error!(error = %e, "failed to listen for Ctrl+C"), + } + }); + + // Start polling loop. + let (update_tx, mut update_rx) = tokio::sync::mpsc::channel::(64); + let poll_api = Arc::clone(&bot_api); + let poll_cancel = cancel.clone(); + let poll_handle = tokio::spawn(async move { + if let Err(e) = polling::run(poll_api, update_tx, poll_cancel).await { + error!(error = %e, "polling loop exited with error"); + } + }); + + // Spawn registration watcher — polls register/ directory. + let reg_state = Arc::clone(&state); + let reg_cancel = cancel.clone(); + tokio::spawn(async move { + let mut interval = + tokio::time::interval(std::time::Duration::from_secs(2)); + loop { + tokio::select! { + _ = interval.tick() => {} + _ = reg_cancel.cancelled() => return, + } + if let Err(e) = process_registrations(®_state).await { + warn!(error = %e, "registration scan failed"); + } + } + }); + + // Spawn outbox poller — reads outbox files and sends to Telegram topics. + let outbox_state = Arc::clone(&state); + let outbox_api = Arc::clone(&bot_api); + let outbox_cancel = cancel.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_millis( + cfg.outbox_poll_interval_ms, + )); + loop { + tokio::select! { + _ = interval.tick() => {} + _ = outbox_cancel.cancelled() => return, + } + poll_outbox(&outbox_state, &outbox_api).await; + } + }); + + // Spawn health checker — detects dead PIDs and closes their topics. + let health_state = Arc::clone(&state); + let health_cancel = cancel.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs( + cfg.health_check_interval_s, + )); + loop { + tokio::select! { + _ = interval.tick() => {} + _ = health_cancel.cancelled() => return, + } + check_session_health(&health_state).await; + } + }); + + // Spawn idle shutdown watcher — exits when no active sessions for grace period. + let idle_state = Arc::clone(&state); + let idle_cancel = cancel.clone(); + let auto_shutdown_delay = cfg.auto_shutdown_delay_s; + if auto_shutdown_delay > 0 { + tokio::spawn(async move { + let mut idle_since: Option = None; + let mut interval = + tokio::time::interval(std::time::Duration::from_secs(10)); + // Skip the first immediate tick. + interval.tick().await; + loop { + tokio::select! { + _ = interval.tick() => {} + _ = idle_cancel.cancelled() => return, + } + let s = idle_state.lock().await; + let has_active = !s.registry.active_sessions().is_empty(); + drop(s); + + if has_active { + idle_since = None; + } else { + let since = *idle_since.get_or_insert_with(tokio::time::Instant::now); + if since.elapsed().as_secs() >= auto_shutdown_delay { + info!( + idle_secs = since.elapsed().as_secs(), + "no active sessions, auto-shutting down" + ); + idle_cancel.cancel(); + return; + } + } + } + }); + } + + // Spawn heartbeat writer — updates router.lock every 30s. + let heartbeat_sd = sd.clone(); + let heartbeat_cancel = cancel.clone(); + write_heartbeat(&heartbeat_sd); // initial write + tokio::spawn(async move { + let mut interval = + tokio::time::interval(std::time::Duration::from_secs(30)); + loop { + tokio::select! { + _ = interval.tick() => {} + _ = heartbeat_cancel.cancelled() => return, + } + write_heartbeat(&heartbeat_sd); + } + }); + + info!("router started — listening for updates"); + + // Announce startup in General topic. + let startup_msg = format!( + "\u{1f7e2} Router started \u{00b7} v{} \u{00b7} pid {} \u{00b7} built {}", + env!("CARGO_PKG_VERSION"), + std::process::id(), + exe_build_time(), + ); + let _ = bot_api + .send_message(&cfg.supergroup_id, &startup_msg, None, None, None, None) + .await; + + // Main loop: process incoming Telegram updates. + let process_cancel = cancel.clone(); + tokio::select! { + _ = async { + while let Some(update) = update_rx.recv().await { + if let Err(e) = handle_update(&update, &state).await { + warn!(error = %e, "failed to handle update"); + } + } + } => {} + _ = process_cancel.cancelled() => { + info!("update processor cancelled"); + } + } + + // Shutdown — don't close active sessions. If their PIDs are still + // alive they will be picked up on the next router start by reconcile. + cancel.cancel(); + + // Announce shutdown in General topic. Bounded so a wedged HTTP client + // can't keep the process alive past shutdown. + let shutdown_msg = format!( + "\u{1f534} Router stopped \u{00b7} pid {} \u{00b7} uptime {}", + std::process::id(), + format_uptime(start_time.elapsed()), + ); + let _ = tokio::time::timeout( + std::time::Duration::from_secs(SHUTDOWN_SEND_TIMEOUT_SECS), + bot_api.send_message(&cfg.supergroup_id, &shutdown_msg, None, None, None, None), + ) + .await; + + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), poll_handle).await; + + // Remove heartbeat file. + let _ = std::fs::remove_file(sd.join("router.lock")); + + info!("router stopped"); + + // Force exit: if any spawned task lingers (detached blocking I/O, HTTP + // client background thread, etc.) the runtime drop can hang, leaving a + // zombie process whose heartbeat task is already gone but whose lock + // file gets stale. Exit 0 unconditionally after cleanup. + std::process::exit(0); +} + +// --------------------------------------------------------------------------- +// Inbound: Telegram → inbox +// --------------------------------------------------------------------------- + +/// Handle an incoming Telegram update: route supergroup topic messages +/// to the correct session's inbox file. +async fn handle_update( + update: &types::Update, + state: &Arc>, +) -> Result<()> { + let msg = match &update.message { + Some(m) => m, + None => return Ok(()), + }; + + let s = state.lock().await; + let chat_id = msg.chat.id.to_string(); + + // Only handle messages from the configured supergroup. + if chat_id != s.config.supergroup_id { + // Check allowed users for DMs (future: could be used for admin commands). + debug!(chat_id, "ignoring message from non-supergroup chat"); + return Ok(()); + } + + // Check if sender is allowed. + let sender_id = msg.from.as_ref().map(|u| u.id.to_string()); + let is_allowed = s.config.allowed_users.is_empty() + || sender_id + .as_ref() + .map(|id| s.config.allowed_users.contains(id)) + .unwrap_or(false); + + if !is_allowed { + debug!("message from non-allowed user, ignoring"); + return Ok(()); + } + + // Determine which topic the message belongs to. + let thread_id = match msg.message_thread_id { + Some(tid) => tid, + None => { + // Message in General topic — handle as router command. + let api = Arc::clone(s.topic_mgr.api()); + let supergroup_id = s.config.supergroup_id.clone(); + let text = msg.text.as_deref().unwrap_or("").to_string(); + + // Collect active sessions as owned data before releasing lock. + let active: Vec<(String, sessions::SessionEntry)> = s + .registry + .active_sessions() + .into_iter() + .map(|(id, e)| (id.to_string(), e.clone())) + .collect(); + + drop(s); // release lock before I/O + + // /kill needs mutable access to state — handle separately. + if text.starts_with("/kill ") { + let target = text.strip_prefix("/kill ").unwrap().trim(); + let reply = handle_kill_command(target, state).await; + let _ = api + .send_message(&supergroup_id, &reply, None, None, None, None) + .await; + } else if let Some(reply) = handle_general_command(&text, &active) { + let _ = api + .send_message(&supergroup_id, &reply, None, None, None, None) + .await; + } + + return Ok(()); + } + }; + + // Find the session for this topic. + let session_id = match s.registry.session_by_topic(thread_id) { + Some(id) => id.to_string(), + None => { + debug!(thread_id, "no session mapped to this topic"); + return Ok(()); + } + }; + + let from = msg.from.as_ref(); + let user = from + .and_then(|u| u.username.as_deref()) + .unwrap_or("?"); + let user_id = from + .map(|u| u.id.to_string()) + .unwrap_or_default(); + let text = msg + .text + .as_deref() + .or(msg.caption.as_deref()) + .unwrap_or("") + .to_string(); + let ts = chrono::DateTime::from_timestamp(msg.date, 0) + .unwrap_or_default() + .to_rfc3339(); + + let inbox_msg = mailbox::InboxMessage { + text, + user: user.to_string(), + user_id, + chat_id: chat_id.clone(), + message_id: msg.message_id, + ts, + image_path: None, // TODO Phase 3+: download photos + attachment_file_id: None, + attachment_kind: None, + attachment_name: None, + attachment_mime: None, + attachment_size: None, + }; + + let inbox_path = s.inbox_dir.join(format!("{session_id}.jsonl")); + mailbox::append_line(&inbox_path, &inbox_msg)?; + + info!( + session_id, + thread_id, + from = user, + "routed message to inbox" + ); + + Ok(()) +} + +// --------------------------------------------------------------------------- +// General topic commands +// --------------------------------------------------------------------------- + +/// Handle a text command sent in the General topic. +/// Returns `Some(reply)` for recognized commands, `None` otherwise. +fn handle_general_command( + text: &str, + active: &[(String, sessions::SessionEntry)], +) -> Option { + match text.trim() { + "/status" => { + if active.is_empty() { + return Some("No active sessions.".to_string()); + } + let mut lines = vec!["Active sessions:".to_string()]; + for (id, entry) in active { + let pid = entry + .pid + .map(|p| format!(" (PID {p})")) + .unwrap_or_default(); + let cwd = entry + .cwd + .as_deref() + .unwrap_or("?"); + lines.push(format!( + "• {}{pid}\n topic {} · {}", + entry.label, entry.topic_id, cwd + )); + lines.push(format!(" id: {id}")); + } + Some(lines.join("\n")) + } + "/help" => Some( + "Router commands:\n\ + /status — list active sessions\n\ + /kill — close a session\n\ + /help — show this help" + .to_string(), + ), + _ => None, + } +} + +/// Handle `/kill ` — needs mutable state access. +async fn handle_kill_command( + target: &str, + state: &Arc>, +) -> String { + let mut s = state.lock().await; + + let has_topic = s.registry.topic_by_session(target).is_some(); + if !has_topic { + return format!("Session not found: {target}"); + } + + let RouterState { + topic_mgr, + registry, + .. + } = &mut *s; + topic_mgr.close_topic(target, registry).await; + + format!("Session killed: {target}") +} + +// --------------------------------------------------------------------------- +// Outbound: outbox → Telegram +// --------------------------------------------------------------------------- + +/// Poll all outbox files and send pending messages to their topics. +async fn poll_outbox( + state: &Arc>, + api: &Arc, +) { + let s = state.lock().await; + + // Collect active sessions to iterate. + let active: Vec<(String, i64)> = s + .registry + .active_sessions() + .iter() + .map(|(id, e)| (id.to_string(), e.topic_id)) + .collect(); + + let outbox_dir = s.outbox_dir.clone(); + let supergroup_id = s.config.supergroup_id.clone(); + drop(s); // release lock before I/O + + for (session_id, topic_id) in active { + let outbox_path = outbox_dir.join(format!("{session_id}.jsonl")); + let pos_path = mailbox::pos_path_for(&outbox_path); + + // Peek — pos is committed per-message after successful delivery so + // a crash between read and send doesn't silently drop messages. + let messages: Vec<(u64, mailbox::OutboxMessage)> = + match mailbox::peek_new_lines(&outbox_path, &pos_path) { + Ok(m) => m, + Err(e) => { + warn!(session_id, error = %e, "failed to read outbox"); + continue; + } + }; + + for (end_offset, msg) in messages { + // Rename requests take a different path — they need mutable access + // to the registry + TopicManager, not the raw Telegram API. + if let Some(ref new_title) = msg.rename_to { + let mut s = state.lock().await; + let RouterState { topic_mgr, registry, .. } = &mut *s; + topic_mgr + .rename_topic(&session_id, new_title, registry) + .await; + drop(s); + mailbox::commit_pos(&pos_path, end_offset); + continue; + } + + match deliver_outbox_message(api, &supergroup_id, topic_id, &msg).await { + Ok(()) => mailbox::commit_pos(&pos_path, end_offset), + Err(e) => { + warn!( + session_id, + error = %e, + "failed to deliver outbox message — will retry next tick" + ); + // Don't advance pos; re-read this message on the next + // poll. Stop processing this session's batch to avoid + // reordering past the stuck message. + break; + } + } + } + } +} + +/// Deliver a single outbox message to a Telegram forum topic. +async fn deliver_outbox_message( + api: &api::BotApi, + supergroup_id: &str, + topic_id: i64, + msg: &mailbox::OutboxMessage, +) -> Result<()> { + // Handle reactions. + if let (Some(react_msg_id), Some(ref emoji)) = + (msg.react_message_id, &msg.react_emoji) + { + api.set_message_reaction(supergroup_id, react_msg_id, emoji) + .await?; + return Ok(()); + } + + // Handle edits. + if let Some(edit_id) = msg.edit_message_id { + let parse_mode = if msg.format == "markdownv2" { + Some("MarkdownV2") + } else { + None + }; + api.edit_message_text(supergroup_id, edit_id, &msg.text, parse_mode) + .await?; + return Ok(()); + } + + // Regular message send. + let parse_mode = if msg.format == "markdownv2" { + Some("MarkdownV2") + } else { + None + }; + api.send_message( + supergroup_id, + &msg.text, + msg.reply_to, + parse_mode, + None, + Some(topic_id), + ) + .await?; + + // TODO: handle msg.files (send_photo / send_document) + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Registration watcher +// --------------------------------------------------------------------------- + +/// Scan the register/ directory for new or updated registration files. +async fn process_registrations( + state: &Arc>, +) -> Result<()> { + let register_dir = { + let s = state.lock().await; + s.register_dir.clone() + }; + + let entries = match std::fs::read_dir(®ister_dir) { + Ok(e) => e, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(e) => return Err(e.into()), + }; + + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + + let raw = match std::fs::read_to_string(&path) { + Ok(r) => r, + Err(e) => { + warn!(path = %path.display(), error = %e, "failed to read registration"); + continue; + } + }; + + let reg: sessions::Registration = match serde_json::from_str(&raw) { + Ok(r) => r, + Err(e) => { + warn!(path = %path.display(), error = %e, "malformed registration"); + continue; + } + }; + + if reg.disconnected { + // Session is disconnecting — close its topic. + let mut s = state.lock().await; + let has_topic = s.registry.topic_by_session(®.session_id).is_some(); + if has_topic { + info!(session_id = %reg.session_id, "session disconnected"); + let RouterState { topic_mgr, registry, .. } = &mut *s; + topic_mgr.close_topic(®.session_id, registry).await; + } + drop(s); + if let Err(e) = std::fs::remove_file(&path) { + warn!( + path = %path.display(), + error = %e, + "failed to remove disconnect marker" + ); + } + continue; + } + + // Check if already registered. + { + let s = state.lock().await; + if s.registry.topic_by_session(®.session_id).is_some() { + continue; // already registered, skip + } + } + + // Skip stale registrations whose MCP process is already dead. Creating + // a topic just to have the health checker close it again seconds later + // spams the supergroup on router restart. + if let Some(pid) = reg.pid { + if !sessions::SessionRegistry::is_pid_alive(pid) { + info!( + session_id = %reg.session_id, + pid, + "dropping stale registration — MCP PID is dead" + ); + if let Err(e) = std::fs::remove_file(&path) { + warn!( + path = %path.display(), + error = %e, + "failed to remove stale registration" + ); + } + continue; + } + } + + // New session — create/reopen topic. + info!( + session_id = %reg.session_id, + label = %reg.label, + pid = ?reg.pid, + "new session registration" + ); + + let mut s = state.lock().await; + let RouterState { topic_mgr, registry, .. } = &mut *s; + match topic_mgr + .ensure_topic(®.session_id, ®, registry) + .await + { + Ok(topic_id) => { + info!( + session_id = %reg.session_id, + topic_id, + "session registered" + ); + // Clean up registration file after successful processing. + if let Err(e) = std::fs::remove_file(&path) { + warn!( + path = %path.display(), + error = %e, + "failed to remove registration file" + ); + } + } + Err(e) => { + error!( + session_id = %reg.session_id, + error = %e, + "failed to create topic for session" + ); + } + } + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Health checker +// --------------------------------------------------------------------------- + +/// Reconcile sessions on startup — close any sessions whose PIDs died +/// while the router was not running. +async fn reconcile_sessions(state: &Arc>) { + let stale: Vec = { + let s = state.lock().await; + s.registry + .active_sessions() + .iter() + .filter(|(_, entry)| { + entry + .pid + .map(|pid| !sessions::SessionRegistry::is_pid_alive(pid)) + .unwrap_or(false) + }) + .map(|(id, _)| id.to_string()) + .collect() + }; + + if stale.is_empty() { + info!("no stale sessions to reconcile"); + return; + } + + info!(count = stale.len(), "reconciling stale sessions from previous run"); + + for session_id in &stale { + let mut s = state.lock().await; + let RouterState { topic_mgr, registry, .. } = &mut *s; + info!(session_id, "closing stale session (PID dead)"); + topic_mgr.close_topic(session_id, registry).await; + } +} + +/// Format the build time as the mtime of the current executable, in local +/// time with a UTC offset so the reader isn't guessing which timezone. +fn exe_build_time() -> String { + std::env::current_exe() + .ok() + .and_then(|p| std::fs::metadata(p).ok()) + .and_then(|m| m.modified().ok()) + .map(|t| { + chrono::DateTime::::from(t) + .format("%Y-%m-%d %H:%M %:z") + .to_string() + }) + .unwrap_or_else(|| "unknown".into()) +} + +/// Human-friendly uptime: `3h 42m`, `42m`, or `17s`. +fn format_uptime(d: std::time::Duration) -> String { + let s = d.as_secs(); + let h = s / 3600; + let m = (s % 3600) / 60; + if h > 0 { + format!("{h}h {m}m") + } else if m > 0 { + format!("{m}m") + } else { + format!("{s}s") + } +} + +/// Write a heartbeat file so MCP servers can detect a live router. +fn write_heartbeat(state_dir: &Path) { + let lock_path = state_dir.join("router.lock"); + let pid = std::process::id(); + let ts = chrono::Utc::now().to_rfc3339(); + let content = format!("{{\"pid\":{pid},\"heartbeat\":\"{ts}\"}}\n"); + let _ = std::fs::write(&lock_path, content); +} + +/// Try to acquire the OS-level exclusive lock on `router.guard`. The returned +/// `File` must be kept alive for the lifetime of the process — when it is +/// dropped (or the process exits for any reason, including SIGKILL or power +/// loss) the OS releases the lock and the next router can start immediately. +/// No PID-reuse race, no heartbeat staleness heuristic. +fn acquire_lock_guard(state_dir: &Path) -> Result { + use fs4::fs_std::FileExt; + + let guard_path = state_dir.join(LOCK_GUARD_FILE); + let file = std::fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(&guard_path) + .with_context(|| format!("open lock guard {}", guard_path.display()))?; + + let locked = file + .try_lock_exclusive() + .with_context(|| format!("lock {}", guard_path.display()))?; + if !locked { + anyhow::bail!("lock held by another process"); + } + + Ok(file) +} + +/// Check if active sessions' PIDs are still alive. Close topics for dead ones. +async fn check_session_health(state: &Arc>) { + let dead_sessions: Vec = { + let s = state.lock().await; + s.registry + .active_sessions() + .iter() + .filter(|(_, entry)| { + entry + .pid + .map(|pid| !sessions::SessionRegistry::is_pid_alive(pid)) + .unwrap_or(false) + }) + .map(|(id, _)| id.to_string()) + .collect() + }; + + for session_id in dead_sessions { + info!(session_id, "session PID dead, closing topic"); + let mut s = state.lock().await; + let RouterState { topic_mgr, registry, .. } = &mut *s; + topic_mgr.close_topic(&session_id, registry).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_entry(label: &str, pid: Option, cwd: Option<&str>) -> sessions::SessionEntry { + sessions::SessionEntry { + topic_id: 42, + label: label.to_string(), + pid, + cwd: cwd.map(|s| s.to_string()), + state: sessions::SessionState::Active, + registered_at: "2026-04-11T15:00:00Z".to_string(), + closed_at: None, + title: None, + } + } + + #[test] + fn status_no_sessions() { + let active: Vec<(String, sessions::SessionEntry)> = vec![]; + let reply = handle_general_command("/status", &active); + assert_eq!(reply, Some("No active sessions.".to_string())); + } + + #[test] + fn status_with_sessions() { + let active = vec![ + ("sess-1".to_string(), make_entry("VS Code: project", Some(1234), Some("/home/user/project"))), + ("sess-2".to_string(), make_entry("CLI: sweep", None, None)), + ]; + let reply = handle_general_command("/status", &active).unwrap(); + assert!(reply.contains("VS Code: project")); + assert!(reply.contains("PID 1234")); + assert!(reply.contains("sess-1")); + assert!(reply.contains("CLI: sweep")); + assert!(reply.contains("sess-2")); + } + + #[test] + fn help_command() { + let reply = handle_general_command("/help", &[]).unwrap(); + assert!(reply.contains("/status")); + assert!(reply.contains("/kill")); + assert!(reply.contains("/help")); + } + + #[test] + fn unknown_command_returns_none() { + assert_eq!(handle_general_command("hello", &[]), None); + assert_eq!(handle_general_command("/unknown", &[]), None); + } + + #[test] + fn heartbeat_writes_file() { + let dir = tempfile::tempdir().unwrap(); + write_heartbeat(dir.path()); + let lock_path = dir.path().join("router.lock"); + assert!(lock_path.exists()); + let content = std::fs::read_to_string(&lock_path).unwrap(); + let v: serde_json::Value = serde_json::from_str(&content).unwrap(); + assert!(v["pid"].is_number()); + assert!(v["heartbeat"].is_string()); + } +} diff --git a/src/router/mod.rs b/src/router/mod.rs new file mode 100644 index 0000000..ad82437 --- /dev/null +++ b/src/router/mod.rs @@ -0,0 +1,9 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Router module — multiplexes multiple Claude Code sessions through +//! forum topics in a single Telegram supergroup. + +pub mod config; +pub mod mailbox; +pub mod sessions; +pub mod topics; diff --git a/src/router/sessions.rs b/src/router/sessions.rs new file mode 100644 index 0000000..ffa1bf3 --- /dev/null +++ b/src/router/sessions.rs @@ -0,0 +1,263 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Session registry — tracks active Claude Code sessions and their +//! associated forum topics. Persisted to `sessions.json`. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use tracing::warn; + +/// State of a session in the registry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SessionState { + Active, + Closed, +} + +/// A single session entry in the registry. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionEntry { + pub topic_id: i64, + pub label: String, + pub pid: Option, + pub cwd: Option, + pub state: SessionState, + pub registered_at: String, + pub closed_at: Option, + /// Last rendered topic name. Lets the router skip redundant + /// `editForumTopic` calls when the title hasn't changed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, +} + +/// Registration request written by an MCP server to `register/.json`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Registration { + pub session_id: String, + pub label: String, + pub pid: Option, + pub cwd: Option, + pub registered_at: String, + #[serde(default)] + pub disconnected: bool, +} + +/// In-memory session registry backed by `sessions.json`. +#[derive(Debug)] +pub struct SessionRegistry { + path: PathBuf, + pub sessions: HashMap, +} + +impl SessionRegistry { + /// Load from disk or start empty. + pub fn load(state_dir: &Path) -> Self { + let path = state_dir.join("sessions.json"); + let sessions = match std::fs::read_to_string(&path) { + Ok(raw) => match serde_json::from_str(&raw) { + Ok(map) => map, + Err(e) => { + warn!(error = %e, "sessions.json corrupt, starting fresh"); + HashMap::new() + } + }, + Err(_) => HashMap::new(), + }; + Self { path, sessions } + } + + /// Persist current state to disk. + pub fn save(&self) { + let tmp = format!("{}.tmp", self.path.display()); + match serde_json::to_string_pretty(&self.sessions) { + Ok(json) => { + if std::fs::write(&tmp, format!("{json}\n")).is_ok() { + let _ = std::fs::rename(&tmp, &self.path); + } + } + Err(e) => warn!(error = %e, "failed to serialize sessions.json"), + } + } + + /// Register a new session or update an existing one. + pub fn register(&mut self, session_id: &str, topic_id: i64, reg: &Registration) { + self.sessions.insert( + session_id.to_string(), + SessionEntry { + topic_id, + label: reg.label.clone(), + pid: reg.pid, + cwd: reg.cwd.clone(), + state: SessionState::Active, + registered_at: reg.registered_at.clone(), + closed_at: None, + title: None, + }, + ); + self.save(); + } + + /// Record a new topic title for a session. Returns `true` if the title + /// actually changed (caller should then call `editForumTopic`). + pub fn set_title(&mut self, session_id: &str, title: &str) -> bool { + let entry = match self.sessions.get_mut(session_id) { + Some(e) => e, + None => return false, + }; + if entry.title.as_deref() == Some(title) { + return false; + } + entry.title = Some(title.to_string()); + self.save(); + true + } + + /// Mark a session as closed. + pub fn close(&mut self, session_id: &str) { + if let Some(entry) = self.sessions.get_mut(session_id) { + entry.state = SessionState::Closed; + entry.closed_at = Some(chrono::Utc::now().to_rfc3339()); + self.save(); + } + } + + /// Find session_id by topic_id. + pub fn session_by_topic(&self, topic_id: i64) -> Option<&str> { + self.sessions + .iter() + .find(|(_, e)| e.topic_id == topic_id && e.state == SessionState::Active) + .map(|(id, _)| id.as_str()) + } + + /// Find topic_id by session_id. + pub fn topic_by_session(&self, session_id: &str) -> Option { + self.sessions.get(session_id).map(|e| e.topic_id) + } + + /// List active sessions. + pub fn active_sessions(&self) -> Vec<(&str, &SessionEntry)> { + self.sessions + .iter() + .filter(|(_, e)| e.state == SessionState::Active) + .map(|(id, e)| (id.as_str(), e)) + .collect() + } + + /// Check if a PID is still alive (platform-specific). + pub fn is_pid_alive(pid: u32) -> bool { + #[cfg(unix)] + { + // kill(pid, 0) checks existence without sending a signal. + unsafe { libc::kill(pid as i32, 0) == 0 } + } + #[cfg(windows)] + { + use windows_sys::Win32::System::Threading::{ + GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, + }; + use windows_sys::Win32::Foundation::CloseHandle; + const STILL_ACTIVE: u32 = 259; + let handle = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) }; + if handle.is_null() { + return false; + } + let mut exit_code: u32 = 0; + let ok = unsafe { GetExitCodeProcess(handle, &mut exit_code) }; + unsafe { CloseHandle(handle) }; + ok != 0 && exit_code == STILL_ACTIVE + } + #[cfg(not(any(unix, windows)))] + { + let _ = pid; + true // assume alive on unknown platforms + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registry_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + assert!(reg.sessions.is_empty()); + + let registration = Registration { + session_id: "abc-123".into(), + label: "VS Code: test".into(), + pid: Some(12345), + cwd: Some("/home/user/project".into()), + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + }; + + reg.register("abc-123", 42, ®istration); + assert_eq!(reg.sessions.len(), 1); + assert_eq!(reg.topic_by_session("abc-123"), Some(42)); + assert_eq!(reg.session_by_topic(42), Some("abc-123")); + + // Reload from disk. + let reg2 = SessionRegistry::load(dir.path()); + assert_eq!(reg2.sessions.len(), 1); + assert_eq!(reg2.topic_by_session("abc-123"), Some(42)); + } + + #[test] + fn close_session() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + + let registration = Registration { + session_id: "s1".into(), + label: "test".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + }; + reg.register("s1", 10, ®istration); + + reg.close("s1"); + assert_eq!( + reg.sessions["s1"].state, + SessionState::Closed + ); + // Closed sessions not found by topic lookup. + assert_eq!(reg.session_by_topic(10), None); + } + + #[test] + fn active_sessions_filter() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + + let r1 = Registration { + session_id: "a".into(), + label: "a".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + }; + let r2 = Registration { + session_id: "b".into(), + label: "b".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + }; + reg.register("a", 1, &r1); + reg.register("b", 2, &r2); + reg.close("a"); + + let active = reg.active_sessions(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].0, "b"); + } +} diff --git a/src/router/topics.rs b/src/router/topics.rs new file mode 100644 index 0000000..b9401f6 --- /dev/null +++ b/src/router/topics.rs @@ -0,0 +1,192 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Forum topic management — creates, closes, and reopens topics +//! in the configured supergroup, coordinating with the session registry. + +use std::sync::Arc; + +use anyhow::Result; +use tracing::{info, warn}; + +use crate::telegram::api::BotApi; + +use super::config::RouterConfig; +use super::sessions::{Registration, SessionRegistry}; + +/// Default forum-topic icon color (Telegram's orange, 0xFB6F5F). +/// Base color can only be set at creation — editForumTopic accepts name + custom emoji only. +/// See https://core.telegram.org/bots/api#createforumtopic +const TOPIC_ICON_COLOR: i64 = 16478047; + +/// Manages forum topics for the router. +pub struct TopicManager { + api: Arc, + supergroup_id: String, + close_on_disconnect: bool, +} + +impl TopicManager { + pub fn new(api: Arc, config: &RouterConfig) -> Self { + Self { + api, + supergroup_id: config.supergroup_id.clone(), + close_on_disconnect: config.close_topic_on_disconnect, + } + } + + /// Get a reference to the underlying API client. + pub fn api(&self) -> &Arc { + &self.api + } + + /// Create a new forum topic for a session. + /// + /// Returns the `message_thread_id` (topic ID). + pub async fn ensure_topic( + &self, + session_id: &str, + reg: &Registration, + registry: &mut SessionRegistry, + ) -> Result { + // Check if session already has a topic. + if let Some(topic_id) = registry.topic_by_session(session_id) { + return Ok(topic_id); + } + + // Always create a new topic — each session is independent. + let topic = self + .api + .create_forum_topic(&self.supergroup_id, ®.label, Some(TOPIC_ICON_COLOR)) + .await?; + + let topic_id = topic.message_thread_id; + info!( + session_id, + topic_id, + label = %reg.label, + "created forum topic" + ); + + registry.register(session_id, topic_id, reg); + self.send_welcome(topic_id, reg).await; + + Ok(topic_id) + } + + /// Close the forum topic for a session. + pub async fn close_topic( + &self, + session_id: &str, + registry: &mut SessionRegistry, + ) { + if !self.close_on_disconnect { + return; + } + + let topic_id = match registry.topic_by_session(session_id) { + Some(id) => id, + None => return, + }; + + // Send disconnect message before closing. + let entry = registry.sessions.get(session_id); + let label = entry.map(|e| e.label.as_str()).unwrap_or("?"); + let disconnect_text = format!("Session disconnected: {label}"); + let _ = self + .api + .send_message( + &self.supergroup_id, + &disconnect_text, + None, + None, + None, + Some(topic_id), + ) + .await; + + if let Err(e) = self + .api + .close_forum_topic(&self.supergroup_id, topic_id) + .await + { + warn!(error = %e, topic_id, "failed to close topic"); + } else { + info!(session_id, topic_id, "closed forum topic"); + } + + registry.close(session_id); + } + + /// Send a welcome message to a newly opened/reopened topic. + async fn send_welcome(&self, topic_id: i64, reg: &Registration) { + let pid_info = reg + .pid + .map(|p| format!(" (PID {p})")) + .unwrap_or_default(); + let text = format!("Session connected: {}{pid_info}", reg.label); + let _ = self + .api + .send_message( + &self.supergroup_id, + &text, + None, + None, + None, + Some(topic_id), + ) + .await; + } + + /// Rename a session's forum topic. Returns Ok even if the title didn't + /// change — only logs a warning on API errors to avoid blocking the + /// outbox pipeline on a transient Telegram failure. + pub async fn rename_topic( + &self, + session_id: &str, + new_title: &str, + registry: &mut SessionRegistry, + ) { + let topic_id = match registry.topic_by_session(session_id) { + Some(id) => id, + None => { + warn!(session_id, "rename_topic: no topic for session"); + return; + } + }; + + if !registry.set_title(session_id, new_title) { + return; // unchanged — skip API call + } + + match self + .api + .edit_forum_topic(&self.supergroup_id, topic_id, new_title) + .await + { + Ok(()) => info!(session_id, topic_id, title = %new_title, "renamed topic"), + Err(e) => warn!(session_id, topic_id, error = %e, "editForumTopic failed"), + } + } + + /// Send a message to a session's topic. + pub async fn send_to_topic( + &self, + topic_id: i64, + text: &str, + reply_to: Option, + ) -> Result { + let msg = self + .api + .send_message( + &self.supergroup_id, + text, + reply_to, + None, + None, + Some(topic_id), + ) + .await?; + Ok(msg.message_id) + } +} + diff --git a/src/telegram/api.rs b/src/telegram/api.rs index 1b37ccc..61f12ae 100644 --- a/src/telegram/api.rs +++ b/src/telegram/api.rs @@ -134,11 +134,15 @@ impl BotApi { reply_to: Option, parse_mode: Option<&str>, reply_markup: Option<&InlineKeyboardMarkup>, + message_thread_id: Option, ) -> Result { let mut body = json!({ "chat_id": chat_id, "text": text, }); + if let Some(thread_id) = message_thread_id { + body["message_thread_id"] = json!(thread_id); + } if let Some(rt) = reply_to { body["reply_parameters"] = json!({ "message_id": rt }); } @@ -433,6 +437,120 @@ impl BotApi { Ok(()) } + // ------------------------------------------------------------------ + // Forum topics + // ------------------------------------------------------------------ + + pub async fn create_forum_topic( + &self, + chat_id: &str, + name: &str, + icon_color: Option, + ) -> Result { + let mut body = json!({ + "chat_id": chat_id, + "name": name, + }); + if let Some(color) = icon_color { + body["icon_color"] = json!(color); + } + let resp: CreateForumTopicResponse = self + .client + .post(self.url("createForumTopic")) + .json(&body) + .send() + .await + .context("createForumTopic request")? + .json() + .await + .context("createForumTopic parse")?; + if !resp.ok { + bail!( + "createForumTopic failed: {}", + resp.description.unwrap_or_default() + ); + } + resp.result.context("createForumTopic: missing result") + } + + pub async fn close_forum_topic(&self, chat_id: &str, message_thread_id: i64) -> Result<()> { + let body = json!({ + "chat_id": chat_id, + "message_thread_id": message_thread_id, + }); + let resp: GenericResponse = self + .client + .post(self.url("closeForumTopic")) + .json(&body) + .send() + .await + .context("closeForumTopic request")? + .json() + .await + .context("closeForumTopic parse")?; + if !resp.ok { + bail!( + "closeForumTopic failed: {}", + resp.description.unwrap_or_default() + ); + } + Ok(()) + } + + pub async fn reopen_forum_topic(&self, chat_id: &str, message_thread_id: i64) -> Result<()> { + let body = json!({ + "chat_id": chat_id, + "message_thread_id": message_thread_id, + }); + let resp: GenericResponse = self + .client + .post(self.url("reopenForumTopic")) + .json(&body) + .send() + .await + .context("reopenForumTopic request")? + .json() + .await + .context("reopenForumTopic parse")?; + if !resp.ok { + bail!( + "reopenForumTopic failed: {}", + resp.description.unwrap_or_default() + ); + } + Ok(()) + } + + pub async fn edit_forum_topic( + &self, + chat_id: &str, + message_thread_id: i64, + name: &str, + ) -> Result<()> { + let body = json!({ + "chat_id": chat_id, + "message_thread_id": message_thread_id, + "name": name, + }); + let resp: GenericResponse = self + .client + .post(self.url("editForumTopic")) + .json(&body) + .send() + .await + .context("editForumTopic request")? + .json() + .await + .context("editForumTopic parse")?; + if !resp.ok { + bail!( + "editForumTopic failed: {}", + resp.description.unwrap_or_default() + ); + } + Ok(()) + } + pub async fn edit_message_text_with_markup( &self, chat_id: &str, diff --git a/src/telegram/handlers.rs b/src/telegram/handlers.rs index 74c575f..742c95d 100644 --- a/src/telegram/handlers.rs +++ b/src/telegram/handlers.rs @@ -125,7 +125,7 @@ pub async fn process_update(update: &Update, ctx: &HandlerContext) -> Option { @@ -515,7 +515,7 @@ async fn handle_command(msg: &Message, text: &str, ctx: &HandlerContext) { /status \u{2014} check your pairing state"; let _ = ctx .api - .send_message(&chat_id, reply, None, None, None) + .send_message(&chat_id, reply, None, None, None, None) .await; } "/status" => { @@ -552,7 +552,7 @@ async fn handle_command(msg: &Message, text: &str, ctx: &HandlerContext) { }; let _ = ctx .api - .send_message(&chat_id, &reply, None, None, None) + .send_message(&chat_id, &reply, None, None, None, None) .await; } _ => { @@ -707,7 +707,7 @@ async fn handle_voice_transcription( ); let _ = ctx .api - .send_message(chat_id, &echo_text, Some(msg.message_id), None, None) + .send_message(chat_id, &echo_text, Some(msg.message_id), None, None, None) .await; // Store pending transcription. @@ -917,7 +917,7 @@ pub async fn send_reply( && (reply_mode == ReplyToMode::All || i == 0); let rt = if should_reply_to { reply_to } else { None }; let msg = api - .send_message(chat_id, chunk, rt, parse_mode, None) + .send_message(chat_id, chunk, rt, parse_mode, None, None) .await .map_err(|e| { anyhow::anyhow!( diff --git a/src/telegram/permission.rs b/src/telegram/permission.rs index 5bf63b6..0705275 100644 --- a/src/telegram/permission.rs +++ b/src/telegram/permission.rs @@ -48,7 +48,7 @@ pub async fn handle_permission_request(params: &Value, api: &Arc, state_ for chat_id in &access_data.allow_from { if let Err(e) = api - .send_message(chat_id, &text, None, None, Some(&keyboard)) + .send_message(chat_id, &text, None, None, Some(&keyboard), None) .await { warn!(chat_id, error = %e, "failed to send permission request"); diff --git a/src/telegram/tools.rs b/src/telegram/tools.rs index db8319e..2c3848d 100644 --- a/src/telegram/tools.rs +++ b/src/telegram/tools.rs @@ -3,7 +3,8 @@ //! MCP tool schemas and call handlers for the Telegram channel. //! -//! Four tools: `reply`, `react`, `edit_message`, `download_attachment`. +//! Tools: `reply`, `react`, `edit_message`, `download_attachment`, +//! `set_topic_title` (router mode only). use std::path::Path; @@ -84,6 +85,20 @@ pub fn tool_schemas() -> Value { }, "required": ["chat_id", "message_id", "text"] } + }, + { + "name": "set_topic_title", + "description": "Rename this session's Telegram forum topic. Call when the conversation subject becomes clear or shifts significantly \u{2014} e.g. after the first real task is understood, or when switching to a new subject. Keep titles short (2\u{2013}5 words) and descriptive so the user can find this session in their topic sidebar. Don't rename for minor follow-ups. Router mode only.", + "inputSchema": { + "type": "object", + "properties": { + "title": { + "type": "string", + "description": "New topic name, 1\u{2013}128 characters." + } + }, + "required": ["title"] + } } ]) } @@ -101,6 +116,12 @@ pub async fn handle_tool_call( "react" => handle_react(args, api, state_dir).await, "download_attachment" => handle_download(args, api, inbox_dir).await, "edit_message" => handle_edit(args, api, state_dir).await, + "set_topic_title" => { + anyhow::bail!( + "set_topic_title is only available in router mode — \ + direct mode has no forum-topic concept" + ) + } _ => anyhow::bail!("unknown tool: {name}"), } } diff --git a/src/telegram/types.rs b/src/telegram/types.rs index eaf122f..b7d321c 100644 --- a/src/telegram/types.rs +++ b/src/telegram/types.rs @@ -48,6 +48,7 @@ pub struct Message { pub video: Option