diff --git a/Cargo.lock b/Cargo.lock index cfde852..7d8cfb6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -88,6 +88,29 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "aquamarine" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f50776554130342de4836ba542aa85a4ddb361690d7e8df13774d7284c3d5c2" +dependencies = [ + "include_dir", + "itertools 0.10.5", + "proc-macro-error2", + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "ar_archive_writer" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4087686b4b0a3427190bae57a1d9a478dbb2d40c5dc1bd6e2b6d797913bdd348" +dependencies = [ + "object", +] + [[package]] name = "argon2" version = "0.5.3" @@ -639,6 +662,15 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "colored" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "combine" version = "4.6.7" @@ -1109,13 +1141,34 @@ dependencies = [ "serde_core", ] +[[package]] +name = "derive_more" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05" +dependencies = [ + "derive_more-impl 1.0.0", +] + [[package]] name = "derive_more" version = "2.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134" dependencies = [ - "derive_more-impl", + "derive_more-impl 2.1.1", +] + +[[package]] +name = "derive_more-impl" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", + "unicode-xid", ] [[package]] @@ -1276,6 +1329,16 @@ dependencies = [ "serde", ] +[[package]] +name = "dptree" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db96968fcf52fe063a98c75df1d1f2b1fba304e7ae29b72fdc81c1165b7e2fd0" +dependencies = [ + "colored", + "futures", +] + [[package]] name = "dtoa" version = "1.0.11" @@ -1406,6 +1469,16 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "erasable" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "437cfb75878119ed8265685c41a115724eae43fb7cc5a0bf0e4ecc3b803af1c4" +dependencies = [ + "autocfg", + "scopeguard", +] + [[package]] name = "erased-serde" version = "0.4.10" @@ -1859,11 +1932,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", "wasip2", "wasip3", + "wasm-bindgen", ] [[package]] @@ -2188,7 +2263,7 @@ checksum = "ec9d92d097f4749b64e8cc33d924d9f40a2d4eb91402b458014b781f5733d60f" name = "honcho-client" version = "0.1.0" dependencies = [ - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "tracing", @@ -2311,6 +2386,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", + "webpki-roots 1.0.7", ] [[package]] @@ -2487,6 +2563,25 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "include_dir" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "923d117408f1e49d914f1a379a309cffe4f18c05cf4e3d12e613a15fc81bd0dd" +dependencies = [ + "include_dir_macros", +] + +[[package]] +name = "include_dir_macros" +version = "0.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cab85a7ed0bd5f0e76d93846e0147172bed2e2d3f859bcc33a8d9699cad1a75" +dependencies = [ + "proc-macro2", + "quote", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -2919,7 +3014,7 @@ dependencies = [ "anyhow", "async-trait", "core-api", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "tracing", @@ -2940,6 +3035,12 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "managed" version = "0.8.0" @@ -2979,7 +3080,7 @@ dependencies = [ "anyhow", "async-trait", "base64 0.22.1", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "tokio", @@ -3478,6 +3579,15 @@ dependencies = [ "objc2-foundation", ] +[[package]] +name = "object" +version = "0.37.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe" +dependencies = [ + "memchr", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -3793,6 +3903,26 @@ dependencies = [ "siphasher", ] +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -3835,7 +3965,7 @@ dependencies = [ "anyhow", "async-trait", "core-api", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "tokio", @@ -3850,7 +3980,7 @@ dependencies = [ "async-trait", "core-api", "parking_lot", - "reqwest", + "reqwest 0.13.4", "serde_json", "tokio", "tracing", @@ -3871,6 +4001,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "plugin-telegram-bot" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "chrono", + "core-api", + "rand 0.10.1", + "regex", + "serde", + "serde_json", + "teloxide", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "plugin-transcribe-whisper-local" version = "0.1.0" @@ -3892,7 +4040,7 @@ dependencies = [ "anyhow", "async-trait", "core-api", - "reqwest", + "reqwest 0.13.4", "serde_json", "tokio", "tracing", @@ -3905,7 +4053,7 @@ dependencies = [ "anyhow", "async-trait", "core-api", - "reqwest", + "reqwest 0.13.4", "serde_json", "tokio", "tracing", @@ -4234,6 +4382,16 @@ version = "3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3" +[[package]] +name = "psm" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "645dbe486e346d9b5de3ef16ede18c26e6c70ad97418f4874b8b1889d6e761ea" +dependencies = [ + "ar_archive_writer", + "cc", +] + [[package]] name = "quick-xml" version = "0.41.0" @@ -4243,6 +4401,62 @@ dependencies = [ "memchr", ] +[[package]] +name = "quinn" +version = "0.11.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.18", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +dependencies = [ + "bytes", + "getrandom 0.4.2", + "lru-slab", + "rand 0.10.1", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.45" @@ -4350,12 +4564,30 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "raw-window-handle" version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539" +[[package]] +name = "rc-box" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "897fecc9fac6febd4408f9e935e86df739b0023b625e610e0357535b9c8adad0" +dependencies = [ + "erasable", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -4442,6 +4674,48 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "mime_guess", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tokio-util", + "tower", + "tower-http 0.6.11", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-streams 0.4.2", + "web-sys", + "webpki-roots 1.0.7", +] + [[package]] name = "reqwest" version = "0.13.4" @@ -4481,10 +4755,19 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.5.0", "web-sys", ] +[[package]] +name = "rgb" +version = "0.8.53" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b34b781b31e5d73e9fbc8689c70551fd1ade9a19e3e28cfec8580a79290cc4" +dependencies = [ + "bytemuck", +] + [[package]] name = "ring" version = "0.17.14" @@ -4601,6 +4884,7 @@ version = "1.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" dependencies = [ + "web-time", "zeroize", ] @@ -4770,7 +5054,7 @@ checksum = "c5d9c0c92a92d33f08817311cf3f2c29a3538a8240e94a6a3c622ce652d7e00c" dependencies = [ "bitflags 2.11.1", "cssparser", - "derive_more", + "derive_more 2.1.1", "log", "new_debug_unreachable", "phf 0.13.1", @@ -5124,10 +5408,11 @@ dependencies = [ "plugin-comfyui", "plugin-elevenlabs", "plugin-tailscale-remote", + "plugin-telegram-bot", "plugin-transcribe-whisper-local", "plugin-tts-kokoro", "plugin-tts-orpheus-3b", - "reqwest", + "reqwest 0.13.4", "rustls", "serde", "serde_json", @@ -5174,7 +5459,7 @@ dependencies = [ "quote", "rand 0.10.1", "regex", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "sha2 0.10.9", @@ -5268,7 +5553,7 @@ dependencies = [ "jsonwebtoken", "prost", "rand 0.8.6", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "sha2 0.10.9", @@ -5576,6 +5861,19 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "stacker" +version = "0.1.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "640c8cdd92b6b12f5bcb1803ca3bbf5ab96e5e6b6b96b9ab77dabe9e880b3190" +dependencies = [ + "cc", + "cfg-if", + "libc", + "psm", + "windows-sys 0.61.2", +] + [[package]] name = "static_assertions" version = "1.1.0" @@ -5771,6 +6069,18 @@ dependencies = [ "url", ] +[[package]] +name = "take_mut" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f764005d11ee5f36500a149ace24e00e3da98b0158b3e2d53a7495660d3f4d60" + +[[package]] +name = "takecell" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20f34339676cdcab560c9a82300c4c2581f68b9369aedf0fae86f2ff9565ff3e" + [[package]] name = "tao" version = "0.35.3" @@ -5858,7 +6168,7 @@ dependencies = [ "percent-encoding", "plist", "raw-window-handle", - "reqwest", + "reqwest 0.13.4", "serde", "serde_json", "serde_repr", @@ -6041,6 +6351,76 @@ dependencies = [ "toml 1.1.2+spec-1.1.0", ] +[[package]] +name = "teloxide" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84992abeed3ae42e8401b25d266d12bcba1def0abe59d22f6b9781167545f71e" +dependencies = [ + "aquamarine", + "bytes", + "derive_more 1.0.0", + "dptree", + "either", + "futures", + "log", + "mime", + "pin-project", + "serde", + "serde_json", + "teloxide-core", + "teloxide-macros", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tokio-util", + "url", +] + +[[package]] +name = "teloxide-core" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f7a34ca8e971fa892e633858c07547fe138ef4a02e4a4eaa1d35e517d6e0bc4" +dependencies = [ + "bitflags 2.11.1", + "bytes", + "chrono", + "derive_more 1.0.0", + "either", + "futures", + "log", + "mime", + "once_cell", + "pin-project", + "rc-box", + "reqwest 0.12.28", + "rgb", + "serde", + "serde_json", + "serde_with", + "stacker", + "take_mut", + "takecell", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "url", + "uuid", +] + +[[package]] +name = "teloxide-macros" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300fadcaf0c182f19b5ca10bf23a45dc9a48925f00c704405fd90ee2c03942f9" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -7554,6 +7934,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasm-streams" version = "0.5.0" @@ -7589,6 +7982,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "web_atoms" version = "0.2.5" diff --git a/Cargo.toml b/Cargo.toml index 638f890..4de31c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "crates/core-api", "crates/mcp-client", "crates/plugin-tailscale-remote", + "crates/plugin-telegram-bot", "crates/plugin-transcribe-whisper-local", "crates/plugin-comfyui", "crates/plugin-tts-orpheus-3b", @@ -81,6 +82,7 @@ llm-client = { path = "crates/llm-client" } core-api = { path = "crates/core-api" } mcp-client = { path = "crates/mcp-client" } plugin-tailscale-remote = { path = "crates/plugin-tailscale-remote" } +plugin-telegram-bot = { path = "crates/plugin-telegram-bot" } plugin-transcribe-whisper-local = { path = "crates/plugin-transcribe-whisper-local", optional = true } plugin-comfyui = { path = "crates/plugin-comfyui" } plugin-tts-orpheus-3b = { path = "crates/plugin-tts-orpheus-3b" } diff --git a/crates/core-api/src/config_api.rs b/crates/core-api/src/config_api.rs new file mode 100644 index 0000000..09adf52 --- /dev/null +++ b/crates/core-api/src/config_api.rs @@ -0,0 +1,14 @@ +use anyhow::Result; +use async_trait::async_trait; + +/// Read/write access to the instance-wide key/value config store +/// (`config` table in `system.db`). +/// +/// [`ConfigApi::set`] emits `ConfigKeyUpdated` on the system bus when the +/// value changes, so subscribers (e.g. the Telegram plugin reloading its +/// bindings) are notified without polling. +#[async_trait] +pub trait ConfigApi: Send + Sync { + async fn get(&self, key: &str) -> Result>; + async fn set(&self, key: &str, value: &str) -> Result<()>; +} diff --git a/crates/core-api/src/lib.rs b/crates/core-api/src/lib.rs index 4d0b2de..95a50d0 100644 --- a/crates/core-api/src/lib.rs +++ b/crates/core-api/src/lib.rs @@ -3,6 +3,7 @@ pub const APP_NAME: &str = "Skald"; pub mod approval; pub mod bus; +pub mod config_api; pub mod system_bus; pub mod chatbot; pub mod chat_hub; @@ -18,6 +19,7 @@ pub mod plugin; pub mod provider; pub mod remote; pub mod tool; +pub mod user_channel; pub mod secrets; pub mod transcribe; pub mod tts; diff --git a/crates/core-api/src/plugin.rs b/crates/core-api/src/plugin.rs index 25a1a92..6026bc0 100644 --- a/crates/core-api/src/plugin.rs +++ b/crates/core-api/src/plugin.rs @@ -5,12 +5,10 @@ use async_trait::async_trait; use serde_json::Value; use tokio::sync::RwLock; -use crate::approval::ApprovalApi; use crate::command::CommandApi; +use crate::config_api::ConfigApi; use crate::system_bus::SystemEventBus; -use crate::chat_hub::ChatHubApi; use crate::image_generate::ImageGenerateRegistry; -use crate::inbox::InboxApi; use crate::location::LocationUpdater; use crate::memory::Memory; use crate::provider::ApiProviderRegistry; @@ -18,6 +16,7 @@ use crate::remote::RemoteAccess; use crate::secrets::SecretsApi; use crate::transcribe::{TranscribeProvider, TranscribeRegistry}; use crate::tts::{TtsProvider, TtsRegistry}; +use crate::user_channel::UserChannelApi; /// Closure that builds a fresh Axum router (e.g. for the mesh-facing server). pub type RouterFactory = Arc axum::Router + Send + Sync>; @@ -33,6 +32,9 @@ pub struct PluginContext { /// Custom file-based slash commands (`commands//`). Read-only from the /// plugin side — lets the Telegram bot resolve `/command` expansions. pub command: Arc, + /// Key/value config store (`config` table in `system.db`). `set` emits + /// `ConfigKeyUpdated` on the system bus. + pub config: Arc, /// Skald's shared SQLite pool — lets plugins create/use their own tables /// (e.g. `relay_*`) in the main DB. See plugin.md §12.1. pub db: Arc, @@ -45,6 +47,10 @@ pub struct PluginContext { pub api_provider_registry: Arc, pub location: Arc, pub system_bus: Arc, + /// Channel-to-session resolver (blueprint §13). Lets channel plugins + /// (Telegram, mobile, …) look up an unlocked user's chat hub, approval + /// manager and event stream by user id. + pub user_channel: Arc, pub web_port: u16, pub remote_slot: Arc>>>, pub router_factory: RouterFactory, diff --git a/crates/core-api/src/user_channel.rs b/crates/core-api/src/user_channel.rs new file mode 100644 index 0000000..5fbce09 --- /dev/null +++ b/crates/core-api/src/user_channel.rs @@ -0,0 +1,57 @@ +//! Channel-to-session contract (blueprint §13). +//! +//! In the multi-user architecture, chat hubs, approval managers and event +//! streams are per-user (inside [`UserContext`]). External channels (Telegram, +//! mobile, …) need a way to resolve a user's owner-bound runtime at runtime, +//! without depending on the concrete `Skald` / `UserContext` types. +//! +//! [`UserChannelApi`] is the lookup seam: given a `user_id`, returns a +//! [`UserChannelHandle`] when the user's database is unlocked (§9), or `None` +//! when it is still locked. The handle exposes the per-user [`ChatHubApi`], +//! [`ApprovalApi`] and event stream — everything a channel adapter needs to +//! route a message and receive the response events. +//! +//! This is the "one contract" of §13: each channel (Telegram, mobile, …) is a +//! thin adapter over it, so N channel rewrites become 1 contract + N adapters. + +use std::sync::Arc; + +use async_trait::async_trait; +use tokio::sync::broadcast; + +use crate::approval::ApprovalApi; +use crate::chat_hub::ChatHubApi; +use crate::events::GlobalEvent; + +/// Resolves an unlocked user's channel handle. +/// +/// Implemented by the application core (`Skald`) and injected into +/// [`crate::plugin::PluginContext`] as `user_channel`. +#[async_trait] +pub trait UserChannelApi: Send + Sync { + /// Returns the user's handle if their database is unlocked in this boot + /// (§9: from first login until restart). `None` = locked — the caller + /// should prompt the user to log in. + async fn resolve_user(&self, user_id: &str) -> Option>; +} + +/// Handle to one unlocked user's owner-bound runtime. +/// +/// Lifetime = the user's pool lifetime (§9). Cloning the returned `Arc`s is +/// cheap (they share the underlying state). The event receiver obtained from +/// [`UserChannelHandle::subscribe`] is independent per call — each subscriber +/// gets every future event. +pub trait UserChannelHandle: Send + Sync { + /// The opaque user id this handle belongs to. + fn user_id(&self) -> &str; + + /// The user's chat hub — send messages, manage sessions, query context. + fn chat_hub(&self) -> Arc; + + /// The user's approval manager — resolve pending tool-call approvals. + fn approval(&self) -> Arc; + + /// Subscribe to the user's server→client event stream. + /// Events are scoped to this user; no cross-user leakage. + fn subscribe(&self) -> broadcast::Receiver; +} diff --git a/crates/plugin-telegram-bot/src/auth.rs b/crates/plugin-telegram-bot/src/auth.rs index 56d4391..be82a31 100644 --- a/crates/plugin-telegram-bot/src/auth.rs +++ b/crates/plugin-telegram-bot/src/auth.rs @@ -1,79 +1,91 @@ -use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::time::SystemTime; -use anyhow::Result; use chrono::{DateTime, Local, Utc}; use rand::RngExt; use serde::{Deserialize, Serialize}; use teloxide::prelude::*; use teloxide::types::ParseMode; -use tokio::time::Duration; +use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; -use tracing::{error, info}; +use tracing::{error, info, warn}; + +use core_api::config_api::ConfigApi; +use core_api::system_bus::SystemEvent; use super::TgShared; -// ── Whitelist file schema ───────────────────────────────────────────────────── -// -// Written to secrets/telegram_whitelist.json. -// The main agent edits this file directly to authorise users. +/// Config-table key under which all Telegram bindings are stored as JSON. +pub(crate) const CONFIG_KEY: &str = "telegram"; -#[derive(Debug, Serialize, Deserialize, Default)] -pub struct WhitelistFile { +// ── Bindings schema (stored as JSON in the `config` table) ──────────────────── + +#[derive(Debug, Serialize, Deserialize, Default, Clone)] +pub struct TelegramConfig { #[serde(default)] - pub whitelist: Vec, + pub bindings: Vec, #[serde(default)] pub pending_pairings: Vec, } -#[derive(Debug, Serialize, Deserialize)] -pub struct PairingEntry { - pub code: String, - pub chat_id: i64, - pub issued_at: String, +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct Binding { + pub chat_id: i64, + pub user_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display: Option, } -pub(crate) async fn load_wl(secrets_dir: &Path) -> WhitelistFile { - let path = secrets_dir.join("telegram_whitelist.json"); - match tokio::fs::read_to_string(&path).await { - Ok(s) => serde_json::from_str(&s).unwrap_or_default(), - Err(_) => WhitelistFile::default(), +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct PairingEntry { + pub code: String, + pub chat_id: i64, + pub issued_at: String, +} + +// ── Config-table read/write ──────────────────────────────────────────────────── + +/// Reads the Telegram config from the `config` table. Returns `Default` when +/// the key is absent or unparseable (never fails the caller). +pub(crate) async fn load_config(config: &dyn ConfigApi) -> anyhow::Result { + match config.get(CONFIG_KEY).await? { + Some(json) => Ok(serde_json::from_str(&json).unwrap_or_default()), + None => Ok(TelegramConfig::default()), } } -pub(crate) async fn save_wl(secrets_dir: &Path, wl: &WhitelistFile) -> Result<()> { - tokio::fs::create_dir_all(secrets_dir).await?; - let path = secrets_dir.join("telegram_whitelist.json"); - tokio::fs::write(&path, serde_json::to_string_pretty(wl)?).await?; - Ok(()) +/// Writes the Telegram config to the `config` table. `ConfigApi::set` emits a +/// `ConfigKeyUpdated` event when the value changes, so the in-memory cache and +/// any forwarders are updated automatically. +pub(crate) async fn save_config( + config: &dyn ConfigApi, + cfg: &TelegramConfig, +) -> anyhow::Result<()> { + config.set(CONFIG_KEY, &serde_json::to_string_pretty(cfg)?).await } // ── Pairing ─────────────────────────────────────────────────────────────────── -/// Pairing codes older than this are considered abandoned and pruned, so the -/// whitelist file does not accumulate stale `pending_pairings` entries. +/// Pairing codes older than this are considered abandoned and pruned. const PAIRING_TTL_HOURS: i64 = 24; +/// Called when an unbound `chat_id` sends a message. Generates (or reuses) a +/// pairing code, persists it to the config table, and replies with instructions. pub(crate) async fn handle_pairing(bot: &Bot, chat_id: ChatId, shared: &Arc) { - let mut wl = load_wl(&shared.secrets_dir).await; + let mut cfg = shared.bindings.read().await.clone(); - // Drop pairing codes past their TTL. Entries with an unparseable timestamp - // are kept (don't silently lose data on a format change). + // Prune expired codes. let cutoff = Utc::now() - chrono::Duration::hours(PAIRING_TTL_HOURS); - let before = wl.pending_pairings.len(); - wl.pending_pairings.retain(|e| match DateTime::parse_from_rfc3339(&e.issued_at) { + cfg.pending_pairings.retain(|e| match DateTime::parse_from_rfc3339(&e.issued_at) { Ok(ts) => ts.with_timezone(&Utc) > cutoff, Err(_) => true, }); - let pruned = wl.pending_pairings.len() != before; - // Re-use an existing (non-expired) code if one is already pending for this chat. - let (code, added) = if let Some(entry) = wl.pending_pairings.iter().find(|e| e.chat_id == chat_id.0) { + // Reuse an existing code for this chat, or generate a new one. + let (code, added) = if let Some(entry) = cfg.pending_pairings.iter().find(|e| e.chat_id == chat_id.0) { (entry.code.clone(), false) } else { let code = generate_code(); - wl.pending_pairings.push(PairingEntry { + cfg.pending_pairings.push(PairingEntry { code: code.clone(), chat_id: chat_id.0, issued_at: Local::now().format("%Y-%m-%dT%H:%M:%S%:z").to_string(), @@ -81,14 +93,16 @@ pub(crate) async fn handle_pairing(bot: &Bot, chat_id: ChatId, shared: &ArcPairing required.\n\n\ Code: {code}\n\n\ - Provide this code to the web agent to authorize access.", + Ask the admin to authorize this chat using the telegram_pairing tool.", ), ) .parse_mode(ParseMode::Html) @@ -110,60 +124,38 @@ pub(crate) fn generate_code() -> String { (0..6).map(|_| CHARS[rng.random_range(0..CHARS.len())] as char).collect() } -// ── Whitelist watchdog ──────────────────────────────────────────────────────── -// -// Polls telegram_whitelist.json every 10 s for mtime changes. -// When a new chat_id appears in `whitelist` (agent moved it from pending), -// sends a welcome message so the user knows they are authorized. - -pub(crate) async fn whitelist_watchdog(bot: Bot, secrets_dir: PathBuf, cancel: CancellationToken) { - let path = secrets_dir.join("telegram_whitelist.json"); - - let mut last_mtime: Option = tokio::fs::metadata(&path).await.ok() - .and_then(|m| m.modified().ok()); - let mut known_wl = load_wl(&secrets_dir).await.whitelist; - - let mut interval = tokio::time::interval(Duration::from_secs(10)); - interval.tick().await; // skip the immediate first tick +// ── Config listener ──────────────────────────────────────────────────────────── +/// Subscribes to the system bus and reloads the in-memory bindings whenever the +/// `"telegram"` config key changes. Replaces the old file-polling watchdog. +pub(crate) async fn config_listener( + shared: Arc, + mut rx: broadcast::Receiver, + cancel: CancellationToken, +) { + info!("telegram: config listener started"); loop { tokio::select! { - _ = cancel.cancelled() => break, - _ = interval.tick() => { - let new_mtime = tokio::fs::metadata(&path).await.ok() - .and_then(|m| m.modified().ok()); - if new_mtime.is_none() || new_mtime == last_mtime { - continue; - } - last_mtime = new_mtime; - - let wl = load_wl(&secrets_dir).await; - let newly_authorized: Vec = wl.whitelist.iter() - .filter(|id| !known_wl.contains(id)) - .cloned() - .collect(); - - if !newly_authorized.is_empty() { - info!(users = ?newly_authorized, "telegram: new users authorized — sending welcome"); - for &chat_id in &newly_authorized { - bot.send_message( - ChatId(chat_id), - "✅ Access granted!\n\ - You can now talk to your agent.\n\n\ - /help for available commands.", - ) - .parse_mode(ParseMode::Html) - .await - .ok(); + _ = cancel.cancelled() => { + info!("telegram: config listener stopped"); + return; + } + result = rx.recv() => match result { + Ok(SystemEvent::ConfigKeyUpdated { key, new_value, .. }) if key == CONFIG_KEY => { + match serde_json::from_str::(&new_value) { + Ok(cfg) => { + let n = cfg.bindings.len(); + *shared.bindings.write().await = cfg; + info!(bindings = n, "telegram: bindings reloaded from config event"); + } + Err(e) => warn!(error = %e, "telegram: failed to parse config from event"), } } - - known_wl = wl.whitelist; - info!( - whitelist = known_wl.len(), - pending = wl.pending_pairings.len(), - "telegram: whitelist file reloaded" - ); + Ok(_) => {} + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!(skipped = n, "telegram: config listener lagged"); + } + Err(broadcast::error::RecvError::Closed) => return, } } } diff --git a/crates/plugin-telegram-bot/src/events.rs b/crates/plugin-telegram-bot/src/events.rs index 97a2ffd..e9248d8 100644 --- a/crates/plugin-telegram-bot/src/events.rs +++ b/crates/plugin-telegram-bot/src/events.rs @@ -1,4 +1,5 @@ use std::sync::Arc; + use teloxide::prelude::*; use teloxide::types::{InlineKeyboardButton, InlineKeyboardMarkup, ParseMode}; use tokio::sync::broadcast; @@ -6,17 +7,17 @@ use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use core_api::events::{GlobalEvent, ServerEvent}; +use core_api::user_channel::UserChannelHandle; use super::TgShared; -use super::auth::load_wl; use super::helpers::{escape_html, label_to_html, send_long}; - -/// Sends an inline keyboard for an approval request and records the request_id. +/// Sends an inline keyboard for an approval request and records the request. async fn send_approval_keyboard( bot: &Bot, chat_id: ChatId, text: String, + user_id: String, request_id: i64, shared: &Arc, ) { @@ -37,101 +38,122 @@ async fn send_approval_keyboard( .reply_markup(keyboard) .await { - Ok(m) => { shared.pending_approvals.lock().await.insert(m.id, request_id); } + Ok(m) => { + shared.pending_approvals.lock().await.insert( + m.id, + super::PendingApproval { user_id, request_id }, + ); + } Err(e) => error!(error = %e, "telegram: failed to send approval message"), } } -// ── Persistent background forwarder ────────────────────────────────────────── +// ── Per-user forwarder ──────────────────────────────────────────────────────── -/// Spawned once when the plugin starts. -/// Stays subscribed to the "telegram" broadcast channel forever, forwarding -/// events to the home chat_id. This is the only subscriber — per-message -/// subscriptions are not used — so it also catches background notifications -/// that arrive without a user message triggering them. -/// -/// Re-subscribes immediately after each `Done`/`Error` so no events from the -/// next turn are missed. Safe because Tokio's cooperative scheduler guarantees -/// no other task runs between the re-subscription point and the next `await`, -/// and the processing mutex in `ChatSessionHandler` serialises turns. -pub(crate) async fn persistent_forwarder( - bot: Bot, - shared: Arc, - cancel: CancellationToken, +/// Spawns forwarders for all bound users whose contexts are already unlocked. +/// Called at plugin start. Users who log in later get their forwarder spawned +/// lazily on first incoming message. +pub(crate) async fn spawn_forwarders_for_bound_users( + bot: &Bot, + shared: &Arc, + cancel: &CancellationToken, ) { - info!("telegram: persistent forwarder started"); + let bindings = shared.bindings.read().await.clone(); + for b in &bindings.bindings { + if let Some(handle) = shared.user_channel.resolve_user(&b.user_id).await { + ensure_forwarder(bot.clone(), Arc::clone(shared), &b.user_id, b.chat_id, handle, cancel.clone()).await; + } + } +} - let mut rx = shared.chat_hub.events("telegram"); +/// Spawns a per-user forwarder if one is not already running for `user_id`. +/// The forwarder subscribes to the user's event stream and routes `ServerEvent`s +/// to the bound Telegram `chat_id`. +pub(crate) async fn ensure_forwarder( + bot: Bot, + shared: Arc, + user_id: &str, + chat_id: i64, + handle: Arc, + cancel: CancellationToken, +) { + let mut forwarders = shared.forwarders.lock().await; + if forwarders.contains(user_id) { + return; + } + forwarders.insert(user_id.to_string()); + + let uid = user_id.to_string(); + info!(user_id = %uid, chat_id, "telegram: spawning per-user forwarder"); + + let shared_c = Arc::clone(&shared); + let cancel_c = cancel.clone(); + tokio::spawn(user_forwarder(bot, shared_c, uid, chat_id, handle, cancel_c)); +} + +/// One forwarder per unlocked user. Subscribes to the user's `global_tx` and +/// routes events to Telegram. Exits when the broadcast channel closes (user +/// context dropped at restart / lock) or the plugin is cancelled. +async fn user_forwarder( + bot: Bot, + shared: Arc, + user_id: String, + chat_id: i64, + handle: Arc, + cancel: CancellationToken, +) { + let mut rx = handle.subscribe(); + let tg_chat = ChatId(chat_id); - // Single loop: rx is updated in-place on Done/Error so we never miss events - // from the next turn (re-subscription happens before the async send). loop { let ge: GlobalEvent = tokio::select! { _ = cancel.cancelled() => { - info!("telegram: persistent forwarder stopped"); - return; + info!(user_id = %user_id, "telegram: forwarder cancelled"); + break; } result = rx.recv() => match result { Ok(e) => e, Err(broadcast::error::RecvError::Lagged(n)) => { - warn!(skipped = n, "telegram: persistent forwarder lagged"); + warn!(user_id = %user_id, skipped = n, "telegram: forwarder lagged"); continue; } - Err(broadcast::error::RecvError::Closed) => return, + Err(broadcast::error::RecvError::Closed) => { + info!(user_id = %user_id, "telegram: forwarder — user context closed, exiting"); + break; + } }, }; - // ApprovalResolved is handled regardless of source so Telegram removes its - // keyboard even when the approval was resolved via web or REST. - if let ServerEvent::ApprovalResolved { request_id, approved, .. } = ge.event { - let label = if approved { "✅ Approved" } else { "❌ Rejected" }; + // ApprovalResolved is handled regardless of source so Telegram removes + // its keyboard even when the approval was resolved via web or REST. + if let ServerEvent::ApprovalResolved { request_id, .. } = ge.event { let mut pending = shared.pending_approvals.lock().await; - if let Some((&msg_id, _)) = pending.iter().find(|(_, rid)| **rid == request_id) { + if let Some((&msg_id, _)) = pending.iter().find(|(_, pa)| pa.request_id == request_id) { let msg_id = msg_id; pending.remove(&msg_id); drop(pending); - if let Some(cid) = resolve_chat_id(&shared).await { - bot.delete_message(cid, msg_id).await.ok(); - } + bot.delete_message(tg_chat, msg_id).await.ok(); } continue; } - // All other events: only process if they belong to the "telegram" source. + // Only process events from the "telegram" source. if ge.source.as_deref() != Some("telegram") { - tracing::debug!(event_type = ge.event.type_name(), source = ?ge.source, "persistent_forwarder: skipping non-telegram event"); continue; } let event = ge.event; - tracing::debug!(event_type = event.type_name(), "persistent_forwarder: processing telegram event"); - - // Resolve the destination chat_id (last known user, or first in whitelist). - // For terminal events (Done/Error) with no known chat, still re-subscribe. - let chat_id = match resolve_chat_id(&shared).await { - Some(id) => id, - None => { - warn!(event_type = %event.type_name(), "telegram: persistent_forwarder — no chat_id resolved, dropping event"); - if matches!(event, ServerEvent::Done { .. } | ServerEvent::Error { .. }) { - rx = shared.chat_hub.events("telegram"); - } - continue; - } - }; match event { ServerEvent::Done { content, .. } => { - // Re-subscribe BEFORE any await so we don't miss the next turn. - rx = shared.chat_hub.events("telegram"); if !content.trim().is_empty() { - send_long(&bot, chat_id, &content, Some(ParseMode::Html)).await; + send_long(&bot, tg_chat, &content, Some(ParseMode::Html)).await; } } ServerEvent::Error { message } => { - rx = shared.chat_hub.events("telegram"); bot.send_message( - chat_id, + tg_chat, format!("⚠️ Error: {}", escape_html(&message)), ) .parse_mode(ParseMode::Html) @@ -140,7 +162,7 @@ pub(crate) async fn persistent_forwarder( } ServerEvent::ToolStart { label_short, .. } => { - bot.send_message(chat_id, format!("🔧 {}…", label_to_html(&label_short))) + bot.send_message(tg_chat, format!("🔧 {}…", label_to_html(&label_short))) .parse_mode(ParseMode::Html) .await .ok(); @@ -148,7 +170,7 @@ pub(crate) async fn persistent_forwarder( ServerEvent::Thinking { content, .. } => { if !content.trim().is_empty() { - send_long(&bot, chat_id, &content, Some(ParseMode::Html)).await; + send_long(&bot, tg_chat, &content, Some(ParseMode::Html)).await; } } @@ -156,7 +178,7 @@ pub(crate) async fn persistent_forwarder( let preview = prompt_preview.chars().take(300).collect::(); let ellipsis = if prompt_preview.len() > 300 { "…" } else { "" }; bot.send_message( - chat_id, + tg_chat, format!( "🤖 {} → {}\n
{}{ellipsis}
", escape_html(&parent_agent_id), @@ -173,7 +195,7 @@ pub(crate) async fn persistent_forwarder( let preview = result_preview.chars().take(300).collect::(); let ellipsis = if result_preview.len() > 300 { "…" } else { "" }; bot.send_message( - chat_id, + tg_chat, format!( "✅ {} finished → {}\n
{}{ellipsis}
", escape_html(&agent_id), @@ -195,7 +217,7 @@ pub(crate) async fn persistent_forwarder( escape_html(&path), escape_html(&preview), ); - send_approval_keyboard(&bot, chat_id, text, request_id, &shared).await; + send_approval_keyboard(&bot, tg_chat, text, user_id.clone(), request_id, &shared).await; } ServerEvent::ApprovalRequired { request_id, tool_name, arguments, .. } => { @@ -209,16 +231,15 @@ pub(crate) async fn persistent_forwarder( escape_html(&tool_name), escape_html(&args_preview), ); - send_approval_keyboard(&bot, chat_id, text, request_id, &shared).await; + send_approval_keyboard(&bot, tg_chat, text, user_id.clone(), request_id, &shared).await; } - ServerEvent::AgentQuestion { request_id, tool_call_id, title, question, suggested_answers, .. } => { - info!(request_id, tool_call_id, %question, "telegram: persistent_forwarder received AgentQuestion"); + ServerEvent::AgentQuestion { request_id, title, question, suggested_answers, .. } => { + info!(request_id, %question, "telegram: forwarder received AgentQuestion"); - // If a previous question is still pending, disable its (now-dead) - // buttons so tapping them doesn't silently no-op. - if let Some(prev) = shared.pending_question.lock().await.take() { - bot.edit_message_reply_markup(chat_id, prev.message_id) + // Disable any previously-pending question for this chat. + if let Some(prev) = shared.pending_questions.lock().await.remove(&chat_id) { + bot.edit_message_reply_markup(tg_chat, prev.message_id) .reply_markup(InlineKeyboardMarkup::new(vec![vec![ InlineKeyboardButton::callback("⏭ Superseded by a newer question", "noop"), ]])) @@ -240,27 +261,27 @@ pub(crate) async fn persistent_forwarder( .collect(); Some(InlineKeyboardMarkup::new(buttons)) }; - let mut req = bot.send_message(chat_id, header).parse_mode(ParseMode::Html); + let mut req = bot.send_message(tg_chat, header).parse_mode(ParseMode::Html); if let Some(kb) = keyboard { req = req.reply_markup(kb); } match req.await { Ok(m) => { - info!(request_id, msg_id = m.id.0, "telegram: AgentQuestion sent to user, pending_question set"); - *shared.pending_question.lock().await = Some(super::PendingQuestion { + shared.pending_questions.lock().await.insert(chat_id, super::PendingQuestion { + user_id: user_id.clone(), request_id, message_id: m.id, suggested_answers, }); } - Err(e) => error!(error = %e, request_id, "telegram: failed to send AgentQuestion to user"), + Err(e) => error!(error = %e, request_id, "telegram: failed to send AgentQuestion"), } } ServerEvent::LlmFailed { tried, last_error } => { let models = tried.join(", "); bot.send_message( - chat_id, + tg_chat, format!( "⚠️ LLM unavailable\nTried: {}\n{}", escape_html(&models), @@ -277,23 +298,14 @@ pub(crate) async fn persistent_forwarder( _ => {} } } -} -/// Resolves the Telegram chat_id to use for outbound messages. -/// Prefers the last chat_id that sent a message; falls back to the first -/// whitelisted user. -async fn resolve_chat_id(shared: &TgShared) -> Option { - if let Some(id) = *shared.home_chat_id.lock().await { - return Some(id); - } - let wl = load_wl(&shared.secrets_dir).await; - wl.whitelist.first().map(|&id| ChatId(id)) + // Clean up: remove this user from the active forwarders set. + shared.forwarders.lock().await.remove(&user_id); + info!(user_id = %user_id, "telegram: forwarder exited"); } // ── Helpers ─────────────────────────────────────────────────────────────────── -/// Truncates `s` to at most `max_chars` Unicode scalar values. -/// Appends `…` if truncated. Never panics on multibyte UTF-8 content. fn truncate_chars(s: &str, max_chars: usize) -> String { let mut chars = s.chars(); let truncated: String = chars.by_ref().take(max_chars).collect(); @@ -333,14 +345,20 @@ pub(crate) async fn callback_handler( let req_id = parts.next().and_then(|s| s.parse::().ok()); let idx_str = parts.next().and_then(|s| s.parse::().ok()); if let (Some(request_id), Some(idx)) = (req_id, idx_str) { - let mut pq = shared.pending_question.lock().await; - if let Some(pq_inner) = pq.as_ref() { - if pq_inner.request_id == request_id { - let answer = pq_inner.suggested_answers.get(idx).cloned().unwrap_or_default(); - drop(pq); - *shared.pending_question.lock().await = None; - shared.chat_hub.resolve_question("telegram", request_id, answer.clone()).await; - info!(request_id, %answer, "telegram: clarification answered via button"); + let pq_map = shared.pending_questions.lock().await; + if let Some(pq) = pq_map.get(&msg_chat_id.0) { + if pq.request_id == request_id { + let user_id = pq.user_id.clone(); + let answer = pq.suggested_answers.get(idx).cloned().unwrap_or_default(); + drop(pq_map); + shared.pending_questions.lock().await.remove(&msg_chat_id.0); + + if let Some(handle) = shared.user_channel.resolve_user(&user_id).await { + handle.chat_hub().resolve_question("telegram", request_id, answer.clone()).await; + info!(request_id, %answer, "telegram: clarification answered via button"); + } else { + warn!(user_id = %user_id, "telegram: user locked, cannot resolve clarification"); + } bot.edit_message_reply_markup(msg_chat_id, msg_id) .reply_markup(InlineKeyboardMarkup::new(vec![vec![ InlineKeyboardButton::callback(format!("✅ {answer}"), "noop"), @@ -380,20 +398,25 @@ pub(crate) async fn callback_handler( if let Some((request_id, action, label)) = parsed { let stored = shared.pending_approvals.lock().await.remove(&msg_id); - if let Some(stored_id) = stored { - if stored_id == request_id { - match action { - ApprovalAction::Approve => - shared.approval.approve(request_id).await, - ApprovalAction::Reject => - shared.approval.reject(request_id, String::new()).await, - ApprovalAction::BypassTime(secs) => - shared.approval.approve_with_bypass(request_id, Some(secs)).await, - ApprovalAction::BypassSession => - shared.approval.approve_with_bypass(request_id, None).await, + if let Some(pa) = stored { + if pa.request_id == request_id { + if let Some(handle) = shared.user_channel.resolve_user(&pa.user_id).await { + let approval = handle.approval(); + match action { + ApprovalAction::Approve => + approval.approve(request_id).await, + ApprovalAction::Reject => + approval.reject(request_id, String::new()).await, + ApprovalAction::BypassTime(secs) => + approval.approve_with_bypass(request_id, Some(secs)).await, + ApprovalAction::BypassSession => + approval.approve_with_bypass(request_id, None).await, + } + info!(request_id, label, "telegram: approval resolved"); + bot.delete_message(msg_chat_id, msg_id).await.ok(); + } else { + warn!(user_id = %pa.user_id, "telegram: user locked, cannot resolve approval"); } - info!(request_id, label, "telegram: approval resolved"); - bot.delete_message(msg_chat_id, msg_id).await.ok(); } } else { warn!(request_id, "telegram: approval not found (already resolved?)"); diff --git a/crates/plugin-telegram-bot/src/handlers.rs b/crates/plugin-telegram-bot/src/handlers.rs index b62b009..dc42ba5 100644 --- a/crates/plugin-telegram-bot/src/handlers.rs +++ b/crates/plugin-telegram-bot/src/handlers.rs @@ -8,11 +8,13 @@ use core_api::chat_hub::{ModelCommandOutcome, SendMessageOptions}; use core_api::command::expand_template; use core_api::location::GpsCoord; use core_api::message_meta::{CommandRef, MessageMetadata}; +use core_api::user_channel::UserChannelHandle; use super::TELEGRAM_FORMAT_CONTEXT; use super::TgShared; use super::attachments::TelegramAttachment; -use super::auth::{handle_pairing, load_wl}; +use super::auth::handle_pairing; +use super::events::ensure_forwarder; // ── Available commands help text (shared by /help and unknown-command replies) ── const HELP_TEXT: &str = "Available commands\n\n\ @@ -28,9 +30,6 @@ const HELP_TEXT: &str = "Available commands\n\n\ /sethome — receive agent notifications here\n\ /help — this message"; -/// Builds the `/help` text: the static system-command list plus a dynamically -/// discovered "Custom commands" section (`commands//`). Descriptions are -/// HTML-escaped since the message is sent with `ParseMode::Html`. fn help_text(command: &dyn core_api::command::CommandApi) -> String { let mut out = String::from(HELP_TEXT); let cmds = command.list_enabled(); @@ -48,9 +47,6 @@ fn help_text(command: &dyn core_api::command::CommandApi) -> String { } // ── Incoming message classification ─────────────────────────────────────────── -// -// To add a new media type: add a variant to IncomingEvent, handle it in -// classify_message, then dispatch it in message_handler. pub(crate) enum IncomingEvent { Text(String), @@ -93,13 +89,6 @@ pub(crate) fn classify_message(msg: &Message) -> Option { let text = msg.text()?; - // A command is any message that *starts* with '/'. We deliberately do NOT - // rely on teloxide's BotCommand entities: those are emitted for every - // "/token" anywhere in the text, so a normal sentence containing a "/path" - // (e.g. "stop /usr/bin/foo") would be misclassified as a command. A leading - // slash is the only signal. Arguments are parsed from the message `text` - // (not `entity.text()`, which spans only "/model" and would drop the arg). - // An unknown command is handled by the dispatcher, which replies with help. if text.starts_with('/') { let full = text.trim_start_matches('/'); let mut parts = full.splitn(2, ' '); @@ -126,31 +115,58 @@ pub(crate) async fn message_handler( ) -> ResponseResult<()> { let chat_id = msg.chat.id; - // Whitelist check — re-read the file on every message so agent edits are - // picked up without a plugin restart. - let wl = load_wl(&shared.secrets_dir).await; - if !wl.whitelist.contains(&chat_id.0) { - handle_pairing(&bot, chat_id, &shared).await; - return Ok(()); - } + // Resolve chat_id → user_id from bindings. + let user_id = match shared.user_for_chat(chat_id.0).await { + Some(uid) => uid, + None => { + handle_pairing(&bot, chat_id, &shared).await; + return Ok(()); + } + }; - // Track the last active chat_id so the persistent forwarder knows - // where to send background notifications. - *shared.home_chat_id.lock().await = Some(chat_id); + // Resolve the user's per-user context (must be unlocked, §9). + let handle = match shared.user_channel.resolve_user(&user_id).await { + Some(h) => h, + None => { + bot.send_message( + chat_id, + "🔒 Your account is locked. Please log in via the web app first, \ + then send another message.", + ) + .await + .ok(); + return Ok(()); + } + }; + + // Ensure a per-user forwarder is running so the response events reach + // this Telegram chat. The forwarder will exit on broadcast-close (user + // locked) or plugin stop; a fresh CancellationToken is fine here since + // the global plugin cancel drops the dispatcher (and thus the shared Arc). + ensure_forwarder( + bot.clone(), + Arc::clone(&shared), + &user_id, + chat_id.0, + Arc::clone(&handle), + tokio_util::sync::CancellationToken::new(), + ).await; let Some(incoming) = classify_message(&msg) else { bot.send_message(chat_id, "Unsupported message format.").await.ok(); return Ok(()); }; + let hub = handle.chat_hub(); + match incoming { IncomingEvent::Command { ref name, .. } if name == "clear" || name == "new" => { - handle_clear(&bot, chat_id, &shared).await; + handle_clear(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, .. } if name == "sethome" => { - match shared.chat_hub.set_home("telegram").await { + match hub.set_home("telegram").await { Ok(_) => { - info!("telegram: set as home source"); + info!("telegram: set as home source for user {}", user_id); bot.send_message(chat_id, "🏠 Telegram set as home. Agent notifications will be delivered here.") .parse_mode(ParseMode::Html) .await @@ -168,30 +184,28 @@ pub(crate) async fn message_handler( .ok(); } IncomingEvent::Command { ref name, .. } if name == "stop" => { - handle_stop(&bot, chat_id, &shared).await; + hub.cancel("telegram").await; + info!("telegram: agent cancelled via /stop"); + bot.send_message(chat_id, "⏹ Agent stopped.").await.ok(); } IncomingEvent::Command { ref name, .. } if name == "context" => { - handle_context(&bot, chat_id, &shared).await; + handle_context(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, .. } if name == "cost" => { - handle_cost(&bot, chat_id, &shared).await; + handle_cost(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, .. } if name == "compact" => { - handle_compact(&bot, chat_id, &shared).await; + handle_compact(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, .. } if name == "resettools" => { - handle_reset_mcp(&bot, chat_id, &shared).await; + handle_reset_mcp(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, .. } if name == "models" => { - handle_list_models(&bot, chat_id, &shared).await; + handle_list_models(&bot, chat_id, &hub).await; } IncomingEvent::Command { ref name, ref args, .. } if name == "model" => { - handle_set_model(&bot, chat_id, args, &shared).await; + handle_set_model(&bot, chat_id, args, &hub).await; } - // A recognised custom `/command` expands its `COMMAND.md` template into a - // normal user message (fully interactive: the model can then ask questions, - // iterate, dispatch sub-agents). Any other `/...` is an unknown command and - // is never forwarded to the LLM — reply with a not-found notice + help. IncomingEvent::Command { ref name, ref args, .. } => { if let Some(resolved) = shared.command.resolve(name) { let args_str = args.join(" "); @@ -208,7 +222,7 @@ pub(crate) async fn message_handler( }), ..Default::default() }; - handle_llm_message(bot, chat_id, content, Some(metadata), shared).await; + handle_llm_message(bot, chat_id, content, Some(metadata), shared, &handle).await; } else { bot.send_message( chat_id, @@ -220,10 +234,10 @@ pub(crate) async fn message_handler( } } IncomingEvent::Voice { file_id } => { - handle_voice(&bot, chat_id, file_id, &shared).await; + handle_voice(&bot, chat_id, file_id, &shared, &handle).await; } IncomingEvent::Attachment(attachment) => { - handle_attachment(bot, chat_id, attachment, shared).await; + handle_attachment(bot, chat_id, attachment, shared, &handle).await; } _ => { let text = match &incoming { @@ -233,14 +247,15 @@ pub(crate) async fn message_handler( | IncomingEvent::Attachment(_) => unreachable!(), }; - // If a clarification question is pending, treat any text as the answer. + // If a clarification question is pending for this chat, treat any + // text as the answer. { - let mut pq = shared.pending_question.lock().await; - if let Some(pq_inner) = pq.take() { - let request_id = pq_inner.request_id; - let question_msg_id = pq_inner.message_id; - drop(pq); - shared.chat_hub.resolve_question("telegram", request_id, text.clone()).await; + let mut pq_map = shared.pending_questions.lock().await; + if let Some(pq) = pq_map.remove(&chat_id.0) { + let request_id = pq.request_id; + let question_msg_id = pq.message_id; + drop(pq_map); + hub.resolve_question("telegram", request_id, text.clone()).await; tracing::info!(request_id, %text, "telegram: clarification answered via text"); bot.edit_message_reply_markup(chat_id, question_msg_id) .reply_markup(teloxide::types::InlineKeyboardMarkup::new(vec![vec![ @@ -255,17 +270,17 @@ pub(crate) async fn message_handler( } } - handle_llm_message(bot, chat_id, text, None, shared).await; + handle_llm_message(bot, chat_id, text, None, shared, &handle).await; } } Ok(()) } -// ── /clear command ──────────────────────────────────────────────────────────── +// ── Command handlers ────────────────────────────────────────────────────────── -async fn handle_clear(bot: &Bot, chat_id: ChatId, shared: &Arc) { - match shared.chat_hub.clear("telegram").await { +async fn handle_clear(bot: &Bot, chat_id: ChatId, hub: &Arc) { + match hub.clear("telegram").await { Ok(_) => { info!("telegram: session cleared via /clear"); bot.send_message(chat_id, "🆕 New conversation started.").await.ok(); @@ -277,10 +292,8 @@ async fn handle_clear(bot: &Bot, chat_id: ChatId, shared: &Arc) { } } -// ── /context command ────────────────────────────────────────────────────────── - -async fn handle_context(bot: &Bot, chat_id: ChatId, shared: &Arc) { - match shared.chat_hub.context_info("telegram").await { +async fn handle_context(bot: &Bot, chat_id: ChatId, hub: &Arc) { + match hub.context_info("telegram").await { Ok((input, output)) => { let input_str = input.map_or("?".to_string(), |t| t.to_string()); let output_str = output.map_or("?".to_string(), |t| t.to_string()); @@ -298,10 +311,8 @@ async fn handle_context(bot: &Bot, chat_id: ChatId, shared: &Arc) { } } -// ── /cost command ───────────────────────────────────────────────────────────── - -async fn handle_cost(bot: &Bot, chat_id: ChatId, shared: &Arc) { - match shared.chat_hub.cost_info("telegram").await { +async fn handle_cost(bot: &Bot, chat_id: ChatId, hub: &Arc) { + match hub.cost_info("telegram").await { Ok(Some(c)) => { bot.send_message(chat_id, format!("💰 Session cost: ${c:.4}")).await.ok(); } @@ -314,10 +325,8 @@ async fn handle_cost(bot: &Bot, chat_id: ChatId, shared: &Arc) { } } -// ── /compact command ────────────────────────────────────────────────────────── - -async fn handle_compact(bot: &Bot, chat_id: ChatId, shared: &Arc) { - match shared.chat_hub.force_compact("telegram").await { +async fn handle_compact(bot: &Bot, chat_id: ChatId, hub: &Arc) { + match hub.force_compact("telegram").await { Ok(true) => { info!("telegram: manual compaction succeeded"); bot.send_message(chat_id, "✅ Context compacted.").await.ok(); @@ -332,10 +341,8 @@ async fn handle_compact(bot: &Bot, chat_id: ChatId, shared: &Arc) { } } -// ── /resettools command ─────────────────────────────────────────────────────── - -async fn handle_reset_mcp(bot: &Bot, chat_id: ChatId, shared: &Arc) { - match shared.chat_hub.reset_mcp("telegram").await { +async fn handle_reset_mcp(bot: &Bot, chat_id: ChatId, hub: &Arc) { + match hub.reset_mcp("telegram").await { Ok(()) => { info!("telegram: tool-group grants reset via /resettools"); bot.send_message(chat_id, "✅ Activated tool groups removed from the session.").await.ok(); @@ -347,23 +354,8 @@ async fn handle_reset_mcp(bot: &Bot, chat_id: ChatId, shared: &Arc) { } } -// ── /stop command ──────────────────────────────────────────────────────────── - -async fn handle_stop(bot: &Bot, chat_id: ChatId, shared: &Arc) { - shared.chat_hub.cancel("telegram").await; - info!("telegram: agent cancelled via /stop"); - bot.send_message(chat_id, "⏹ Agent stopped.").await.ok(); -} - -// ── /models and /model commands ────────────────────────────────────────────── -// -// Business logic (resolve arg, mutate pin, broadcast) lives in -// `ChatHub::apply_model_command` / `ChatHub::list_clients_marked`. Here we only -// format for Telegram (HTML) and send via the bot — same pattern the web WS -// handler uses with Markdown. - -async fn handle_list_models(bot: &Bot, chat_id: ChatId, shared: &Arc) { - let items = shared.chat_hub.list_clients_marked("telegram").await; +async fn handle_list_models(bot: &Bot, chat_id: ChatId, hub: &Arc) { + let items = hub.list_clients_marked("telegram").await; let mut text = String::from("Available models\n\n"); for (i, name, is_current) in &items { let marker = if *is_current { "●" } else { "○" }; @@ -381,9 +373,9 @@ async fn handle_list_models(bot: &Bot, chat_id: ChatId, shared: &Arc) .ok(); } -async fn handle_set_model(bot: &Bot, chat_id: ChatId, args: &[String], shared: &Arc) { +async fn handle_set_model(bot: &Bot, chat_id: ChatId, args: &[String], hub: &Arc) { let arg = args.first().cloned().unwrap_or_default(); - let outcome = shared.chat_hub.apply_model_command("telegram", &arg).await; + let outcome = hub.apply_model_command("telegram", &arg).await; let text = match outcome { ModelCommandOutcome::Set(name) => format!("✅ Model set: {}", super::helpers::escape_html(&name)), ModelCommandOutcome::Cleared => "✅ Model reset to auto.".to_string(), @@ -403,25 +395,22 @@ async fn handle_llm_message( text: String, metadata: Option, shared: Arc, + handle: &Arc, ) { bot.send_chat_action(chat_id, ChatAction::Typing).await.ok(); - // The persistent_forwarder (spawned once in start()) is always subscribed - // to the "telegram" broadcast channel and will pick up all events for this - // turn — including Done → send to Telegram. No per-message subscription needed. - let client_name = shared.chat_hub.get_selected_client("telegram").await; + let hub = handle.chat_hub(); + let client_name = hub.get_selected_client("telegram").await; let opts = SendMessageOptions { client_name, extra_system_context: Some(TELEGRAM_FORMAT_CONTEXT.to_string()), tail_reminder: Some(super::TELEGRAM_FORMAT_REMINDER.to_string()), - interface_tools: super::tools::interface_tools(bot, chat_id, &*shared.tts).await, + interface_tools: super::tools::interface_tools(bot.clone(), chat_id, &*shared.tts).await, metadata, ..Default::default() }; - // send_message only enqueues — the turn runs on ChatHub's per-source consumer — - // so awaiting inline keeps this message handler responsive. - if let Err(e) = shared.chat_hub.send_message("telegram", &text, opts).await { + if let Err(e) = hub.send_message("telegram", &text, opts).await { error!(error = %e, "telegram: enqueue error"); } } @@ -433,6 +422,7 @@ async fn handle_voice( chat_id: ChatId, file_id: String, shared: &Arc, + handle: &Arc, ) { use teloxide::net::Download; @@ -477,7 +467,7 @@ async fn handle_voice( The user sent a voice message. The following is the audio transcript:\n\n\ {text}" ); - handle_llm_message(bot.clone(), chat_id, message, None, Arc::clone(shared)).await; + handle_llm_message(bot.clone(), chat_id, message, None, Arc::clone(shared), handle).await; } // ── Edited message (live location updates) ──────────────────────────────────── @@ -500,8 +490,8 @@ async fn handle_attachment( chat_id: ChatId, attachment: TelegramAttachment, shared: Arc, + handle: &Arc, ) { - // Update LocationManager immediately, before any LLM dispatch. if let TelegramAttachment::Location { latitude, longitude, accuracy, is_live } = &attachment { let coord = GpsCoord { latitude: *latitude, longitude: *longitude }; shared.location.update("telegram", coord, *accuracy, *is_live); @@ -519,9 +509,6 @@ async fn handle_attachment( }; match saved { - // Document / Photo: carry the file as structured metadata (rendered as a - // chip in the copilot UI; the LLM gets the shared [SYSTEM INFO] block). - // The caption, if any, becomes the user's text for this turn. Some(att) => { info!(chat_id = chat_id.0, path = %att.path, "telegram: attachment saved, forwarding to LLM"); let caption = match &attachment { @@ -530,12 +517,11 @@ async fn handle_attachment( TelegramAttachment::Location { .. } => None, }.unwrap_or_default(); let metadata = MessageMetadata { attachments: vec![att], ..Default::default() }; - handle_llm_message(bot, chat_id, caption, Some(metadata), shared).await; + handle_llm_message(bot, chat_id, caption, Some(metadata), shared, handle).await; } - // Location (no file): keep the textual system-info block. None => { let message = attachment.system_info_message(None); - handle_llm_message(bot, chat_id, message, None, shared).await; + handle_llm_message(bot, chat_id, message, None, shared, handle).await; } } } diff --git a/crates/plugin-telegram-bot/src/lib.rs b/crates/plugin-telegram-bot/src/lib.rs index 56efca0..359d12d 100644 --- a/crates/plugin-telegram-bot/src/lib.rs +++ b/crates/plugin-telegram-bot/src/lib.rs @@ -1,22 +1,29 @@ /// Telegram plugin — connects the Skald LLM to a private Telegram bot. /// +/// # Multi-user architecture (blueprint §13) +/// +/// One bot serves many Telegram chats, each bound to a Skald user via the +/// `chat_id ↔ user_id` pairing stored in the config table (key `"telegram"`). +/// Incoming messages resolve the user's per-user context via +/// [`UserChannelApi`], then dispatch through that user's `ChatHub`. A per-user +/// forwarder subscribes to the user's event stream and routes `ServerEvent`s +/// back to the bound Telegram chat. +/// /// # Pairing -/// Unknown users receive a pairing code in chat. The code is also written to -/// `secrets/telegram_whitelist.json` under `pending_pairings`. The main agent -/// (via `read_file` / `write_file`) can inspect that file and move the -/// `chat_id` into the `whitelist` array to complete the authorisation — no -/// code changes required, just a file edit. +/// +/// Unknown chats receive a pairing code. The admin's agent calls the +/// `telegram_pairing` tool (category `Config`) to bind the `chat_id` to a +/// `user_id`. The binding is written to the config table; the resulting +/// `ConfigKeyUpdated` event reloads the in-memory cache instantly. /// /// # Human-in-the-loop approvals -/// Tool calls requiring approval emit a `PendingWrite` event; the plugin -/// forwards it to Telegram as an inline-keyboard message with -/// [✅ Approve] [❌ Reject] / [⏱ 15 min] [🔄 Session] buttons. /// -/// # Adding new message types -/// 1. Add a variant to `IncomingEvent` in `handlers.rs`. -/// 2. Handle it in `classify_message` (same file). -/// 3. Dispatch it in `message_handler` (same file). -use std::collections::HashMap; +/// Tool calls requiring approval emit a `PendingWrite` / `ApprovalRequired` +/// event; the per-user forwarder sends it to Telegram as an inline-keyboard +/// message. Button presses resolve the approval through that user's +/// `ApprovalApi`. + +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -26,18 +33,18 @@ use async_trait::async_trait; use serde_json::{Value, json}; use teloxide::prelude::*; use teloxide::types::MessageId; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, RwLock}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; -use core_api::approval::ApprovalApi; -use core_api::chat_hub::ChatHubApi; use core_api::command::CommandApi; +use core_api::config_api::ConfigApi; use core_api::location::LocationUpdater; use core_api::plugin::{Plugin, PluginContext}; -use core_api::transcribe::{Transcribe, TranscribeProvider}; +use core_api::transcribe::TranscribeProvider; use core_api::tts::TtsProvider; +use core_api::user_channel::UserChannelApi; mod attachments; mod auth; @@ -56,7 +63,6 @@ FORBIDDEN (will appear as raw symbols): ** * _ ` # | and Markdown tables.\n\ • Headers → text\n\ • Structured data → bullet lists with •, never | tables\n\ • Escape & < > as & < >"; - /// Short reminder injected near the end of the message list to counter /// instruction drift in long conversations. pub(crate) const TELEGRAM_FORMAT_REMINDER: &str = "\ @@ -67,61 +73,90 @@ No Markdown: no ** * _ ` # |. No tables — use bullet lists."; /// A pending `ask_user_clarification` question waiting for the user's reply. pub(crate) struct PendingQuestion { - pub(crate) request_id: i64, - pub(crate) message_id: MessageId, + pub(crate) user_id: String, + pub(crate) request_id: i64, + pub(crate) message_id: MessageId, /// Suggested answers (used to resolve the selection when the user taps a button). pub(crate) suggested_answers: Vec, } +/// A pending tool-call approval shown as an inline keyboard. +pub(crate) struct PendingApproval { + pub(crate) user_id: String, + pub(crate) request_id: i64, +} + +/// Global state shared across all Telegram handlers and the per-user forwarders. +/// +/// Per-user state (ChatHub, ApprovalApi, event stream) is resolved at runtime +/// via [`UserChannelApi`] — it is NOT held here. Only global capabilities and +/// pairing/multiplexing state live in `TgShared`. pub(crate) struct TgShared { - pub(crate) chat_hub: Arc, - /// Custom slash-command resolver (`commands//`). Read-only: lets the - /// bot expand a recognised `/command` into a template before forwarding it to - /// the LLM, mirroring the WS handler. - pub(crate) command: Arc, - pub(crate) approval: Arc, - pub(crate) transcribe: Arc, - pub(crate) tts: Arc, - pub(crate) location: Arc, - /// MessageId of the approval message → request_id. - pub(crate) pending_approvals: Mutex>, - /// Currently active clarification question (at most one at a time per session). - pub(crate) pending_question: Mutex>, - pub(crate) secrets_dir: PathBuf, - /// Base directory for file attachments: `/uploads/telegram/`. - pub(crate) uploads_dir: PathBuf, - /// Last chat_id that sent a message — used as the target for background notifications. - /// Set on every incoming message; read by the persistent event forwarder. - pub(crate) home_chat_id: Mutex>, + // ── Global capabilities ── + pub(crate) user_channel: Arc, + pub(crate) command: Arc, + pub(crate) config: Arc, + pub(crate) transcribe: Arc, + pub(crate) tts: Arc, + pub(crate) location: Arc, + pub(crate) uploads_dir: PathBuf, + + // ── Pairing / bindings (config-table-backed, cached in memory) ── + pub(crate) bindings: RwLock, + + // ── Per-chat pending state ── + /// Approval message_id → pending approval (carries user_id for routing). + pub(crate) pending_approvals: Mutex>, + /// chat_id → pending clarification question (at most one per chat). + pub(crate) pending_questions: Mutex>, + + // ── Forwarder tracking ── + /// user_ids with an active per-user forwarder task. + pub(crate) forwarders: Mutex>, } impl TgShared { - pub(crate) async fn transcriber(&self) -> Option> { + pub(crate) async fn transcriber(&self) -> Option> { self.transcribe.get().await } + + /// Looks up the `user_id` bound to a Telegram `chat_id`, if any. + pub(crate) async fn user_for_chat(&self, chat_id: i64) -> Option { + self.bindings.read().await + .bindings.iter() + .find(|b| b.chat_id == chat_id) + .map(|b| b.user_id.clone()) + } } // ── Plugin struct ───────────────────────────────────────────────────────────── pub struct TelegramPlugin { - secrets_dir: PathBuf, /// Bot token — set by reload() before start() is called. token: Mutex, running: Arc, cancel: Mutex>, handle: Mutex>>, + /// Runtime shared state, populated by `start()`. Accessible to the pairing + /// tool so it can write bindings before/after the dispatcher is running. + shared: std::sync::OnceLock>, } impl TelegramPlugin { - pub fn new(secrets_dir: impl Into) -> Self { + pub fn new() -> Self { Self { - secrets_dir: secrets_dir.into(), - token: Mutex::new(String::new()), - running: Arc::new(AtomicBool::new(false)), - cancel: Mutex::new(None), - handle: Mutex::new(None), + token: Mutex::new(String::new()), + running: Arc::new(AtomicBool::new(false)), + cancel: Mutex::new(None), + handle: Mutex::new(None), + shared: std::sync::OnceLock::new(), } } + + /// Returns the shared runtime state if the plugin is running. + pub(crate) fn shared(&self) -> Option<&Arc> { + self.shared.get() + } } #[async_trait] @@ -149,7 +184,6 @@ impl Plugin for TelegramPlugin { } fn as_any(&self) -> &dyn std::any::Any { self } - fn as_arc_any(self: Arc) -> Arc { self } async fn reload(&self, enabled: bool, config: Value, ctx: PluginContext) -> Result<()> { @@ -189,49 +223,59 @@ impl Plugin for TelegramPlugin { anyhow::bail!("telegram: token is empty — set it via the plugins API"); } - let uploads_dir = self.secrets_dir - .parent() - .unwrap_or(std::path::Path::new(".")) + let uploads_dir = std::env::current_dir() + .unwrap_or_default() .join("uploads") .join("telegram"); - // Register "telegram" source with ChatHub (idempotent). - // ChatHub restores the active session from the sources table automatically. - ctx.chat_hub.register("telegram").await; - info!("telegram: registered with ChatHub"); + // Load bindings from the config table (or default if absent). + let telegram_config = auth::load_config(&*ctx.config).await + .unwrap_or_default(); + info!( + bindings = telegram_config.bindings.len(), + pending = telegram_config.pending_pairings.len(), + "telegram: config loaded", + ); let shared = Arc::new(TgShared { - chat_hub: Arc::clone(&ctx.chat_hub), + user_channel: Arc::clone(&ctx.user_channel), command: Arc::clone(&ctx.command), - approval: Arc::clone(&ctx.approval), + config: Arc::clone(&ctx.config), transcribe: Arc::clone(&ctx.transcribe), tts: Arc::clone(&ctx.tts_provider), location: Arc::clone(&ctx.location), - pending_approvals: Mutex::new(HashMap::new()), - pending_question: Mutex::new(None), - secrets_dir: self.secrets_dir.clone(), uploads_dir, - home_chat_id: Mutex::new(None), + bindings: RwLock::new(telegram_config), + pending_approvals: Mutex::new(HashMap::new()), + pending_questions: Mutex::new(HashMap::new()), + forwarders: Mutex::new(HashSet::new()), }); + let _ = self.shared.set(Arc::clone(&shared)); + let bot = Bot::new(&token); let cancel = CancellationToken::new(); - tokio::spawn(events::persistent_forwarder( - bot.clone(), - Arc::clone(&shared), - cancel.clone(), - )); + // Config listener: reloads bindings when the "telegram" config key + // changes (e.g. the pairing tool writes a new binding). + { + let shared_c = Arc::clone(&shared); + let cancel_c = cancel.clone(); + let bus_rx = ctx.system_bus.subscribe(); + tokio::spawn(auth::config_listener(shared_c, bus_rx, cancel_c)); + } - let hub_clone = Arc::clone(&ctx.chat_hub); - tokio::spawn(async move { - if let Err(e) = hub_clone.resume("telegram").await { - tracing::warn!(error = %e, "telegram: startup resume failed"); - } - }); + // Spawn forwarders for already-unlocked paired users. + { + let shared_c = Arc::clone(&shared); + let bot_c = bot.clone(); + let cancel_c = cancel.clone(); + tokio::spawn(async move { + events::spawn_forwarders_for_bound_users(&bot_c, &shared_c, &cancel_c).await; + }); + } let cancel_clone = cancel.clone(); - let cancel_wdg = cancel.clone(); let running_clone = Arc::clone(&self.running); self.running.store(true, Ordering::Relaxed); @@ -240,9 +284,6 @@ impl Plugin for TelegramPlugin { .branch(Update::filter_edited_message().endpoint(handlers::edited_message_handler)) .branch(Update::filter_callback_query().endpoint(events::callback_handler)); - let secrets_dir_wdg = self.secrets_dir.clone(); - let bot_wdg = bot.clone(); - let task = tokio::spawn(async move { let mut dispatcher = Dispatcher::builder(bot, handler) .dependencies(dptree::deps![shared]) @@ -250,9 +291,8 @@ impl Plugin for TelegramPlugin { info!("telegram plugin: dispatcher starting"); tokio::select! { - _ = cancel_clone.cancelled() => info!("telegram plugin: cancellation received"), - _ = dispatcher.dispatch() => warn!("telegram plugin: dispatcher exited unexpectedly"), - _ = auth::whitelist_watchdog(bot_wdg, secrets_dir_wdg, cancel_wdg) => {} + _ = cancel_clone.cancelled() => info!("telegram plugin: cancellation received"), + _ = dispatcher.dispatch() => warn!("telegram plugin: dispatcher exited unexpectedly"), } running_clone.store(false, Ordering::Relaxed); info!("telegram plugin: stopped"); @@ -273,4 +313,8 @@ impl Plugin for TelegramPlugin { self.running.store(false, Ordering::Relaxed); Ok(()) } + + fn tools(self: Arc) -> Vec> { + vec![Arc::new(tools::TelegramPairingTool::new(self))] + } } diff --git a/crates/plugin-telegram-bot/src/tools.rs b/crates/plugin-telegram-bot/src/tools.rs index 384e94a..4683be5 100644 --- a/crates/plugin-telegram-bot/src/tools.rs +++ b/crates/plugin-telegram-bot/src/tools.rs @@ -1,12 +1,17 @@ use std::sync::Arc; -use serde_json::json; +use anyhow::Result; +use serde_json::{json, Value}; use teloxide::prelude::*; use teloxide::types::InputFile; use core_api::interface_tool::InterfaceTool; +use core_api::tool::{Tool, ToolCategory, ToolDescriptionLength}; use core_api::tts::{TextToSpeech, TtsProvider}; +use super::auth::{Binding, load_config, save_config}; +use super::TelegramPlugin; + /// Returns all LLM-callable tools available in a Telegram session. /// /// Each tool captures `bot` and `chat_id` so its handler can send content @@ -225,3 +230,155 @@ async fn to_ogg_opus(audio: Vec, format: &str) -> anyhow::Result> { } Ok(out.stdout) } + +// ── telegram_pairing (registry tool, category Config) ───────────────────────── + +/// Tool that binds a Telegram `chat_id` to a Skald `user_id`. +/// +/// Category `Config` — excluded from the default tool list, activated +/// explicitly by the admin's agent. The admin calls this after a user reports +/// their pairing code from Telegram. +/// +/// The binding is written to the config table (key `"telegram"`); the +/// resulting `ConfigKeyUpdated` event reloads the plugin's in-memory cache +/// instantly. +pub struct TelegramPairingTool { + plugin: Arc, +} + +impl TelegramPairingTool { + pub fn new(plugin: Arc) -> Self { + Self { plugin } + } +} + +impl Tool for TelegramPairingTool { + fn name(&self) -> &str { "telegram_pairing" } + fn category(&self) -> ToolCategory { ToolCategory::Config } + + fn description(&self) -> &str { + "Bind a Telegram chat to a Skald user so they can chat with the agent via Telegram. \ + Use `action: \"bind\"` with either a `code` (from the pairing message the user received) \ + or a `chat_id` + `user_id`. Use `action: \"list\"` to see current bindings. \ + Use `action: \"unbind\"` with a `chat_id` to remove a binding." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["bind", "unbind", "list"], + "description": "bind: create a chat_id→user_id binding. unbind: remove it. list: show all bindings.", + "default": "bind" + }, + "code": { + "type": "string", + "description": "Pairing code shown to the Telegram user (alternative to chat_id+user_id)." + }, + "chat_id": { + "type": "integer", + "description": "Telegram chat id (use when not resolving via code)." + }, + "user_id": { + "type": "string", + "description": "Skald user id to bind to (required for bind when not using code)." + } + } + }) + } + + fn describe(&self, _args: &Value, _length: ToolDescriptionLength) -> String { + "telegram_pairing".to_string() + } + + fn execute(&self, args: Value) -> Result { + let shared = self.plugin.shared() + .ok_or_else(|| anyhow::anyhow!("telegram: plugin is not running"))? + .clone(); + + let action = args.get("action") + .and_then(Value::as_str) + .unwrap_or("bind"); + + // Block on since Tool::execute is sync. + let rt = tokio::runtime::Handle::try_current() + .map_err(|e| anyhow::anyhow!("telegram_pairing: no tokio runtime: {e}"))?; + + rt.block_on(async { + let cfg_api = &*shared.config; + + match action { + "list" => { + let cfg = load_config(cfg_api).await.unwrap_or_default(); + if cfg.bindings.is_empty() { + return Ok("No Telegram bindings.".to_string()); + } + let lines: Vec = cfg.bindings.iter() + .map(|b| format!(" chat_id={} → user_id={}{}", b.chat_id, b.user_id, + b.display.as_ref().map(|d| format!(" ({d})")).unwrap_or_default())) + .collect(); + Ok(format!("Telegram bindings:\n{}", lines.join("\n"))) + } + + "unbind" => { + let chat_id = args.get("chat_id") + .and_then(Value::as_i64) + .ok_or_else(|| anyhow::anyhow!("telegram_pairing: `chat_id` required for unbind"))?; + + let mut cfg = load_config(cfg_api).await.unwrap_or_default(); + let before = cfg.bindings.len(); + cfg.bindings.retain(|b| b.chat_id != chat_id); + if cfg.bindings.len() == before { + return Ok(format!("chat_id {chat_id} is not bound.")); + } + save_config(cfg_api, &cfg).await?; + Ok(format!("Unbound chat_id {chat_id}.")) + } + + "bind" => { + let mut cfg = load_config(cfg_api).await.unwrap_or_default(); + + // Resolve chat_id + user_id either from a pairing code or + // from explicit arguments. + let (chat_id, user_id) = if let Some(code) = args.get("code").and_then(Value::as_str) { + let entry = cfg.pending_pairings.iter() + .find(|e| e.code == code) + .ok_or_else(|| anyhow::anyhow!("telegram_pairing: code '{code}' not found (it may have expired or already been used)"))?; + let chat_id = entry.chat_id; + let user_id = args.get("user_id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("telegram_pairing: `user_id` required (the code only identifies the chat)"))? + .to_string(); + // Remove the used pairing entry. + cfg.pending_pairings.retain(|e| e.code != code); + (chat_id, user_id) + } else { + let chat_id = args.get("chat_id") + .and_then(Value::as_i64) + .ok_or_else(|| anyhow::anyhow!("telegram_pairing: either `code` or `chat_id`+`user_id` required"))?; + let user_id = args.get("user_id") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("telegram_pairing: `user_id` required"))? + .to_string(); + (chat_id, user_id) + }; + + // Replace any existing binding for this chat_id. + cfg.bindings.retain(|b| b.chat_id != chat_id); + cfg.bindings.push(Binding { + chat_id, + user_id: user_id.clone(), + display: None, + }); + + save_config(cfg_api, &cfg).await?; + Ok(format!("Bound Telegram chat_id {chat_id} to user_id {user_id}.")) + } + + other => Err(anyhow::anyhow!("telegram_pairing: unknown action '{other}'")), + } + }) + } +} diff --git a/crates/skald-core/src/config_store.rs b/crates/skald-core/src/config_store.rs index c14c24e..2101489 100644 --- a/crates/skald-core/src/config_store.rs +++ b/crates/skald-core/src/config_store.rs @@ -2,13 +2,16 @@ use std::sync::Arc; use sqlx::SqlitePool; +use core_api::system_bus::{SystemEvent, SystemEventBus}; + pub struct GlobalConfigManager { - pool: Arc, + pool: Arc, + system_bus: Arc, } impl GlobalConfigManager { - pub fn new(pool: Arc) -> Self { - Self { pool } + pub fn new(pool: Arc, system_bus: Arc) -> Self { + Self { pool, system_bus } } pub async fn get(&self, key: &str) -> anyhow::Result> { @@ -19,7 +22,16 @@ impl GlobalConfigManager { Ok(row.map(|(v,)| v)) } + /// Sets a config key and emits [`SystemEvent::ConfigKeyUpdated`] on the + /// system bus when the value actually changes. No-op (no write, no event) + /// when the new value equals the current one. pub async fn set(&self, key: &str, value: &str) -> anyhow::Result<()> { + let old_value = self.get(key).await?; + + if old_value.as_deref() == Some(value) { + return Ok(()); + } + sqlx::query( "INSERT INTO config (key, value, updated_at) VALUES (?, ?, datetime('now')) ON CONFLICT(key) DO UPDATE SET @@ -30,6 +42,13 @@ impl GlobalConfigManager { .bind(value) .execute(&*self.pool) .await?; + + self.system_bus.send(SystemEvent::ConfigKeyUpdated { + key: key.to_string(), + old_value, + new_value: value.to_string(), + }); + Ok(()) } @@ -41,3 +60,13 @@ impl GlobalConfigManager { Ok(()) } } + +#[async_trait::async_trait] +impl core_api::config_api::ConfigApi for GlobalConfigManager { + async fn get(&self, key: &str) -> anyhow::Result> { + GlobalConfigManager::get(self, key).await + } + async fn set(&self, key: &str, value: &str) -> anyhow::Result<()> { + GlobalConfigManager::set(self, key, value).await + } +} diff --git a/crates/skald-core/src/plugin/mod.rs b/crates/skald-core/src/plugin/mod.rs index 15381c7..ccb43fd 100644 --- a/crates/skald-core/src/plugin/mod.rs +++ b/crates/skald-core/src/plugin/mod.rs @@ -97,6 +97,7 @@ impl PluginManager { Ok(PluginContext { command: Arc::clone(skald.command_manager()) as _, + config: Arc::clone(skald.config()) as Arc, db: Arc::clone(skald.db()), secrets: Arc::clone(skald.secrets()) as _, transcribe: Arc::clone(skald.transcribe_manager()) as _, @@ -107,6 +108,7 @@ impl PluginManager { api_provider_registry: Arc::clone(skald.provider_registry()) as _, location: Arc::clone(skald.location_manager()) as _, system_bus: Arc::clone(skald.system_bus()), + user_channel: self.skald()? as Arc, web_port, remote_slot: Arc::clone(skald.remote()), router_factory, diff --git a/crates/skald-core/src/skald/accessors.rs b/crates/skald-core/src/skald/accessors.rs index 1f57c15..2b3c10b 100644 --- a/crates/skald-core/src/skald/accessors.rs +++ b/crates/skald-core/src/skald/accessors.rs @@ -16,6 +16,7 @@ use tokio_util::sync::CancellationToken; use core_api::remote::RemoteAccess; use core_api::system_bus::SystemEventBus; +use core_api::user_channel::UserChannelApi; use crate::approval::ApprovalManager; use crate::chat_event_bus::ChatEventBus; @@ -112,3 +113,15 @@ impl Skald { pub fn location_manager(&self) -> &Arc { &self.infra.location_manager } pub fn remote(&self) -> &Arc>>> { &self.infra.remote } } + +// ── UserChannelApi ──────────────────────────────────────────────────────────── + +use super::user_context::UserContextHandle; + +#[async_trait::async_trait] +impl UserChannelApi for Skald { + async fn resolve_user(&self, user_id: &str) -> Option> { + let ctx = self.user_context(user_id).await?; + Some(std::sync::Arc::new(UserContextHandle::new(ctx))) + } +} diff --git a/crates/skald-core/src/skald/runtime.rs b/crates/skald-core/src/skald/runtime.rs index d0282a3..033b708 100644 --- a/crates/skald-core/src/skald/runtime.rs +++ b/crates/skald-core/src/skald/runtime.rs @@ -45,14 +45,14 @@ pub(super) struct Runtime { impl Runtime { /// Wires the cross-cutting primitives. Infallible. pub(super) fn bootstrap(pool: Arc) -> Self { - let config = Arc::new(GlobalConfigManager::new(Arc::clone(&pool))); + let system_bus = Arc::new(SystemEventBus::new()); + info!("system event bus ready"); + + let config = Arc::new(GlobalConfigManager::new(Arc::clone(&pool), Arc::clone(&system_bus))); let users = Arc::new(UserManager::new(Arc::clone(&pool))); let sessions = Arc::new(SessionStore::new(Arc::clone(&users))); - let system_bus = Arc::new(SystemEventBus::new()); - info!("system event bus ready"); - let event_bus = Arc::new(ChatEventBus::new()); info!("chat event bus ready"); diff --git a/crates/skald-core/src/skald/user_context.rs b/crates/skald-core/src/skald/user_context.rs index 5694e71..bdd208f 100644 --- a/crates/skald-core/src/skald/user_context.rs +++ b/crates/skald-core/src/skald/user_context.rs @@ -29,8 +29,11 @@ use sqlx::SqlitePool; use tokio::sync::{broadcast, Mutex}; use tokio_util::sync::CancellationToken; +use core_api::approval::ApprovalApi; +use core_api::chat_hub::ChatHubApi; use core_api::events::GlobalEvent; use core_api::system_bus::SystemEventBus; +use core_api::user_channel::UserChannelHandle; use crate::approval::ApprovalManager; use crate::chat_event_bus::ChatEventBus; @@ -257,3 +260,38 @@ impl UserContextRegistry { Ok(ctx) } } + +// ── UserChannelHandle impl ──────────────────────────────────────────────────── + +/// Concrete [`UserChannelHandle`] wrapping a live [`UserContext`]. +/// +/// Constructed by [`Skald`](super::Skald) when resolving a user for a channel +/// plugin. The concrete type stays private — callers receive +/// `Arc`. +pub(super) struct UserContextHandle { + ctx: Arc, +} + +impl UserContextHandle { + pub(super) fn new(ctx: Arc) -> Self { + Self { ctx } + } +} + +impl UserChannelHandle for UserContextHandle { + fn user_id(&self) -> &str { + &self.ctx.user_id + } + + fn chat_hub(&self) -> Arc { + Arc::clone(&self.ctx.chat_hub) as Arc + } + + fn approval(&self) -> Arc { + Arc::clone(&self.ctx.approval) as Arc + } + + fn subscribe(&self) -> broadcast::Receiver { + self.ctx.global_tx.subscribe() + } +} diff --git a/src/frontend/api/config.rs b/src/frontend/api/config.rs index e3e45ec..30b4600 100644 --- a/src/frontend/api/config.rs +++ b/src/frontend/api/config.rs @@ -9,7 +9,6 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use core_api::PropertyType; -use core_api::system_bus::SystemEvent; use skald_core::skald::Skald; use super::ApiError; @@ -108,20 +107,9 @@ pub async fn set_property( return Err(ApiError::not_found("unknown config key")); } - let old_value = skald.config().get(&p.key).await?; - - // No-op if value didn't change. - if old_value.as_deref() == Some(body.value.as_str()) { - return Ok(StatusCode::OK); - } - + // GlobalConfigManager::set handles the no-op check and emits + // ConfigKeyUpdated on the system bus when the value actually changes. skald.config().set(&p.key, &body.value).await?; - skald.system_bus().send(SystemEvent::ConfigKeyUpdated { - key: p.key.clone(), - old_value, - new_value: body.value, - }); - Ok(StatusCode::OK) } diff --git a/src/main.rs b/src/main.rs index 9567831..53453fa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -164,12 +164,9 @@ pub async fn run_backend() -> Result { /// Build the plugin list. Extracted so both entry points share the same set. fn build_plugins() -> Vec> { - // NOTE (multi-user slice): telegram-bot, mobile-connector and honcho are - // single-user / global-ChatEventBus-subscriber plugins. They assume one user - // and cannot work under per-user isolation yet, so they are dropped from the - // build until they become multi-user-aware. See blueprint §17 and the plan. let mut plugins: Vec> = vec![ Arc::new(plugin_tailscale_remote::RemotePlugin::new()), + Arc::new(plugin_telegram_bot::TelegramPlugin::new()), Arc::new(plugin_comfyui::ComfyUIPlugin::new()), Arc::new(plugin_tts_orpheus_3b::OrpheusTtsPlugin::new()), Arc::new(plugin_tts_kokoro::KokoroTtsPlugin::new()),