diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4d7828d..f35492c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,6 +38,28 @@ jobs: - name: cargo test run: cargo test --all-features + test: + name: Test (${{ matrix.os }}) + needs: check + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Cache cargo registry + build + uses: Swatinem/rust-cache@v2 + with: + key: test-${{ matrix.os }} + + - name: cargo test + run: cargo test --all-targets --all-features + build: name: Build (${{ matrix.target }}) needs: check @@ -87,19 +109,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..3d7414c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -122,7 +122,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -149,7 +149,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -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,8 +311,10 @@ dependencies = [ "anyhow", "chrono", "dirs", + "fs4", "futures-util", "hex", + "libc", "rand 0.8.5", "regex-lite", "reqwest", @@ -314,6 +326,7 @@ dependencies = [ "tracing", "tracing-subscriber", "uuid", + "windows-sys 0.61.2", ] [[package]] @@ -705,7 +718,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -830,7 +843,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -1019,7 +1032,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1230,7 +1243,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1741,6 +1754,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..2a23d3f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,21 @@ 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", "Win32_System_Diagnostics_ToolHelp"] } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[[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/claude_session.rs b/src/claude_session.rs new file mode 100644 index 0000000..1988c32 --- /dev/null +++ b/src/claude_session.rs @@ -0,0 +1,193 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Discovery of Claude Code's `sessionId` from the MCP process. +//! +//! Claude Code writes `~/.claude/sessions/{PID}.json` for every running +//! session (CLI, VS Code, or desktop). The file contains the stable +//! `sessionId` and the session's `cwd`. We walk up from our own PPID to +//! find the parent process that is Claude Code, then read its file. +//! +//! The sessionId can change during an MCP's lifetime — notably: +//! - `claude --resume` without an ID shows an interactive picker; the +//! file is first written with a fresh UUID, then rewritten with the +//! picked sessionId after the user chooses (arbitrary wait — user may +//! take seconds or minutes). +//! - `/resume ` inside a session switches to another transcript. +//! - `/clear` starts a fresh sessionId in the same process. +//! +//! Because the wait is unbounded, we do **not** block at startup. MCP +//! reads the current value once and proceeds; a background watcher +//! observes the file for changes and lets the router rebind topics +//! after the fact. + +use std::path::PathBuf; + +use serde::Deserialize; + +/// Env var that disables all sessionId discovery (read and watch). Set by +/// the integration test which runs outside a real Claude process. +pub const SKIP_ENV: &str = "HDCD_SKIP_SESSION_DISCOVERY"; + +/// Max depth to walk when looking for Claude in the parent chain. +/// Wrappers (cmd.exe, devcontainer launchers, shell aliases) can insert +/// a layer or two. 8 is ample — real chains are 1–3 deep. +const MAX_PARENT_WALK: u32 = 8; + +/// Deserialized `~/.claude/sessions/{PID}.json`. Extra fields tolerated. +#[derive(Debug, Clone, Deserialize)] +struct PidInfo { + #[serde(rename = "sessionId")] + session_id: String, +} + +fn pid_info_path(pid: u32) -> Option { + Some( + dirs::home_dir()? + .join(".claude") + .join("sessions") + .join(format!("{pid}.json")), + ) +} + +fn read_session_id(pid: u32) -> Option { + let path = pid_info_path(pid)?; + let raw = std::fs::read_to_string(path).ok()?; + let info: PidInfo = serde_json::from_str(&raw).ok()?; + Some(info.session_id) +} + +fn parent_pid() -> Option { + #[cfg(unix)] + unsafe { + Some(libc::getppid() as u32) + } + #[cfg(windows)] + { + parent_of(std::process::id()) + } + #[cfg(not(any(unix, windows)))] + { + None + } +} + +#[cfg(target_os = "linux")] +fn parent_of(pid: u32) -> Option { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + let close = stat.rfind(')')?; + let rest = stat.get(close + 2..)?; + let mut it = rest.split_whitespace(); + let _state = it.next()?; + it.next()?.parse().ok() +} + +#[cfg(target_os = "macos")] +fn parent_of(pid: u32) -> Option { + if pid == 0 || pid > i32::MAX as u32 { + return None; + } + + let output = std::process::Command::new("ps") + .args(["-o", "ppid=", "-p", &pid.to_string()]) + .output() + .ok()?; + if !output.status.success() { + return None; + } + + let ppid = std::str::from_utf8(&output.stdout) + .ok()? + .trim() + .parse() + .ok()?; + (ppid > 0).then_some(ppid) +} + +#[cfg(all(unix, not(any(target_os = "linux", target_os = "macos"))))] +fn parent_of(_pid: u32) -> Option { + None +} + +#[cfg(windows)] +fn parent_of(pid: u32) -> Option { + use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W, + TH32CS_SNAPPROCESS, + }; + + unsafe { + let snapshot = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0); + if snapshot == INVALID_HANDLE_VALUE { + return None; + } + let mut entry: PROCESSENTRY32W = std::mem::zeroed(); + entry.dwSize = std::mem::size_of::() as u32; + + let mut found = None; + if Process32FirstW(snapshot, &mut entry) != 0 { + loop { + if entry.th32ProcessID == pid { + found = Some(entry.th32ParentProcessID); + break; + } + if Process32NextW(snapshot, &mut entry) == 0 { + break; + } + } + } + CloseHandle(snapshot); + found + } +} + +#[cfg(not(any(unix, windows)))] +fn parent_of(_pid: u32) -> Option { + None +} + +/// Walk the parent chain starting from our PPID, returning the first PID +/// that has a readable `sessions/{pid}.json` file. `None` if no ancestor +/// within `MAX_PARENT_WALK` is a Claude process. +pub fn find_claude_pid() -> Option { + let mut pid = parent_pid()?; + for _ in 0..MAX_PARENT_WALK { + if read_session_id(pid).is_some() { + return Some(pid); + } + pid = match parent_of(pid) { + Some(p) if p != 0 && p != pid => p, + _ => break, + }; + } + None +} + +/// Read the current sessionId from Claude's PID file, synchronously. No +/// retries, no waiting — designed to be called often from a watcher poll +/// loop. Returns `None` if the file is missing, unreadable, or discovery +/// is disabled via `HDCD_SKIP_SESSION_DISCOVERY`. +pub fn current_session_id(claude_pid: u32) -> Option { + if std::env::var_os(SKIP_ENV).is_some() { + return None; + } + read_session_id(claude_pid) +} + +/// Discover Claude's PID (parent-chain walk) unless discovery is disabled. +pub fn discover_claude_pid() -> Option { + if std::env::var_os(SKIP_ENV).is_some() { + return None; + } + find_claude_pid() +} + +#[cfg(all(test, any(target_os = "linux", target_os = "macos")))] +mod tests { + use super::parent_of; + + #[test] + fn parent_of_self_returns_some() { + assert!(parent_of(std::process::id()).is_some()); + } +} diff --git a/src/fs_perms.rs b/src/fs_perms.rs new file mode 100644 index 0000000..f708fe6 --- /dev/null +++ b/src/fs_perms.rs @@ -0,0 +1,90 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Maciej Ostaszewski / HyperDev P.S.A. + +//! Defense-in-depth file permissions for router state. +//! +//! The router state directory contains the bot token (`config.json`), +//! session registry, and inbox/outbox payloads — all of which should +//! only be readable by the owning user. On Unix we enforce `0700` on +//! directories and `0600` on files; on Windows the helpers are no-ops +//! (NTFS ACLs default to user-private under the profile directory). + +use std::io; +use std::path::Path; + +/// Apply owner-only permissions to a directory (Unix: `0o700`). +#[cfg(unix)] +pub fn secure_dir(path: &Path) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + let perms = std::fs::Permissions::from_mode(0o700); + std::fs::set_permissions(path, perms) +} + +/// No-op on Windows — inherited ACLs from the user profile directory +/// already keep these private. +#[cfg(not(unix))] +pub fn secure_dir(_path: &Path) -> io::Result<()> { + Ok(()) +} + +/// Apply owner-only permissions to a file (Unix: `0o600`). +#[cfg(unix)] +pub fn secure_file(path: &Path) -> io::Result<()> { + use std::os::unix::fs::PermissionsExt; + let perms = std::fs::Permissions::from_mode(0o600); + std::fs::set_permissions(path, perms) +} + +/// No-op on Windows. +#[cfg(not(unix))] +pub fn secure_file(_path: &Path) -> io::Result<()> { + Ok(()) +} + +/// Emit a `warn!` if `path` is group- or world-readable on Unix. +/// Suggests `chmod 600 ` in the message. No-op on Windows. +pub fn warn_if_world_readable(path: &Path) { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if let Ok(meta) = std::fs::metadata(path) { + let mode = meta.permissions().mode(); + if mode & 0o044 != 0 { + tracing::warn!( + path = %path.display(), + "WARNING: {} is world/group-readable, consider: chmod 600 {}", + path.display(), + path.display() + ); + } + } + } + #[cfg(not(unix))] + { + let _ = path; + } +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + + #[test] + fn secure_dir_sets_0700() { + let tmp = tempfile::tempdir().unwrap(); + secure_dir(tmp.path()).unwrap(); + let mode = std::fs::metadata(tmp.path()).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o700); + } + + #[test] + fn secure_file_sets_0600() { + let tmp = tempfile::tempdir().unwrap(); + let f = tmp.path().join("x"); + std::fs::write(&f, b"x").unwrap(); + secure_file(&f).unwrap(); + let mode = std::fs::metadata(&f).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600); + } +} diff --git a/src/lib.rs b/src/lib.rs index cc878ee..e4469c4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,4 +8,8 @@ //! access control, Bot API client, message handlers, polling, tools, //! voice transcription, and shared types. +pub mod claude_session; +pub mod fs_perms; +pub mod router; pub mod telegram; +pub mod token; diff --git a/src/main.rs b/src/main.rs index c065a5e..e9dbcba 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 @@ -23,14 +30,16 @@ use std::path::PathBuf; use std::sync::Arc; -use anyhow::{bail, Context, Result}; +use anyhow::{Context, Result}; use serde_json::{json, Value}; 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}; +use hdcd_telegram::token; /// MCP protocol version. const MCP_PROTOCOL_VERSION: &str = "2024-11-05"; @@ -83,77 +92,6 @@ fn state_dir() -> Result { Ok(dir) } -/// Load the bot token from env or the `.env` file in the state directory. -/// -/// Checks in priority order: -/// 1. `HDCD_TELEGRAM_BOT_TOKEN` env var (preferred, avoids conflict with official plugin) -/// 2. `TELEGRAM_BOT_TOKEN` env var (backward compatible) -/// 3. `HDCD_TELEGRAM_BOT_TOKEN=` line in `.env` file -/// 4. `TELEGRAM_BOT_TOKEN=` line in `.env` file (backward compatible) -fn load_token(state_dir: &std::path::Path) -> Result { - // Check env vars: HDCD_TELEGRAM_BOT_TOKEN takes priority over TELEGRAM_BOT_TOKEN - for var_name in ["HDCD_TELEGRAM_BOT_TOKEN", "TELEGRAM_BOT_TOKEN"] { - if let Ok(token) = std::env::var(var_name) { - if !token.is_empty() { - return Ok(token); - } - } - } - - // Try loading from ~/.claude/channels/telegram/.env - let env_file = state_dir.join(".env"); - if env_file.exists() { - // Warn if the .env file is world-readable (Unix only). - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - if let Ok(meta) = std::fs::metadata(&env_file) { - let mode = meta.permissions().mode(); - if mode & 0o044 != 0 { - warn!( - path = %env_file.display(), - "WARNING: {} is world-readable, consider: chmod 600 {}", - env_file.display(), - env_file.display() - ); - } - } - } - - let content = std::fs::read_to_string(&env_file) - .with_context(|| format!("read {}", env_file.display()))?; - // HDCD_TELEGRAM_BOT_TOKEN takes priority over TELEGRAM_BOT_TOKEN in .env too - for prefix in ["HDCD_TELEGRAM_BOT_TOKEN=", "TELEGRAM_BOT_TOKEN="] { - for line in content.lines() { - if let Some(rest) = line.strip_prefix(prefix) { - // Strip surrounding quotes (common footgun when copying from other tools). - let trimmed = rest.trim(); - let token = trimmed - .strip_prefix('"') - .and_then(|s| s.strip_suffix('"')) - .or_else(|| { - trimmed - .strip_prefix('\'') - .and_then(|s| s.strip_suffix('\'')) - }) - .unwrap_or(trimmed) - .to_string(); - if !token.is_empty() { - return Ok(token); - } - } - } - } - } - - bail!( - "Bot token required — set HDCD_TELEGRAM_BOT_TOKEN (or TELEGRAM_BOT_TOKEN for backward compatibility)\n \ - in env or {}\n \ - format: HDCD_TELEGRAM_BOT_TOKEN=123456789:AAH...", - state_dir.join(".env").display() - ) -} - /// Write one JSON-RPC frame to stdout. async fn write_frame(stdout: &Arc>, frame: &Value) -> Result<()> { let mut buf = serde_json::to_vec(frame).context("serialize frame")?; @@ -164,6 +102,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,8 +171,22 @@ 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 token = token::load_token(&sd)?; let inbox_dir = sd.join("inbox"); std::fs::create_dir_all(&inbox_dir)?; @@ -231,7 +242,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 +268,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 +316,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 +372,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 +389,604 @@ 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"); + + // Read Claude's sessionId synchronously from `~/.claude/sessions/ + // {claude_pid}.json`. Do NOT block waiting for it to "settle" — on + // `claude --resume` without an ID, the file is rewritten only after + // the user picks an entry from the interactive list, which can take + // arbitrarily long. A background watcher below observes changes and + // rewrites the registration so the router can rebind topics. + let claude_pid = hdcd_telegram::claude_session::discover_claude_pid(); + let initial_claude_session_id = + claude_pid.and_then(hdcd_telegram::claude_session::current_session_id); + if let Some(ref id) = initial_claude_session_id { + info!(claude_session_id = %id, ?claude_pid, "discovered initial claude sessionId"); + } else { + warn!("claude sessionId not discoverable — resume will not match prior topic"); + } + + // 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, + claude_session_id: initial_claude_session_id.clone(), + }; + 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()))?; + let _ = hdcd_telegram::fs_perms::secure_file(®_path); + 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 claude-sessionId watcher: runs unconditionally for the life + // of the MCP. Rediscovers Claude's PID on every tick until one is + // found (MCP can start before Claude has written its PID file), then + // observes value changes and rewrites the registration on each one + // (interactive --resume pick, /resume, /clear). Router picks up the + // rewrite on its next registration poll and rebinds topics. + // + // Uses a dedicated cancel token so we can stop the watcher before + // writing the disconnect marker on shutdown — otherwise a tick that + // observes Claude's PID file vanishing would race-overwrite the + // marker with `disconnected: false` and the router would never see + // the disconnect. + let watcher_cancel = tokio_util::sync::CancellationToken::new(); + let watcher_handle = { + let watch_reg = reg.clone(); + let watch_reg_path = reg_path.clone(); + let watch_cancel = watcher_cancel.clone(); + let mut known_pid = claude_pid; + let mut last_seen = initial_claude_session_id; + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); + interval.tick().await; // consume the immediate first tick + loop { + tokio::select! { + _ = interval.tick() => {} + _ = watch_cancel.cancelled() => return, + } + if known_pid.is_none() { + known_pid = hdcd_telegram::claude_session::discover_claude_pid(); + if let Some(p) = known_pid { + info!(claude_pid = p, "discovered claude PID (post-start)"); + } else { + continue; + } + } + let pid = known_pid.unwrap(); + let current = hdcd_telegram::claude_session::current_session_id(pid); + if current == last_seen { + continue; + } + info!( + prev = ?last_seen, + next = ?current, + "claude sessionId changed — rewriting registration" + ); + last_seen = current.clone(); + let mut updated = watch_reg.clone(); + updated.claude_session_id = current; + match serde_json::to_string_pretty(&updated) { + Ok(json) => { + if let Err(e) = std::fs::write(&watch_reg_path, format!("{json}\n")) { + warn!( + error = %e, + path = %watch_reg_path.display(), + "failed to rewrite registration on rebind" + ); + } else { + let _ = hdcd_telegram::fs_perms::secure_file(&watch_reg_path); + } + } + Err(e) => warn!(error = %e, "failed to serialize updated registration"), + } + } + }) + }; + + // 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 — stop the watcher BEFORE writing the disconnect + // marker. Otherwise, if Claude removed its PID file during shutdown, + // the watcher's next tick would see the sessionId vanish and rewrite + // the registration with `disconnected: false`, stomping the marker + // and leaving the router to think the session is still alive. + watcher_cancel.cancel(); + let _ = watcher_handle.await; + + 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, + claude_session_id: reg.claude_session_id, + }; + 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. Fails loudly +/// when prerequisites are missing so the MCP server never silently +/// sits in router mode with no peer to talk to. +fn ensure_router_running(state_dir: &std::path::Path) -> Result<()> { + let config_path = state_dir.join("config.json"); + if !config_path.exists() { + anyhow::bail!( + "router config.json not found at {}\n \ + Create it with at minimum `supergroup_id` (and optionally `bot_token`, \ + `allowed_users`), or drop the `--router` flag to run in standalone mode.\n \ + See README.md for a full config.json example.", + config_path.display() + ); + } + + // Escape hatch for integration tests that exercise the IPC plumbing + // without a real router peer. Never set this in production. + if std::env::var("HDCD_SKIP_ROUTER_LAUNCH").is_ok() { + debug!("HDCD_SKIP_ROUTER_LAUNCH set, skipping hdcd-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() { + anyhow::bail!( + "hdcd-router binary not found at {}\n \ + Install it alongside hdcd-telegram (same directory), or drop the \ + `--router` flag to run in standalone mode.", + router_exe.display() + ); + } + + info!(path = %router_exe.display(), "launching hdcd-router"); + + // Route stderr to a log file in state_dir instead of inheriting from + // the spawning MCP. When that MCP exits (e.g. the user closes Claude + // before `--resume`), an inherited stderr pipe breaks and subsequent + // writes silently kill tracing-backed tasks inside the router. + let log_path = state_dir.join("router.log"); + let log_file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&log_path) + .with_context(|| format!("open router log {}", log_path.display()))?; + + let mut cmd = std::process::Command::new(&router_exe); + cmd.stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::from(log_file)); + + // 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 +998,14 @@ 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..ec9ebf0 --- /dev/null +++ b/src/router/config.rs @@ -0,0 +1,280 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Router configuration — reads `config.json` from the router state directory. + +use std::fmt; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::de::{self, Visitor}; +use serde::{Deserialize, Deserializer}; + +use crate::token; + +/// Router configuration loaded from +/// `~/.claude/channels/telegram-router/config.json`. +#[derive(Debug, Clone, Deserialize)] +pub struct RouterConfig { + /// Bot token. Optional in `config.json` — if missing or empty the + /// loader falls back to `HDCD_TELEGRAM_BOT_TOKEN` / + /// `TELEGRAM_BOT_TOKEN` env vars, then to the standalone channel's + /// `~/.claude/channels/telegram/.env` file. Always populated after + /// `load()` returns. + #[serde(default)] + pub bot_token: String, + /// Supergroup chat ID. Accepts both JSON integer (`-1001234567890`) + /// and JSON string (`"-1001234567890"`) on deserialization so users + /// can copy the id however their tools emit it. + #[serde(deserialize_with = "de_chat_id")] + pub supergroup_id: i64, + #[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, +} + +impl RouterConfig { + /// String form of `supergroup_id` for Telegram API calls that take + /// `chat_id: &str`. + pub fn chat_id_str(&self) -> String { + self.supergroup_id.to_string() + } +} + +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 +} + +/// Accept either a JSON integer or a JSON string for the chat id. +/// Strings are trimmed before parsing to tolerate leading/trailing +/// whitespace from copy-paste. +fn de_chat_id<'de, D: Deserializer<'de>>(d: D) -> std::result::Result { + struct V; + impl<'de> Visitor<'de> for V { + type Value = i64; + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.write_str("a Telegram chat id (integer or string, e.g. -1001234567890)") + } + fn visit_i64(self, v: i64) -> std::result::Result { + Ok(v) + } + fn visit_u64(self, v: u64) -> std::result::Result { + i64::try_from(v).map_err(|_| E::custom("chat id overflows i64")) + } + fn visit_str(self, v: &str) -> std::result::Result { + v.trim().parse::().map_err(|e| { + E::custom(format!( + "invalid chat id {:?}: {} (expected integer like -1001234567890)", + v, e + )) + }) + } + } + d.deserialize_any(V) +} + +/// 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()))?; + let _ = crate::fs_perms::secure_dir(&dir); + 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"); + crate::fs_perms::warn_if_world_readable(&path); + let raw = std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?; + let mut config: RouterConfig = + serde_json::from_str(&raw).with_context(|| format!("parse {}", path.display()))?; + + if config.bot_token.is_empty() { + let fallback_dir = standalone_state_dir()?; + match token::try_load_token(&fallback_dir)? { + Some(t) => config.bot_token = t, + None => anyhow::bail!( + "bot_token required — set one of:\n \ + - `bot_token` field in {}\n \ + - `HDCD_TELEGRAM_BOT_TOKEN` env var (preferred)\n \ + - `TELEGRAM_BOT_TOKEN` env var (legacy)\n \ + - `HDCD_TELEGRAM_BOT_TOKEN=...` line in {}", + path.display(), + fallback_dir.join(".env").display() + ), + } + } + if config.supergroup_id == 0 { + anyhow::bail!( + "supergroup_id is 0 in {} (expected a Telegram chat id like -1001234567890)", + path.display() + ); + } + if config.supergroup_id > 0 { + tracing::warn!( + supergroup_id = config.supergroup_id, + path = %path.display(), + "supergroup_id is positive; Telegram supergroup IDs usually look like -1001234567890" + ); + } + + Ok(config) +} + +/// Standalone channel's state dir (`~/.claude/channels/telegram/`), used +/// for the `.env` fallback when `config.json` omits `bot_token`. +/// Honors `TELEGRAM_STATE_DIR` the same way the standalone binary does, +/// so a user who overrides one also overrides the other. +fn standalone_state_dir() -> Result { + if let Ok(d) = std::env::var("TELEGRAM_STATE_DIR") { + return Ok(PathBuf::from(d)); + } + Ok(dirs::home_dir() + .context("home dir unavailable")? + .join(".claude") + .join("channels") + .join("telegram")) +} + +#[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_eq!(config.chat_id_str(), "-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 supergroup_id_accepts_integer() { + let json = r#"{ + "bot_token": "123:AAHtest", + "supergroup_id": -1001234567890 + }"#; + let config: RouterConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.supergroup_id, -1001234567890); + } + + #[test] + fn supergroup_id_trims_whitespace() { + let json = r#"{ + "bot_token": "123:AAHtest", + "supergroup_id": " -1001234567890 " + }"#; + let config: RouterConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.supergroup_id, -1001234567890); + } + + #[test] + fn supergroup_id_invalid_string_rejected() { + let json = r#"{ + "bot_token": "123:AAHtest", + "supergroup_id": "not-a-number" + }"#; + let err = serde_json::from_str::(json) + .unwrap_err() + .to_string(); + assert!(err.contains("invalid chat id"), "got: {err}"); + } + + #[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); + } + + use crate::token::with_isolated_token_env; + + #[test] + fn empty_token_with_no_fallback_rejected() { + with_isolated_token_env(|| { + let dir = tempfile::tempdir().unwrap(); + let json = r#"{"bot_token": "", "supergroup_id": "-1001234567890"}"#; + std::fs::write(dir.path().join("config.json"), json).unwrap(); + let err = load(dir.path()).unwrap_err().to_string(); + assert!(err.contains("bot_token required"), "got: {err}"); + }); + } + + #[test] + fn missing_bot_token_field_uses_env_fallback() { + with_isolated_token_env(|| { + std::env::set_var("HDCD_TELEGRAM_BOT_TOKEN", "env-fallback-token"); + let dir = tempfile::tempdir().unwrap(); + let json = r#"{"supergroup_id": "-1001234567890"}"#; + std::fs::write(dir.path().join("config.json"), json).unwrap(); + let config = load(dir.path()).unwrap(); + assert_eq!(config.bot_token, "env-fallback-token"); + }); + } + + #[test] + fn zero_supergroup_rejected_by_load() { + let dir = tempfile::tempdir().unwrap(); + let json = r#"{"bot_token": "x", "supergroup_id": 0}"#; + std::fs::write(dir.path().join("config.json"), json).unwrap(); + let err = load(dir.path()).unwrap_err().to_string(); + assert!(err.contains("supergroup_id is 0"), "got: {err}"); + } +} diff --git a/src/router/mailbox.rs b/src/router/mailbox.rs new file mode 100644 index 0000000..753b915 --- /dev/null +++ b/src/router/mailbox.rs @@ -0,0 +1,507 @@ +// 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")?; + let _ = crate::fs_perms::secure_dir(&inbox); + let _ = crate::fs_perms::secure_dir(&outbox); + let _ = crate::fs_perms::secure_dir(®ister); + 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()))?; + + // Tighten perms on first create. Idempotent — already-0600 is a no-op. + let _ = crate::fs_perms::secure_file(path); + + 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) { + if std::fs::write(pos_path, pos.to_string()).is_ok() { + let _ = crate::fs_perms::secure_file(pos_path); + } +} + +// --------------------------------------------------------------------------- +// 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..1545800 --- /dev/null +++ b/src/router/main.rs @@ -0,0 +1,911 @@ +// 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. + let stale = close_dead_sessions(&state).await; + if stale.is_empty() { + info!("no stale sessions to reconcile"); + } else { + info!( + count = stale.len(), + "reconciled stale sessions from previous run" + ); + } + + // 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, + } + close_dead_sessions(&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.chat_id_str(), &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.chat_id_str(), &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<()> { + // ALLOWLIST INVARIANT: any new update variant handled below MUST call + // is_allowed() before dispatching to session-affecting logic. The + // early-return here only covers variants we ignore entirely. + let msg = match &update.message { + Some(m) => m, + None => { + let variant = if update.callback_query.is_some() { + "callback_query" + } else { + "other" + }; + debug!( + update_id = update.update_id, + variant, "skipping non-message update (allowlist enforced only on messages)" + ); + return Ok(()); + } + }; + + let mut s = state.lock().await; + let chat_id = msg.chat.id; + + // 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.chat_id_str(); + 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.to_string(), + 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)?; + s.registry.increment_activity(&session_id); + + 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.chat_id_str(); + 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); + let mut s = state.lock().await; + s.registry.increment_activity(&session_id); + } + 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(()); + } + + let parse_mode = if msg.format == "markdownv2" { + Some("MarkdownV2") + } else { + None + }; + + // Handle edits. + if let Some(edit_id) = msg.edit_message_id { + api.edit_message_text(supergroup_id, edit_id, &msg.text, parse_mode) + .await?; + return Ok(()); + } + + // Regular message send. + 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 { + 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; + } + + // Skip if already reconciled: session known AND claude_session_id + // unchanged. Registration file stays on disk so subsequent rewrites + // by the MCP-side sessionId watcher trigger rebind on next poll. + { + let s = state.lock().await; + let known = s.registry.topic_by_session(®.session_id).is_some(); + let same_claude_id = s + .registry + .claude_session_id_of(®.session_id) + .map(str::to_string) + == reg.claude_session_id; + if known && same_claude_id { + continue; + } + } + + // Skip stale registrations whose MCP process is already dead. + // Creating a topic just to have the health checker close it again + // seconds later would spam 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; + } + } + + info!( + session_id = %reg.session_id, + label = %reg.label, + pid = ?reg.pid, + claude_session_id = reg.claude_session_id.as_deref().unwrap_or("?"), + "reconciling session registration" + ); + + let mut s = state.lock().await; + let RouterState { + topic_mgr, + registry, + .. + } = &mut *s; + if let Err(e) = topic_mgr.reconcile_session(®, registry).await { + error!( + session_id = %reg.session_id, + error = %e, + "failed to reconcile session" + ); + } + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Health checker +// --------------------------------------------------------------------------- + +/// Collect active sessions whose MCP PID is no longer alive and close +/// their topics. Used both at startup (reconcile leftovers from the +/// previous run) and periodically by the health checker. Returns the +/// list of session_ids that were closed so the caller can log context. +async fn close_dead_sessions(state: &Arc>) -> Vec { + let dead: Vec = { + let s = state.lock().await; + s.registry.dead_session_ids() + }; + + for session_id in &dead { + 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; + } + dead +} + +/// 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"); + if std::fs::write(&lock_path, content).is_ok() { + let _ = hdcd_telegram::fs_perms::secure_file(&lock_path); + } +} + +/// 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 _ = hdcd_telegram::fs_perms::secure_file(&guard_path); + + let locked = file + .try_lock_exclusive() + .with_context(|| format!("lock {}", guard_path.display()))?; + if !locked { + anyhow::bail!("lock held by another process"); + } + + Ok(file) +} + +#[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, + claude_session_id: None, + message_count: 0, + } + } + + #[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()); + } + + #[test] + fn lock_guard_is_exclusive() { + // First acquisition wins; second must fail immediately (not block) + // so a double-start of the router surfaces as a clean error + // instead of hanging on the OS lock. + let dir = tempfile::tempdir().unwrap(); + let first = acquire_lock_guard(dir.path()).expect("first acquire"); + let err = acquire_lock_guard(dir.path()).expect_err("second should fail"); + let msg = format!("{err:#}"); + assert!( + msg.contains("lock") || msg.contains(LOCK_GUARD_FILE), + "unexpected error: {msg}" + ); + drop(first); + + // After the holder drops, a fresh acquire succeeds. + let _third = acquire_lock_guard(dir.path()).expect("re-acquire after drop"); + } +} 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..f13c7d9 --- /dev/null +++ b/src/router/sessions.rs @@ -0,0 +1,455 @@ +// 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, + /// Claude Code's sessionId (from `~/.claude/sessions/{PID}.json`). + /// Stable across MCP restarts — enables `claude --resume ` to + /// reattach to the original forum topic. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub claude_session_id: Option, + /// Count of real Telegram messages routed through this session — + /// inbound (Telegram→MCP notifications) plus outbound (MCP `reply` + /// and `edit_message` delivered). Lets us tell parasitic topics + /// (never saw traffic) from topics worth preserving on rebind. + #[serde(default)] + pub message_count: u32, +} + +/// 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub claude_session_id: Option, +} + +/// 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 _ = crate::fs_perms::secure_file(std::path::Path::new(&tmp)); + let _ = std::fs::rename(&tmp, &self.path); + let _ = crate::fs_perms::secure_file(&self.path); + } + } + Err(e) => warn!(error = %e, "failed to serialize sessions.json"), + } + } + + /// Bind a session to a topic. If another entry already owns the same + /// topic (e.g. a closed prior session this `--resume`/`/resume` is + /// reattaching to) it is evicted — one session_id ↔ one topic_id, no + /// ambiguity. Any existing message_count for this topic carries over, + /// so a reattached topic isn't treated as parasitic on later rebind. + pub fn bind(&mut self, session_id: &str, topic_id: i64, reg: &Registration) { + let mut carried: u32 = 0; + let mut dupes: Vec = Vec::new(); + for (id, e) in &self.sessions { + if e.topic_id != topic_id { + continue; + } + carried = carried.max(e.message_count); + if id.as_str() != session_id { + dupes.push(id.clone()); + } + } + for id in dupes { + self.sessions.remove(&id); + } + + 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, + claude_session_id: reg.claude_session_id.clone(), + message_count: carried, + }, + ); + self.save(); + } + + /// Count of real Telegram messages observed on this session. + pub fn activity_count(&self, session_id: &str) -> u32 { + self.sessions + .get(session_id) + .map(|e| e.message_count) + .unwrap_or(0) + } + + /// Bump the activity counter. Called after a successful inbound + /// notification or outbound delivery. Persists on every change so a + /// router crash never loses activity that's already been sent. + pub fn increment_activity(&mut self, session_id: &str) { + if let Some(e) = self.sessions.get_mut(session_id) { + e.message_count = e.message_count.saturating_add(1); + self.save(); + } + } + + /// Current claude_session_id for a session, if any. + pub fn claude_session_id_of(&self, session_id: &str) -> Option<&str> { + self.sessions + .get(session_id) + .and_then(|e| e.claude_session_id.as_deref()) + } + + /// Find a prior entry matching the given Claude sessionId. Returns + /// `(router_session_id, topic_id)` for any entry (active or closed) — + /// used both to reattach on `claude --resume ` and to forget a + /// stale entry whose Telegram topic has been deleted out from under us. + pub fn find_by_claude_session(&self, claude_session_id: &str) -> Option<(&str, i64)> { + self.sessions + .iter() + .find(|(_, e)| e.claude_session_id.as_deref() == Some(claude_session_id)) + .map(|(id, e)| (id.as_str(), e.topic_id)) + } + + /// 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(); + } + } + + /// Drop a session entry entirely. Used when the bound topic has been + /// deleted (parasitic with zero activity) — keeping the closed + /// entry would leave a dangling topic_id that `find_by_claude_session` + /// could later return and drive a failed reopen on resume. + pub fn forget(&mut self, session_id: &str) { + if self.sessions.remove(session_id).is_some() { + 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() + } + + /// IDs of active sessions whose recorded PID is no longer alive. + /// Used by the startup reconcile + periodic health check to decide + /// which topics to close. Sessions without a PID (pid = None) are + /// treated as alive — we can't contradict what we don't know. + pub fn dead_session_ids(&self) -> Vec { + self.active_sessions() + .iter() + .filter(|(_, e)| e.pid.map(|p| !Self::is_pid_alive(p)).unwrap_or(false)) + .map(|(id, _)| id.to_string()) + .collect() + } + + /// Check if a PID is still alive (platform-specific). + pub fn is_pid_alive(pid: u32) -> bool { + #[cfg(unix)] + { + if pid == 0 || pid > i32::MAX as u32 { + return false; + } + // kill(pid, 0) checks existence without sending a signal. + unsafe { libc::kill(pid as libc::pid_t, 0) == 0 } + } + #[cfg(windows)] + { + use windows_sys::Win32::Foundation::CloseHandle; + use windows_sys::Win32::System::Threading::{ + GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, + }; + 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, + claude_session_id: None, + }; + + reg.bind("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, + claude_session_id: None, + }; + reg.bind("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, + claude_session_id: None, + }; + let r2 = Registration { + session_id: "b".into(), + label: "b".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + claude_session_id: None, + }; + reg.bind("a", 1, &r1); + reg.bind("b", 2, &r2); + reg.close("a"); + + let active = reg.active_sessions(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].0, "b"); + } + + #[test] + fn bind_evicts_prior_owner_and_carries_activity() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + + let old = Registration { + session_id: "old".into(), + label: "old".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + claude_session_id: Some("claude-X".into()), + }; + reg.bind("old", 100, &old); + reg.increment_activity("old"); + reg.increment_activity("old"); + reg.close("old"); + + let new = Registration { + session_id: "new".into(), + label: "new".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:05:00Z".into(), + disconnected: false, + claude_session_id: Some("claude-X".into()), + }; + reg.bind("new", 100, &new); + + assert!(!reg.sessions.contains_key("old")); + assert_eq!(reg.activity_count("new"), 2); + assert_eq!(reg.session_by_topic(100), Some("new")); + } + + #[test] + fn dead_session_ids_finds_sessions_with_nonexistent_pid() { + // u32::MAX is reliably not a live PID on either OS. A session + // whose bound PID is dead must show up in `dead_session_ids`. + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + + let dead_reg = Registration { + session_id: "dead".into(), + label: "dead".into(), + pid: Some(u32::MAX), + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + claude_session_id: None, + }; + reg.bind("dead", 10, &dead_reg); + + let no_pid = Registration { + session_id: "nopid".into(), + label: "nopid".into(), + pid: None, + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + claude_session_id: None, + }; + reg.bind("nopid", 11, &no_pid); + + let dead = reg.dead_session_ids(); + assert_eq!(dead, vec!["dead".to_string()]); + } + + #[test] + fn dead_session_ids_ignores_closed_sessions() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + let r = Registration { + session_id: "s".into(), + label: "s".into(), + pid: Some(u32::MAX), + cwd: None, + registered_at: "2026-04-11T15:00:00Z".into(), + disconnected: false, + claude_session_id: None, + }; + reg.bind("s", 1, &r); + reg.close("s"); + assert!(reg.dead_session_ids().is_empty()); + } + + #[test] + fn increment_activity_saturates_on_unknown_session() { + let dir = tempfile::tempdir().unwrap(); + let mut reg = SessionRegistry::load(dir.path()); + reg.increment_activity("ghost"); // must not panic + assert_eq!(reg.activity_count("ghost"), 0); + } +} diff --git a/src/router/topics.rs b/src/router/topics.rs new file mode 100644 index 0000000..b6f93ba --- /dev/null +++ b/src/router/topics.rs @@ -0,0 +1,278 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Forum topic management — creates, closes, reopens, and rebinds 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 yellow, 0xFFD67E — renders +/// as orange in the client). 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 = 16766590; + +/// Manages forum topics for the router. +pub struct TopicManager { + api: Arc, + /// String form of the supergroup chat id, cached at construction + /// because every Bot API call here wants `chat_id: &str`. + supergroup_id: String, + close_on_disconnect: bool, +} + +impl TopicManager { + pub fn new(api: Arc, config: &RouterConfig) -> Self { + Self { + api, + supergroup_id: config.chat_id_str(), + close_on_disconnect: config.close_topic_on_disconnect, + } + } + + /// Get a reference to the underlying API client. + pub fn api(&self) -> &Arc { + &self.api + } + + /// Reconcile a session's topic with its current registration. Handles + /// three cases in one flow: + /// + /// - **First registration:** no entry yet → create or reopen an + /// existing topic for this `claude_session_id`, bind to it. + /// - **Rebind (sessionId changed):** entry exists but `claude_session_id` + /// has shifted (e.g. user picked a `--resume` target, `/resume`, or + /// `/clear`) → dispose of the old topic (delete if it never saw + /// traffic, close otherwise), attach to the new one. + /// - **No-op:** entry already matches registration → return current topic. + /// + /// Returns the bound `message_thread_id`. + pub async fn reconcile_session( + &self, + reg: &Registration, + registry: &mut SessionRegistry, + ) -> Result { + let session_id = reg.session_id.as_str(); + let current_topic = registry.topic_by_session(session_id); + let current_claude_id = registry + .claude_session_id_of(session_id) + .map(str::to_string); + + if let Some(topic) = current_topic { + if current_claude_id == reg.claude_session_id { + return Ok(topic); + } + } + + let existing_for_claude = reg + .claude_session_id + .as_deref() + .and_then(|id| registry.find_by_claude_session(id)) + .map(|(sid, tid)| (sid.to_string(), tid)); + + // Probe the candidate for reattachment. reopen returns + // TOPIC_NOT_MODIFIED if it's already open (fine), TOPIC_ID_INVALID + // if Telegram has lost or deleted the topic (stale registry entry + // — forget it and fall through to create a fresh one). Any other + // failure is transient; keep the candidate and hope for the best. + let mut reattached_topic = None; + let mut stale_entry_for_claude: Option = None; + if let Some((stale_sid, t)) = existing_for_claude { + match self.api.reopen_forum_topic(&self.supergroup_id, t).await { + Ok(()) => reattached_topic = Some(t), + Err(e) => { + let msg = e.to_string(); + if msg.contains("TOPIC_ID_INVALID") { + warn!( + topic_id = t, + "candidate topic no longer exists on Telegram — discarding stale registry entry and creating new topic" + ); + stale_entry_for_claude = Some(stale_sid); + } else { + warn!( + error = %msg, + topic_id = t, + "reopen_forum_topic failed — topic may already be open" + ); + reattached_topic = Some(t); + } + } + } + } + + if let Some(stale) = stale_entry_for_claude { + registry.forget(&stale); + } + + // No prior topic for this claude_session_id. Two sub-cases fall + // here, distinguished by whether the current topic has seen any + // traffic: + // + // - **Picker → real transition (activity == 0):** MCP just + // started, bound to a fresh topic under the picker-phase + // sessionId; watcher has now rewritten the registration with + // the real id. Reuse the current topic — creating a second one + // would flash/delete the first one the user already saw + // appear in the chat. + // - **Mid-session switch (activity > 0):** user did `/resume` or + // `/clear` and the new sessionId is one the router has never + // seen. The current topic already holds history for a + // *different* logical session — don't hijack it. Retire it + // below (closes on non-zero activity, preserving history) and + // create a fresh topic for the new claude_session_id. + let target_topic = match reattached_topic { + Some(t) => t, + None => match current_topic { + Some(current) if registry.activity_count(session_id) == 0 => current, + _ => { + let topic = self + .api + .create_forum_topic(&self.supergroup_id, ®.label, Some(TOPIC_ICON_COLOR)) + .await?; + topic.message_thread_id + } + }, + }; + + if let Some(old) = current_topic { + if old != target_topic { + let activity = registry.activity_count(session_id); + self.retire_topic(old, activity).await; + } + } + + registry.bind(session_id, target_topic, reg); + + info!( + session_id, + topic_id = target_topic, + claude_session_id = reg.claude_session_id.as_deref().unwrap_or("?"), + label = %reg.label, + reattached = reattached_topic.is_some(), + "session bound to topic" + ); + + Ok(target_topic) + } + + /// Close a session's topic on disconnect. Registry state always + /// updates; the actual API call is gated by config. Paraysitic + /// topics (zero activity — picker stubs, aborted resumes, crashed + /// MCPs that never got a real message) are deleted instead of + /// closed, matching the rebind path's `retire_topic` behavior so the + /// supergroup doesn't accumulate empty closed stubs. + pub async fn close_topic(&self, session_id: &str, registry: &mut SessionRegistry) { + let topic_id = match registry.topic_by_session(session_id) { + Some(id) => id, + None => return, + }; + let activity = registry.activity_count(session_id); + + let deleted = if self.close_on_disconnect { + self.retire_topic(topic_id, activity).await + } else { + false + }; + + if deleted { + registry.forget(session_id); + } else { + registry.close(session_id); + } + } + + /// Dispose of a topic we are no longer bound to. Paraysitic topics + /// (zero activity — never saw any Telegram traffic) get deleted so the + /// supergroup isn't littered with empty stubs from every interactive + /// `claude --resume` session-picker. Topics that carry real history + /// are only closed. Delete failures fall back to close. + /// Returns `true` if the topic was deleted, `false` if closed. + async fn retire_topic(&self, topic_id: i64, activity: u32) -> bool { + if activity == 0 { + match self + .api + .delete_forum_topic(&self.supergroup_id, topic_id) + .await + { + Ok(()) => { + info!(topic_id, "deleted parasitic topic (zero activity)"); + return true; + } + Err(e) => warn!( + error = %e, + topic_id, + "delete_forum_topic failed — falling back to close" + ), + } + } + if let Err(e) = self + .api + .close_forum_topic(&self.supergroup_id, topic_id) + .await + { + warn!(error = %e, topic_id, "close_forum_topic failed"); + } else { + info!(topic_id, activity, "closed retired topic"); + } + false + } + + /// 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; + } + + 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..48d23e2 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,144 @@ 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 delete_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("deleteForumTopic")) + .json(&body) + .send() + .await + .context("deleteForumTopic request")? + .json() + .await + .context("deleteForumTopic parse")?; + if !resp.ok { + bail!( + "deleteForumTopic 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