Richard ZinkandClaude Opus 5.5
05c215f685
Automations
- Job files hold only the instructions; name, schedule, mode, model, approval
mode and time limit are settings in the database (jobs.settings). Existing
files are adopted on start-up (frontmatter into the database and removed from
the file); plain Markdown files registered later get default settings and stay
switched off until scheduled.
- New automation: name, project and a starting point (blank example structure or
a template) create .codecraft/jobs/<name>.md and open the editor: settings with
a graphical cron builder (next runs in server time via /jobs/schedule), the
instructions in the Markdown editor and an assistant chat
(POST /jobs/{id}/assistant) whose model edits the file; the editor reloads it
after each turn and warns about conflicts with unsaved edits.
- The schedule tool writes prompt-only files with settings.
Chat list
- Folders and projects keep a user-defined order and can be pinned (drag and
drop, or the group's menu), stored per user via /account/sidebar. A new
message moves the chat to the top of its group, no longer the project itself.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1204 lines
55 KiB
Rust
1204 lines
55 KiB
Rust
//! Telegram bots as a chat interface. Every user can connect their own bot (token from
|
||
//! @BotFather) and pair it with their Telegram account; then plain messages go to the
|
||
//! current session, commands switch sessions, modes and models, and the agent's progress,
|
||
//! answers, images, permission requests and questions come back to Telegram.
|
||
//!
|
||
//! The server polls Telegram (getUpdates), so no public webhook is needed. Only the paired
|
||
//! chat is served; messages from anywhere else are ignored.
|
||
|
||
use crate::app::{App, PermissionReply};
|
||
use crate::store::User;
|
||
use crate::types::{Event, ImageData, Message, Part, Role, Session, new_id, now_ms};
|
||
use anyhow::{Context, Result, anyhow, bail};
|
||
use base64::Engine;
|
||
use serde::{Deserialize, Serialize};
|
||
use serde_json::{Map, Value, json};
|
||
use std::collections::HashMap;
|
||
use std::path::PathBuf;
|
||
use std::sync::{Arc, LazyLock, Mutex};
|
||
use std::time::{Duration, Instant};
|
||
use tokio_util::sync::CancellationToken;
|
||
|
||
/// User setting key.
|
||
const KEY: &str = "telegram";
|
||
/// Telegram's message limit is 4096 characters; leave room for HTML tags.
|
||
const CHUNK: usize = 3500;
|
||
/// Edits of the live progress message at most this often (Telegram rate limits).
|
||
const EDIT_EVERY: Duration = Duration::from_millis(1500);
|
||
/// Files sent to the bot are saved here, relative to the session folder.
|
||
const UPLOADS: &str = ".codecraft/uploads";
|
||
|
||
/// What a user configured (stored in the user's settings).
|
||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||
pub struct Settings {
|
||
#[serde(default)]
|
||
pub token: String,
|
||
/// Bot username from getMe.
|
||
#[serde(default)]
|
||
pub bot: String,
|
||
/// The paired Telegram chat (only this one is served).
|
||
#[serde(default)]
|
||
pub chat_id: Option<i64>,
|
||
/// One-time code for `/start <code>` until the chat is paired.
|
||
#[serde(default)]
|
||
pub pair_code: String,
|
||
/// The session plain messages go to.
|
||
#[serde(default)]
|
||
pub session: Option<String>,
|
||
/// Sessions of the last /sessions list (for `/session <n>`).
|
||
#[serde(default)]
|
||
pub list: Vec<String>,
|
||
/// Reasoning effort for the next messages (/effort).
|
||
#[serde(default)]
|
||
pub effort: Option<String>,
|
||
}
|
||
|
||
fn api_base() -> String {
|
||
std::env::var("CODECRAFT_TELEGRAM_API").unwrap_or_else(|_| "https://api.telegram.org".into()).trim_end_matches('/').to_string()
|
||
}
|
||
|
||
pub fn load(app: &App, user_id: &str) -> Settings {
|
||
app.store.user_setting(user_id, KEY).unwrap_or_default()
|
||
}
|
||
|
||
fn save(app: &App, user_id: &str, s: &Settings) {
|
||
let _ = app.store.set_user_setting(user_id, KEY, s);
|
||
}
|
||
|
||
/// Change the stored settings atomically (the poller and the listener both write).
|
||
fn update(app: &App, user_id: &str, f: impl FnOnce(&mut Settings)) -> Settings {
|
||
static LOCK: Mutex<()> = Mutex::new(());
|
||
let _g = LOCK.lock().unwrap();
|
||
let mut s = load(app, user_id);
|
||
f(&mut s);
|
||
save(app, user_id, &s);
|
||
s
|
||
}
|
||
|
||
// --- Telegram Bot API ---------------------------------------------------------------
|
||
|
||
#[derive(Clone)]
|
||
struct Bot {
|
||
token: String,
|
||
}
|
||
|
||
impl Bot {
|
||
fn url(&self, method: &str) -> String {
|
||
format!("{}/bot{}/{method}", api_base(), self.token)
|
||
}
|
||
|
||
async fn call(&self, method: &str, body: Value) -> Result<Value> {
|
||
let resp = crate::provider::http_client().post(self.url(method)).timeout(Duration::from_secs(60)).json(&body).send().await.context("telegram request failed")?;
|
||
let status = resp.status();
|
||
let v: Value = resp.json().await.unwrap_or_else(|_| json!({}));
|
||
if v["ok"].as_bool() != Some(true) {
|
||
bail!("telegram {method}: HTTP {status}: {}", v["description"].as_str().unwrap_or("error"));
|
||
}
|
||
Ok(v["result"].clone())
|
||
}
|
||
|
||
async fn send(&self, chat: i64, html: &str, keyboard: Option<Value>) -> Result<i64> {
|
||
let mut body = json!({"chat_id": chat, "text": html, "parse_mode": "HTML", "link_preview_options": {"is_disabled": true}});
|
||
if let Some(k) = keyboard {
|
||
body["reply_markup"] = json!({"inline_keyboard": k});
|
||
}
|
||
let r = match self.call("sendMessage", body.clone()).await {
|
||
Ok(r) => r,
|
||
// Broken HTML (should not happen): send as plain text instead of losing it.
|
||
Err(e) if e.to_string().contains("parse") => {
|
||
body.as_object_mut().unwrap().remove("parse_mode");
|
||
body["text"] = json!(strip_tags(html));
|
||
self.call("sendMessage", body).await?
|
||
}
|
||
Err(e) => return Err(e),
|
||
};
|
||
Ok(r["message_id"].as_i64().unwrap_or(0))
|
||
}
|
||
|
||
async fn edit(&self, chat: i64, message: i64, html: &str, keyboard: Option<Value>) -> Result<()> {
|
||
let mut body = json!({"chat_id": chat, "message_id": message, "text": html, "parse_mode": "HTML", "link_preview_options": {"is_disabled": true}});
|
||
body["reply_markup"] = json!({"inline_keyboard": keyboard.unwrap_or_else(|| json!([]))});
|
||
match self.call("editMessageText", body).await {
|
||
Err(e) if e.to_string().contains("not modified") => Ok(()),
|
||
r => r.map(|_| ()),
|
||
}
|
||
}
|
||
|
||
async fn send_photo(&self, chat: i64, image: &ImageData, caption: Option<&str>) -> Result<()> {
|
||
let bytes = base64::engine::general_purpose::STANDARD.decode(&image.data)?;
|
||
let ext = image.media_type.rsplit('/').next().unwrap_or("png").to_string();
|
||
let part = reqwest::multipart::Part::bytes(bytes).file_name(format!("image.{ext}")).mime_str(&image.media_type)?;
|
||
let mut form = reqwest::multipart::Form::new().text("chat_id", chat.to_string()).part("photo", part);
|
||
if let Some(c) = caption {
|
||
form = form.text("caption", c.chars().take(1000).collect::<String>());
|
||
}
|
||
let v: Value = crate::provider::http_client().post(self.url("sendPhoto")).timeout(Duration::from_secs(120)).multipart(form).send().await?.json().await?;
|
||
if v["ok"].as_bool() != Some(true) {
|
||
bail!("telegram sendPhoto: {}", v["description"].as_str().unwrap_or("error"));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Download a file the user sent (photo or document).
|
||
async fn download(&self, file_id: &str) -> Result<(Vec<u8>, String)> {
|
||
let f = self.call("getFile", json!({"file_id": file_id})).await?;
|
||
let path = f["file_path"].as_str().context("telegram file without path")?.to_string();
|
||
let url = format!("{}/file/bot{}/{path}", api_base(), self.token);
|
||
let bytes = crate::provider::http_client().get(url).timeout(Duration::from_secs(120)).send().await?.error_for_status()?.bytes().await?;
|
||
Ok((bytes.to_vec(), path))
|
||
}
|
||
}
|
||
|
||
/// Find the bot token in what the user pasted (also the whole BotFather message).
|
||
pub fn extract_token(input: &str) -> Option<String> {
|
||
let re = regex::Regex::new(r"\b(\d{5,}:[A-Za-z0-9_-]{30,})\b").unwrap();
|
||
re.captures(input).map(|c| c[1].to_string())
|
||
}
|
||
|
||
/// Check a token and return the bot's username.
|
||
pub async fn check_token(token: &str) -> Result<String> {
|
||
let me = Bot { token: token.to_string() }.call("getMe", json!({})).await?;
|
||
me["username"].as_str().map(String::from).context("getMe returned no username")
|
||
}
|
||
|
||
// --- running bots -----------------------------------------------------------------------
|
||
|
||
struct Running {
|
||
cancel: CancellationToken,
|
||
token: String,
|
||
}
|
||
|
||
static RUNNING: LazyLock<Mutex<HashMap<String, Running>>> = LazyLock::new(Default::default);
|
||
/// Last error per user (shown in the settings).
|
||
static ERRORS: LazyLock<Mutex<HashMap<String, String>>> = LazyLock::new(Default::default);
|
||
|
||
pub fn status(user_id: &str) -> (bool, Option<String>) {
|
||
(RUNNING.lock().unwrap().contains_key(user_id), ERRORS.lock().unwrap().get(user_id).cloned())
|
||
}
|
||
|
||
/// Start (or restart) the bot of a user according to their settings.
|
||
pub fn restart(app: &Arc<App>, user_id: &str) {
|
||
stop(user_id);
|
||
let s = load(app, user_id);
|
||
if s.token.is_empty() {
|
||
return;
|
||
}
|
||
let cancel = CancellationToken::new();
|
||
RUNNING.lock().unwrap().insert(user_id.to_string(), Running { cancel: cancel.clone(), token: s.token.clone() });
|
||
ERRORS.lock().unwrap().remove(user_id);
|
||
let (app, uid) = (app.clone(), user_id.to_string());
|
||
tokio::spawn(async move { poll(app, uid, Bot { token: s.token }, cancel).await });
|
||
}
|
||
|
||
pub fn stop(user_id: &str) {
|
||
if let Some(r) = RUNNING.lock().unwrap().remove(user_id) {
|
||
r.cancel.cancel();
|
||
}
|
||
}
|
||
|
||
/// Start all configured bots and the event listener (at server start).
|
||
pub fn spawn(app: Arc<App>) {
|
||
for u in app.store.users().unwrap_or_default() {
|
||
if !u.disabled && !load(&app, &u.id).token.is_empty() {
|
||
restart(&app, &u.id);
|
||
}
|
||
}
|
||
tokio::spawn(listen(app));
|
||
}
|
||
|
||
fn bot_of(user_id: &str) -> Option<Bot> {
|
||
RUNNING.lock().unwrap().get(user_id).map(|r| Bot { token: r.token.clone() })
|
||
}
|
||
|
||
async fn poll(app: Arc<App>, user_id: String, bot: Bot, cancel: CancellationToken) {
|
||
let mut offset: i64 = 0;
|
||
let mut backoff = 2u64;
|
||
loop {
|
||
let req = bot.call("getUpdates", json!({"offset": offset, "timeout": 30, "allowed_updates": ["message", "callback_query"]}));
|
||
let updates = tokio::select! {
|
||
_ = cancel.cancelled() => return,
|
||
r = req => r,
|
||
};
|
||
let updates = match updates {
|
||
Ok(u) => {
|
||
backoff = 2;
|
||
ERRORS.lock().unwrap().remove(&user_id);
|
||
u
|
||
}
|
||
Err(e) => {
|
||
let msg = format!("{e:#}");
|
||
tracing::warn!(user = %user_id, "telegram: {msg}");
|
||
ERRORS.lock().unwrap().insert(user_id.clone(), msg.clone());
|
||
if msg.contains("HTTP 401") || msg.contains("HTTP 404") {
|
||
// Token revoked: stop instead of hammering the API.
|
||
RUNNING.lock().unwrap().remove(&user_id);
|
||
return;
|
||
}
|
||
tokio::select! {
|
||
_ = cancel.cancelled() => return,
|
||
_ = tokio::time::sleep(Duration::from_secs(backoff)) => {}
|
||
}
|
||
backoff = (backoff * 2).min(60);
|
||
continue;
|
||
}
|
||
};
|
||
for u in updates.as_array().into_iter().flatten() {
|
||
offset = offset.max(u["update_id"].as_i64().unwrap_or(0) + 1);
|
||
if let Err(e) = handle_update(&app, &user_id, &bot, u).await {
|
||
tracing::warn!(user = %user_id, "telegram update: {e:#}");
|
||
if let Some(chat) = load(&app, &user_id).chat_id {
|
||
let _ = bot.send(chat, &esc(&format!("{e:#}")), None).await;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// --- incoming messages ---------------------------------------------------------------
|
||
|
||
fn user(app: &App, user_id: &str) -> Result<User> {
|
||
app.store.user(user_id)?.filter(|u| !u.disabled).ok_or_else(|| anyhow!("account disabled"))
|
||
}
|
||
|
||
/// The user reads German (language preference of the web UI).
|
||
pub fn german(app: &App, user_id: &str) -> bool {
|
||
let prefs: Value = app.store.user_setting(user_id, "prefs").unwrap_or_default();
|
||
prefs["lang"].as_str().unwrap_or("en").starts_with("de")
|
||
}
|
||
|
||
/// Pick the German or the English text.
|
||
fn tr<'a>(de: bool, de_text: &'a str, en_text: &'a str) -> &'a str {
|
||
if de { de_text } else { en_text }
|
||
}
|
||
|
||
async fn handle_update(app: &Arc<App>, user_id: &str, bot: &Bot, u: &Value) -> Result<()> {
|
||
let s = load(app, user_id);
|
||
if let Some(q) = u.get("callback_query") {
|
||
let chat = q["message"]["chat"]["id"].as_i64().unwrap_or(0);
|
||
let _ = bot.call("answerCallbackQuery", json!({"callback_query_id": q["id"]})).await;
|
||
if Some(chat) != s.chat_id {
|
||
return Ok(());
|
||
}
|
||
return callback(app, user_id, bot, chat, q["data"].as_str().unwrap_or(""), q["message"]["message_id"].as_i64()).await;
|
||
}
|
||
let Some(m) = u.get("message") else { return Ok(()) };
|
||
let chat = m["chat"]["id"].as_i64().unwrap_or(0);
|
||
let text = m["text"].as_str().or(m["caption"].as_str()).unwrap_or("").trim().to_string();
|
||
let de = german(app, user_id);
|
||
|
||
// Pairing: the first chat that sends the code becomes the only one served.
|
||
if s.chat_id.is_none() {
|
||
let code = text.strip_prefix("/start").map(str::trim).unwrap_or("");
|
||
if !s.pair_code.is_empty() && code == s.pair_code && m["chat"]["type"] == "private" {
|
||
update(app, user_id, |s| {
|
||
s.chat_id = Some(chat);
|
||
s.pair_code.clear();
|
||
});
|
||
let u = user(app, user_id)?;
|
||
bot.send(chat, &esc(&format!("{} {}. /help", tr(de, "Verbunden mit CodeCraft als", "Connected to CodeCraft as"), u.username)), None).await?;
|
||
} else if text.starts_with("/start") {
|
||
bot.send(chat, tr(de, "Bitte den Code aus CodeCraft (Einstellungen → Konto → Telegram) mitsenden: /start CODE", "Please send the code from CodeCraft (Settings → Account → Telegram): /start CODE"), None).await?;
|
||
}
|
||
return Ok(());
|
||
}
|
||
if Some(chat) != s.chat_id {
|
||
return Ok(()); // Not the paired chat.
|
||
}
|
||
let u = user(app, user_id)?;
|
||
|
||
// A pending question of the current session takes a typed answer.
|
||
if !text.starts_with('/') && !text.is_empty() {
|
||
if let Some(sid) = &s.session {
|
||
if let Some((qid, question)) = pending_question(app, sid) {
|
||
let mut answers = Map::new();
|
||
answers.insert(question, json!(text));
|
||
app.answer_question(&qid, Some(answers))?;
|
||
return Ok(());
|
||
}
|
||
}
|
||
}
|
||
|
||
if let Some(cmd) = text.strip_prefix('/') {
|
||
let (name, arg) = cmd.split_once(char::is_whitespace).map(|(a, b)| (a, b.trim())).unwrap_or((cmd, ""));
|
||
let name = name.split('@').next().unwrap_or(name).to_lowercase();
|
||
match name.as_str() {
|
||
"start" | "help" => return bot.send(chat, &help(de), None).await.map(|_| ()),
|
||
"sessions" | "resume" => return list_sessions(app, user_id, bot, chat, de).await,
|
||
"clear" => {
|
||
let cur = current(app, &u, de)?;
|
||
let folder = cur.cwd.clone();
|
||
return new_session(app, &u, bot, chat, &format!("{folder} {}", cur.agent), de).await;
|
||
}
|
||
"permissions" => return set_permissions(app, &u, bot, chat, arg, de).await,
|
||
"prompt" => return chat_instructions(app, &u, bot, chat, arg, de).await,
|
||
"effort" => return set_effort(app, &u, bot, chat, arg, de).await,
|
||
"costs" => return costs(app, &u, bot, chat, de).await,
|
||
"memory" => return memory(app, &u, bot, chat, de).await,
|
||
"jobs" => return jobs(app, &u, bot, chat, de).await,
|
||
"session" => return switch_session(app, &u, bot, chat, arg, de).await,
|
||
"new" => return new_session(app, &u, bot, chat, arg, de).await,
|
||
"mode" => return set_mode(app, &u, bot, chat, arg, de).await,
|
||
"model" => return set_model(app, &u, bot, chat, arg, de).await,
|
||
"stop" => {
|
||
let sid = current(app, &u, de)?;
|
||
crate::runner::abort(app, &sid.id);
|
||
return bot.send(chat, tr(de, "Abgebrochen.", "Stopped."), None).await.map(|_| ());
|
||
}
|
||
"status" => return status_message(app, &u, bot, chat, de).await,
|
||
_ => {} // /compact and slash commands of the session go through as a prompt.
|
||
}
|
||
}
|
||
|
||
// A message for the current session (with photos and files).
|
||
let session = current(app, &u, de)?;
|
||
let mut images = Vec::new();
|
||
let mut notes = Vec::new();
|
||
if let Some(photo) = m["photo"].as_array().and_then(|p| p.last()) {
|
||
let (bytes, _) = bot.download(photo["file_id"].as_str().unwrap_or("")).await?;
|
||
images.push(ImageData { blob: None, size: None, media_type: "image/jpeg".into(), data: base64::engine::general_purpose::STANDARD.encode(bytes) });
|
||
}
|
||
if let Some(doc) = m.get("document") {
|
||
let mime = doc["mime_type"].as_str().unwrap_or("");
|
||
let (bytes, path) = bot.download(doc["file_id"].as_str().unwrap_or("")).await?;
|
||
if mime.starts_with("image/") && bytes.len() < 10 * 1024 * 1024 {
|
||
images.push(ImageData { blob: None, size: None, media_type: mime.into(), data: base64::engine::general_purpose::STANDARD.encode(bytes) });
|
||
} else {
|
||
let name = doc["file_name"].as_str().map(String::from).unwrap_or_else(|| path.rsplit('/').next().unwrap_or("file").to_string());
|
||
let rel = save_upload(&session, &name, &bytes)?;
|
||
notes.push(format!("Attached file: {rel}"));
|
||
}
|
||
}
|
||
let mut prompt = text;
|
||
if !notes.is_empty() {
|
||
prompt = if prompt.is_empty() { notes.join("\n") } else { format!("{prompt}\n\n{}", notes.join("\n")) };
|
||
}
|
||
if prompt.is_empty() && images.is_empty() {
|
||
return Ok(());
|
||
}
|
||
let effort = load(app, user_id).effort;
|
||
let r = crate::api::send_prompt(app, session, prompt, images, effort.as_deref()).map_err(|e| anyhow!("{}", e.1))?;
|
||
if r.get("queued").is_some() {
|
||
bot.send(chat, tr(de, "Eingereiht – wird beim nächsten Schritt übergeben.", "Queued – delivered at the agent's next step."), None).await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Save a file into the session folder (a free name in .codecraft/uploads).
|
||
fn save_upload(session: &Session, name: &str, bytes: &[u8]) -> Result<String> {
|
||
let clean: String = name.chars().map(|c| if c.is_alphanumeric() || ".-_".contains(c) { c } else { '_' }).collect();
|
||
let clean = if clean.trim_matches('.').is_empty() { "file".to_string() } else { clean };
|
||
let dir = PathBuf::from(&session.cwd).join(UPLOADS);
|
||
std::fs::create_dir_all(&dir)?;
|
||
let mut path = dir.join(&clean);
|
||
let mut n = 1;
|
||
while path.exists() {
|
||
n += 1;
|
||
path = dir.join(format!("{n}-{clean}"));
|
||
}
|
||
std::fs::write(&path, bytes)?;
|
||
Ok(format!("{UPLOADS}/{}", path.file_name().unwrap().to_string_lossy()))
|
||
}
|
||
|
||
fn help(de: bool) -> String {
|
||
if de {
|
||
"<b>CodeCraft</b> – Nachrichten gehen an die aktuelle Sitzung (Fotos und Dateien auch).\n\n\
|
||
/sessions – letzte Sitzungen\n/session <Nr|Name> – Sitzung wechseln\n/new [Ordner] [Modus] – neue Sitzung (ohne Ordner: Chat)\n\
|
||
/clear – neue Sitzung im gleichen Ordner\n/mode [Name] – Modus\n/model [Modell] – Modell\n/permissions [Modus] – Freigaben\n/prompt [Text|-] – Anweisungen für diesen Chat\n/effort [Stufe] – Denktiefe\n\
|
||
/stop – Runde abbrechen\n/status – aktuelle Sitzung\n/costs – Kosten und Budgets\n/memory – Gedächtnis\n/jobs – Automationen\n/compact – Unterhaltung komprimieren"
|
||
.into()
|
||
} else {
|
||
"<b>CodeCraft</b> – messages go to the current session (photos and files too).\n\n\
|
||
/sessions – recent sessions\n/session <n|name> – switch session\n/new [folder] [mode] – new session (no folder: chat)\n\
|
||
/clear – new session in the same folder\n/mode [name] – mode\n/model [model] – model\n/permissions [mode] – approvals\n/prompt [text|-] – instructions for this chat\n/effort [level] – reasoning effort\n\
|
||
/stop – stop the turn\n/status – current session\n/costs – costs and budgets\n/memory – memory\n/jobs – automations\n/compact – compact the conversation"
|
||
.into()
|
||
}
|
||
}
|
||
|
||
/// The current session, or the user's most recent one.
|
||
fn current(app: &App, u: &User, de: bool) -> Result<Session> {
|
||
let s = load(app, &u.id);
|
||
if let Some(sess) = s.session.as_deref().and_then(|id| app.store.get_session(id).ok().flatten()).filter(|x| x.owner_id.as_deref() == Some(u.id.as_str())) {
|
||
return Ok(sess);
|
||
}
|
||
let recent = own_sessions(app, &u.id).into_iter().next().ok_or_else(|| anyhow!("{}", tr(de, "Noch keine Sitzung – /new startet eine.", "No session yet – /new starts one.")))?;
|
||
update(app, &u.id, |s| s.session = Some(recent.id.clone()));
|
||
Ok(recent)
|
||
}
|
||
|
||
fn own_sessions(app: &App, user_id: &str) -> Vec<Session> {
|
||
app.store.list_sessions(Some(user_id)).unwrap_or_default().into_iter().filter(|s| s.owner_id.as_deref() == Some(user_id) && !s.archived).collect()
|
||
}
|
||
|
||
fn label(s: &Session) -> String {
|
||
let title = if s.title.is_empty() { "(untitled)".to_string() } else { s.title.clone() };
|
||
let folder = s.cwd.rsplit('/').next().unwrap_or("");
|
||
format!("{} · {folder}", title.chars().take(40).collect::<String>())
|
||
}
|
||
|
||
async fn list_sessions(app: &App, user_id: &str, bot: &Bot, chat: i64, de: bool) -> Result<()> {
|
||
let list: Vec<Session> = own_sessions(app, user_id).into_iter().take(10).collect();
|
||
if list.is_empty() {
|
||
bot.send(chat, tr(de, "Noch keine Sitzungen – /new startet eine.", "No sessions yet – /new starts one."), None).await?;
|
||
return Ok(());
|
||
}
|
||
let cur = load(app, user_id).session;
|
||
let keyboard: Vec<Value> = list
|
||
.iter()
|
||
.enumerate()
|
||
.map(|(i, s)| {
|
||
let mark = if cur.as_deref() == Some(s.id.as_str()) { "• " } else { "" };
|
||
json!([{"text": format!("{mark}{} {}", i + 1, label(s)), "callback_data": format!("s:{}", s.id)}])
|
||
})
|
||
.collect();
|
||
update(app, user_id, |st| st.list = list.iter().map(|s| s.id.clone()).collect());
|
||
bot.send(chat, tr(de, "Sitzung wählen:", "Pick a session:"), Some(json!(keyboard))).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn announce(app: &App, bot: &Bot, chat: i64, s: &Session, de: bool) -> Result<()> {
|
||
let mut text = format!("<b>{}</b>\n{} · {} · {}", esc(&label(s)), esc(&s.cwd), esc(&s.agent), esc(&s.model));
|
||
// The last answer, to pick up the thread.
|
||
if let Some(last) = app.store.messages(&s.id).unwrap_or_default().iter().rev().find(|m| m.role == Role::Assistant).map(text_of).filter(|t| !t.is_empty()) {
|
||
let short: String = last.chars().take(600).collect();
|
||
text.push_str(&format!("\n\n{}{}", md_to_html(&short), if last.chars().count() > 600 { " …" } else { "" }));
|
||
}
|
||
let _ = de;
|
||
bot.send(chat, &text, None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn switch_session(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
if arg.is_empty() {
|
||
return list_sessions(app, &u.id, bot, chat, de).await;
|
||
}
|
||
let st = load(app, &u.id);
|
||
let own = own_sessions(app, &u.id);
|
||
let found = match arg.parse::<usize>() {
|
||
Ok(n) if n >= 1 => st.list.get(n - 1).and_then(|id| own.iter().find(|s| &s.id == id)).cloned(),
|
||
_ => {
|
||
let q = arg.to_lowercase();
|
||
own.iter().find(|s| s.title.to_lowercase().contains(&q) || s.cwd.to_lowercase().ends_with(&q)).cloned()
|
||
}
|
||
};
|
||
let Some(s) = found else {
|
||
bot.send(chat, tr(de, "Keine passende Sitzung – /sessions zeigt die letzten.", "No matching session – /sessions lists the recent ones."), None).await?;
|
||
return Ok(());
|
||
};
|
||
update(app, &u.id, |st| st.session = Some(s.id.clone()));
|
||
announce(app, bot, chat, &s, de).await
|
||
}
|
||
|
||
async fn new_session(app: &Arc<App>, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let roots = app.user_roots(Some(&u.id));
|
||
let root = roots.first().cloned().ok_or_else(|| anyhow!("no folder available"))?;
|
||
let words: Vec<&str> = arg.split_whitespace().collect();
|
||
let agents = crate::prompts::agents_for(app, &root, &root, Some(&u.id));
|
||
let is_mode = |w: &str| agents.iter().any(|a| a.name == w && a.mode == "primary");
|
||
let (folder, mode) = match words.as_slice() {
|
||
[] => (None, "chat".to_string()),
|
||
[m] if is_mode(m) => (None, m.to_string()),
|
||
[f] => (Some(*f), "build".to_string()),
|
||
[f, m, ..] => (Some(*f), m.to_string()),
|
||
};
|
||
if !is_mode(&mode) {
|
||
bot.send(chat, &esc(&format!("{}: {mode}", tr(de, "Unbekannter Modus", "Unknown mode"))), None).await?;
|
||
return Ok(());
|
||
}
|
||
let cwd = match folder {
|
||
None => root.clone(),
|
||
Some(f) => {
|
||
let p = crate::tools::normalize(&if f.starts_with('/') { PathBuf::from(f) } else { root.join(f) });
|
||
if !roots.iter().any(|r| p.starts_with(r)) || !p.is_dir() {
|
||
bot.send(chat, &esc(&format!("{}: {f}", tr(de, "Ordner nicht gefunden", "Folder not found"))), None).await?;
|
||
return Ok(());
|
||
}
|
||
p
|
||
}
|
||
};
|
||
let cfg = app.config();
|
||
let model = Some(cfg.server.default_model.clone()).filter(|m| !m.is_empty()).or_else(|| app.registry().first_model()).ok_or_else(|| anyhow!("no model configured"))?;
|
||
let now = now_ms();
|
||
let s = app.save_session(Session {
|
||
id: new_id(),
|
||
title: String::new(),
|
||
cwd: cwd.display().to_string(),
|
||
model,
|
||
agent: mode,
|
||
permission_mode: "auto".into(),
|
||
parent_id: None,
|
||
owner_id: Some(u.id.clone()),
|
||
created_at: now,
|
||
updated_at: now,
|
||
todos: Value::Null,
|
||
last_usage: None,
|
||
clean_room: None,
|
||
phase: None,
|
||
worktree: None,
|
||
pinned: false,
|
||
archived: false,
|
||
folder: None,
|
||
instructions: String::new(),
|
||
// Started from Telegram: keeps reporting there after switching to another chat.
|
||
telegram_quiet: false,
|
||
})?;
|
||
update(app, &u.id, |st| st.session = Some(s.id.clone()));
|
||
bot.send(chat, &format!("{} <b>{}</b> · {} · {}", tr(de, "Neue Sitzung", "New session"), esc(&s.cwd), esc(&s.agent), esc(&s.model)), None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn set_mode(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let s = current(app, u, de)?;
|
||
let cwd = PathBuf::from(&s.cwd);
|
||
let root = app.root_for(Some(&u.id), &cwd);
|
||
let modes: Vec<String> = crate::prompts::agents_for(app, &root, &cwd, Some(&u.id)).into_iter().filter(|a| a.mode == "primary").map(|a| a.name).collect();
|
||
if arg.is_empty() {
|
||
let keyboard: Vec<Value> = modes
|
||
.chunks(3)
|
||
.map(|row| json!(row.iter().map(|m| json!({"text": if *m == s.agent { format!("• {m}") } else { m.clone() }, "callback_data": format!("m:{m}")})).collect::<Vec<_>>()))
|
||
.collect();
|
||
bot.send(chat, tr(de, "Modus wählen:", "Pick a mode:"), Some(json!(keyboard))).await?;
|
||
return Ok(());
|
||
}
|
||
if !modes.iter().any(|m| m == arg) {
|
||
bot.send(chat, &esc(&format!("{}: {arg}", tr(de, "Unbekannter Modus", "Unknown mode"))), None).await?;
|
||
return Ok(());
|
||
}
|
||
app.update_session(&s.id, |x| x.agent = arg.to_string())?;
|
||
bot.send(chat, &format!("{}: <b>{}</b>", tr(de, "Modus", "Mode"), esc(arg)), None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// All models as "provider/model".
|
||
fn models(app: &App) -> Vec<String> {
|
||
let mut out = Vec::new();
|
||
for p in app.registry().models_json().as_array().into_iter().flatten() {
|
||
for m in p["models"].as_array().into_iter().flatten() {
|
||
out.push(format!("{}/{}", p["provider"].as_str().unwrap_or(""), m["id"].as_str().unwrap_or("")));
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
async fn set_model(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let s = current(app, u, de)?;
|
||
let all = models(app);
|
||
if arg.is_empty() {
|
||
// Callback data is limited to 64 bytes: buttons carry the index.
|
||
let keyboard: Vec<Value> = all
|
||
.iter()
|
||
.enumerate()
|
||
.take(40)
|
||
.map(|(i, m)| json!([{"text": if *m == s.model { format!("• {m}") } else { m.clone() }, "callback_data": format!("M:{i}")}]))
|
||
.collect();
|
||
bot.send(chat, tr(de, "Modell wählen:", "Pick a model:"), Some(json!(keyboard))).await?;
|
||
return Ok(());
|
||
}
|
||
let pick = all.iter().find(|m| m.as_str() == arg).or_else(|| all.iter().find(|m| m.to_lowercase().contains(&arg.to_lowercase()))).cloned();
|
||
let Some(m) = pick else {
|
||
bot.send(chat, &esc(&format!("{}: {arg}", tr(de, "Unbekanntes Modell", "Unknown model"))), None).await?;
|
||
return Ok(());
|
||
};
|
||
app.update_session(&s.id, |x| x.model = m.clone())?;
|
||
bot.send(chat, &format!("{}: <b>{}</b>", tr(de, "Modell", "Model"), esc(&m)), None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn status_message(app: &App, u: &User, bot: &Bot, chat: i64, de: bool) -> Result<()> {
|
||
let s = current(app, u, de)?;
|
||
let busy = app.is_busy(&s.id);
|
||
let ctx = s.last_usage.as_ref().map(|x| x.context_tokens()).unwrap_or(0);
|
||
let text = format!(
|
||
"<b>{}</b>\n{}: {}\n{}: {}\n{}: {}\n{}: {}\n{}",
|
||
esc(&label(&s)),
|
||
tr(de, "Ordner", "Folder"),
|
||
esc(&s.cwd),
|
||
tr(de, "Modus", "Mode"),
|
||
esc(&s.agent),
|
||
tr(de, "Modell", "Model"),
|
||
esc(&s.model),
|
||
tr(de, "Kontext", "Context"),
|
||
ctx,
|
||
if busy { tr(de, "arbeitet …", "working …") } else { tr(de, "bereit", "idle") },
|
||
);
|
||
bot.send(chat, &text, None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Show, set ("/prompt text") or clear ("/prompt -") the instructions of the current chat.
|
||
async fn chat_instructions(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let s = current(app, u, de)?;
|
||
if arg.is_empty() {
|
||
let text = if s.instructions.is_empty() {
|
||
tr(de, "Keine Anweisungen für diesen Chat. Setzen mit /prompt <Text>.", "No instructions for this chat. Set them with /prompt <text>.").to_string()
|
||
} else {
|
||
format!("<b>{}</b>\n{}\n\n<i>{}</i>", tr(de, "Anweisungen für diesen Chat", "Instructions for this chat"), esc(&s.instructions), tr(de, "Löschen mit /prompt -", "Clear with /prompt -"))
|
||
};
|
||
bot.send(chat, &text, None).await?;
|
||
return Ok(());
|
||
}
|
||
let new = if arg == "-" { String::new() } else { arg.to_string() };
|
||
app.update_session(&s.id, |x| x.instructions = new.clone())?;
|
||
let text = if new.is_empty() { tr(de, "Anweisungen gelöscht.", "Instructions cleared.") } else { tr(de, "Anweisungen gesetzt – sie gelten ab der nächsten Nachricht.", "Instructions set – they apply from the next message.") };
|
||
bot.send(chat, text, None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
const PERMISSION_MODES: &[&str] = &["auto", "default", "accept_edits", "bypass"];
|
||
const EFFORTS: &[&str] = &["low", "medium", "high", "xhigh", "max"];
|
||
|
||
/// Buttons in rows of three; the current value is marked.
|
||
fn choice_keyboard(values: &[&str], current: &str, prefix: &str) -> Value {
|
||
json!(values.chunks(3).map(|row| row.iter().map(|v| json!({"text": if *v == current { format!("• {v}") } else { v.to_string() }, "callback_data": format!("{prefix}:{v}")})).collect::<Vec<_>>()).collect::<Vec<_>>())
|
||
}
|
||
|
||
async fn set_permissions(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let s = current(app, u, de)?;
|
||
if arg.is_empty() {
|
||
bot.send(chat, tr(de, "Freigabemodus wählen:", "Pick the approval mode:"), Some(choice_keyboard(PERMISSION_MODES, &s.permission_mode, "P"))).await?;
|
||
return Ok(());
|
||
}
|
||
if !PERMISSION_MODES.contains(&arg) {
|
||
bot.send(chat, &esc(&format!("{}: {arg}", tr(de, "Unbekannter Modus", "Unknown mode"))), None).await?;
|
||
return Ok(());
|
||
}
|
||
app.update_session(&s.id, |x| x.permission_mode = arg.to_string())?;
|
||
bot.send(chat, &format!("{}: <b>{}</b>", tr(de, "Freigaben", "Approvals"), esc(arg)), None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn set_effort(app: &App, u: &User, bot: &Bot, chat: i64, arg: &str, de: bool) -> Result<()> {
|
||
let cur = load(app, &u.id).effort.unwrap_or_else(|| "auto".into());
|
||
if arg.is_empty() {
|
||
let mut values = vec!["auto"];
|
||
values.extend_from_slice(EFFORTS);
|
||
bot.send(chat, tr(de, "Denktiefe für die nächsten Nachrichten:", "Reasoning effort for the next messages:"), Some(choice_keyboard(&values, &cur, "E"))).await?;
|
||
return Ok(());
|
||
}
|
||
if arg != "auto" && !EFFORTS.contains(&arg) {
|
||
bot.send(chat, &esc(&format!("{}: {arg}", tr(de, "Unbekannte Stufe", "Unknown level"))), None).await?;
|
||
return Ok(());
|
||
}
|
||
update(app, &u.id, |s| s.effort = (arg != "auto").then(|| arg.to_string()));
|
||
bot.send(chat, &format!("{}: <b>{}</b>", tr(de, "Denktiefe", "Effort"), esc(arg)), None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn costs(app: &App, u: &User, bot: &Bot, chat: i64, de: bool) -> Result<()> {
|
||
let cfg = app.config();
|
||
let now = chrono::Local::now();
|
||
let offset = now.offset().local_minus_utc() as i64 / 60;
|
||
let day = now.date_naive().and_hms_opt(0, 0, 0).and_then(|d| d.and_local_timezone(chrono::Local).single()).map(|d| d.timestamp_millis()).unwrap_or(0);
|
||
let total = |from: i64| app.store.global_stats(from, now.timestamp_millis() + 1, offset, Some(&u.id)).ok().map(|v| (v["total"]["cost"].as_f64().unwrap_or(0.0), v["total"]["requests"].as_i64().unwrap_or(0))).unwrap_or((0.0, 0));
|
||
let (today, req_today) = total(day);
|
||
let (month, req_month) = total(crate::budget::month_start());
|
||
let cur = &cfg.server.currency;
|
||
let mut text = format!(
|
||
"<b>{}</b>
|
||
{}: {today:.2} {cur} ({req_today})
|
||
{}: {month:.2} {cur} ({req_month})",
|
||
tr(de, "Kosten", "Costs"),
|
||
tr(de, "Heute", "Today"),
|
||
tr(de, "Monat", "Month"),
|
||
);
|
||
if u.is_admin() {
|
||
let meters = crate::budget::meters(app).unwrap_or_default();
|
||
if !meters.is_empty() {
|
||
text.push_str(&format!("\n\n<b>{}</b>", tr(de, "Budgets", "Budgets")));
|
||
for m in meters {
|
||
let name = match m.kind.as_str() {
|
||
"total" => tr(de, "Gesamt", "Total").to_string(),
|
||
"provider" => m.provider.clone(),
|
||
"images" => tr(de, "Bilder", "Images").to_string(),
|
||
w => format!("{} · {w}", m.provider),
|
||
};
|
||
text.push_str(&format!("\n{}: {:.0} % ({:.2} / {:.2})", esc(&name), m.ratio * 100.0, m.used, m.limit));
|
||
}
|
||
}
|
||
}
|
||
bot.send(chat, &text, None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn memory(app: &App, u: &User, bot: &Bot, chat: i64, de: bool) -> Result<()> {
|
||
let personal: String = app.store.user_setting(&u.id, "memory").unwrap_or_default();
|
||
let project = current(app, u, de).ok().and_then(|s| std::fs::read_to_string(PathBuf::from(&s.cwd).join(".codecraft/memory.md")).ok()).unwrap_or_default();
|
||
let cut = |t: &str| -> String { if t.trim().is_empty() { tr(de, "(leer)", "(empty)").to_string() } else { t.chars().take(1500).collect() } };
|
||
let text = format!("<b>{}</b>\n{}\n\n<b>{}</b>\n{}", tr(de, "Persönlich", "Personal"), md_to_html(&cut(&personal)), tr(de, "Projekt", "Project"), md_to_html(&cut(&project)));
|
||
bot.send(chat, &text, None).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn jobs(app: &App, u: &User, bot: &Bot, chat: i64, de: bool) -> Result<()> {
|
||
let list = app.store.jobs(Some(&u.id))?;
|
||
if list.is_empty() {
|
||
bot.send(chat, tr(de, "Keine Automationen.", "No automations."), None).await?;
|
||
return Ok(());
|
||
}
|
||
let keyboard: Vec<Value> = list
|
||
.iter()
|
||
.take(20)
|
||
.map(|j| {
|
||
let d = crate::jobs::describe(j);
|
||
let name = d["spec"]["name"].as_str().unwrap_or(&j.path).to_string();
|
||
let state = if j.enabled { d["spec"]["schedule"].as_str().unwrap_or("").to_string() } else { tr(de, "aus", "off").to_string() };
|
||
json!([{"text": format!("▶ {} ({state})", name.chars().take(40).collect::<String>()), "callback_data": format!("j:{}", j.id)}])
|
||
})
|
||
.collect();
|
||
bot.send(chat, tr(de, "Automationen – antippen startet jetzt:", "Automations – tap to run now:"), Some(json!(keyboard))).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn callback(app: &Arc<App>, user_id: &str, bot: &Bot, chat: i64, data: &str, message: Option<i64>) -> Result<()> {
|
||
let u = user(app, user_id)?;
|
||
let de = german(app, user_id);
|
||
let (kind, rest) = data.split_once(':').unwrap_or((data, ""));
|
||
match kind {
|
||
"s" => {
|
||
let Some(s) = own_sessions(app, user_id).into_iter().find(|s| s.id == rest) else { return Ok(()) };
|
||
update(app, user_id, |st| st.session = Some(s.id.clone()));
|
||
announce(app, bot, chat, &s, de).await
|
||
}
|
||
"m" => set_mode(app, &u, bot, chat, rest, de).await,
|
||
"P" => set_permissions(app, &u, bot, chat, rest, de).await,
|
||
"E" => set_effort(app, &u, bot, chat, rest, de).await,
|
||
"j" => {
|
||
let Some(job) = app.store.job(rest)?.filter(|j| j.owner_id == u.id) else { return Ok(()) };
|
||
let sid = crate::jobs::run(app, &job, "telegram").await?;
|
||
update(app, user_id, |st| st.session = Some(sid.clone()));
|
||
bot.send(chat, tr(de, "Automation gestartet – die Sitzung ist jetzt aktuell.", "Automation started – its session is now current."), None).await?;
|
||
Ok(())
|
||
}
|
||
"M" => {
|
||
let Some(m) = rest.parse::<usize>().ok().and_then(|i| models(app).get(i).cloned()) else { return Ok(()) };
|
||
set_model(app, &u, bot, chat, &m, de).await
|
||
}
|
||
"p" => {
|
||
// p:<permission id>:<o|a|r>
|
||
let (id, what) = rest.rsplit_once(':').unwrap_or((rest, "r"));
|
||
let reply = match what {
|
||
"o" => PermissionReply::Once,
|
||
"a" => PermissionReply::Always,
|
||
_ => PermissionReply::Reject,
|
||
};
|
||
let done = app.reply_permission(id, reply, None, None).is_ok();
|
||
if let Some(mid) = message {
|
||
let text = match (done, what) {
|
||
(false, _) => tr(de, "(bereits erledigt)", "(already handled)"),
|
||
(_, "o") => tr(de, "Erlaubt (einmal)", "Allowed (once)"),
|
||
(_, "a") => tr(de, "Erlaubt (für die Sitzung)", "Allowed (for the session)"),
|
||
_ => tr(de, "Abgelehnt", "Rejected"),
|
||
};
|
||
let _ = bot.edit(chat, mid, &format!("{}\n\n<i>{text}</i>", esc(&PENDING_TEXT.lock().unwrap().remove(id).unwrap_or_default())), None).await;
|
||
}
|
||
Ok(())
|
||
}
|
||
"q" => {
|
||
// q:<question id>:<question index>:<option index>
|
||
let mut it = rest.rsplitn(3, ':');
|
||
let (oi, qi, id) = (it.next().unwrap_or(""), it.next().unwrap_or(""), it.next().unwrap_or(""));
|
||
let Some(req) = QUESTIONS.lock().unwrap().get(id).cloned() else { return Ok(()) };
|
||
let (Ok(qi), Ok(oi)) = (qi.parse::<usize>(), oi.parse::<usize>()) else { return Ok(()) };
|
||
let Some(q) = req.get(qi) else { return Ok(()) };
|
||
let Some(opt) = q.options.get(oi) else { return Ok(()) };
|
||
let all_done = {
|
||
let mut answers = ANSWERS.lock().unwrap();
|
||
let a = answers.entry(id.to_string()).or_default();
|
||
a.insert(q.question.clone(), json!(opt.label));
|
||
a.len() >= req.len()
|
||
};
|
||
if let Some(mid) = message {
|
||
let _ = bot.edit(chat, mid, &format!("{}\n\n→ <b>{}</b>", esc(&q.question), esc(&opt.label)), None).await;
|
||
}
|
||
if all_done {
|
||
let answers = ANSWERS.lock().unwrap().remove(id).unwrap_or_default();
|
||
QUESTIONS.lock().unwrap().remove(id);
|
||
let _ = app.answer_question(id, Some(answers));
|
||
}
|
||
Ok(())
|
||
}
|
||
_ => Ok(()),
|
||
}
|
||
}
|
||
|
||
/// Text of pending permission messages (to keep it when the buttons go).
|
||
static PENDING_TEXT: LazyLock<Mutex<HashMap<String, String>>> = LazyLock::new(Default::default);
|
||
/// Questions sent to Telegram, and the answers collected so far.
|
||
static QUESTIONS: LazyLock<Mutex<HashMap<String, Vec<crate::types::Question>>>> = LazyLock::new(Default::default);
|
||
static ANSWERS: LazyLock<Mutex<HashMap<String, Map<String, Value>>>> = LazyLock::new(Default::default);
|
||
/// Question id per session (a typed message answers the first open question).
|
||
static SESSION_QUESTION: LazyLock<Mutex<HashMap<String, String>>> = LazyLock::new(Default::default);
|
||
|
||
fn pending_question(_app: &App, session_id: &str) -> Option<(String, String)> {
|
||
let id = SESSION_QUESTION.lock().unwrap().get(session_id).cloned()?;
|
||
let qs = QUESTIONS.lock().unwrap().get(&id).cloned()?;
|
||
let answered = ANSWERS.lock().unwrap().get(&id).cloned().unwrap_or_default();
|
||
let q = qs.iter().find(|q| !answered.contains_key(&q.question))?;
|
||
if qs.len() > 1 {
|
||
return None; // Several questions: answer with the buttons.
|
||
}
|
||
QUESTIONS.lock().unwrap().remove(&id);
|
||
SESSION_QUESTION.lock().unwrap().remove(session_id);
|
||
Some((id, q.question.clone()))
|
||
}
|
||
|
||
// --- outgoing: progress, answers, images, approvals -----------------------------------
|
||
|
||
/// Live progress of a turn in the current session.
|
||
#[derive(Default)]
|
||
struct Progress {
|
||
message: Option<i64>,
|
||
steps: Vec<String>,
|
||
text: String,
|
||
answer: Vec<String>,
|
||
last_edit: Option<Instant>,
|
||
}
|
||
|
||
async fn listen(app: Arc<App>) {
|
||
let mut rx = app.subscribe();
|
||
let mut progress: HashMap<String, Progress> = HashMap::new();
|
||
loop {
|
||
let ev = match rx.recv().await {
|
||
Ok(ev) => ev,
|
||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
|
||
Err(_) => return,
|
||
};
|
||
let Some(session_id) = ev.session_id().map(String::from) else { continue };
|
||
let Ok(Some(session)) = app.store.get_session(&session_id) else { continue };
|
||
// Subagents and other users' sessions report through the owner's top-level session.
|
||
let Some(owner) = session.owner_id.clone() else { continue };
|
||
let Some(bot) = bot_of(&owner) else { continue };
|
||
let st = load(&app, &owner);
|
||
let Some(chat) = st.chat_id else { continue };
|
||
let de = german(&app, &owner);
|
||
let is_current = st.session.as_deref() == Some(session_id.as_str());
|
||
// Chats with "not to Telegram": their approvals and questions stay in the web UI
|
||
// (subagents follow their top-level chat; the chat open in Telegram always reports).
|
||
if !is_current && matches!(ev, Event::PermissionRequest { .. } | Event::QuestionRequest { .. }) {
|
||
let root = app.root_session_id(&session_id);
|
||
let quiet = if st.session.as_deref() == Some(root.as_str()) {
|
||
false
|
||
} else if root == session_id { session.telegram_quiet } else { app.store.get_session(&root).ok().flatten().is_some_and(|r| r.telegram_quiet) };
|
||
if quiet {
|
||
continue;
|
||
}
|
||
}
|
||
if let Err(e) = on_event(&app, &bot, chat, de, &session, is_current, &ev, &mut progress).await {
|
||
tracing::debug!("telegram send: {e:#}");
|
||
}
|
||
}
|
||
}
|
||
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn on_event(app: &App, bot: &Bot, chat: i64, de: bool, session: &Session, is_current: bool, ev: &Event, progress: &mut HashMap<String, Progress>) -> Result<()> {
|
||
let sid = session.id.clone();
|
||
match ev {
|
||
// Approvals and questions of every own session: the agent waits for them.
|
||
Event::PermissionRequest { id, title, tool, .. } => {
|
||
let mut text = format!("{}\n<b>{}</b>: {}", tr(de, "Freigabe nötig", "Approval needed"), esc(tool), esc(title));
|
||
if !is_current {
|
||
text = format!("[{}]\n{text}", esc(&label(session)));
|
||
}
|
||
PENDING_TEXT.lock().unwrap().insert(id.clone(), strip_tags(&text));
|
||
let keyboard = json!([[
|
||
{"text": tr(de, "Einmal", "Once"), "callback_data": format!("p:{id}:o")},
|
||
{"text": tr(de, "Immer", "Always"), "callback_data": format!("p:{id}:a")},
|
||
{"text": tr(de, "Ablehnen", "Reject"), "callback_data": format!("p:{id}:r")},
|
||
]]);
|
||
bot.send(chat, &text, Some(keyboard)).await?;
|
||
}
|
||
Event::QuestionRequest { id, questions, .. } => {
|
||
QUESTIONS.lock().unwrap().insert(id.clone(), questions.clone());
|
||
SESSION_QUESTION.lock().unwrap().insert(sid.clone(), id.clone());
|
||
for (qi, q) in questions.iter().enumerate() {
|
||
let mut text = format!("<b>{}</b>", esc(&q.question));
|
||
if !is_current {
|
||
text = format!("[{}]\n{text}", esc(&label(session)));
|
||
}
|
||
if questions.len() == 1 {
|
||
text.push_str(&format!("\n<i>{}</i>", tr(de, "Antippen oder eine eigene Antwort schreiben.", "Tap an option or type your own answer.")));
|
||
}
|
||
let keyboard: Vec<Value> = q.options.iter().enumerate().map(|(oi, o)| json!([{"text": o.label.chars().take(60).collect::<String>(), "callback_data": format!("q:{id}:{qi}:{oi}")}])).collect();
|
||
bot.send(chat, &text, Some(json!(keyboard))).await?;
|
||
}
|
||
}
|
||
Event::QuestionResolved { id, .. } => {
|
||
QUESTIONS.lock().unwrap().remove(id);
|
||
ANSWERS.lock().unwrap().remove(id);
|
||
}
|
||
Event::PermissionResolved { id, .. } => {
|
||
PENDING_TEXT.lock().unwrap().remove(id);
|
||
}
|
||
_ if !is_current => {}
|
||
// The current session: live progress and the answer.
|
||
Event::SessionStatus { busy: true, .. } => {
|
||
let mid = bot.send(chat, &format!("<i>{}</i>", tr(de, "arbeitet …", "working …")), None).await?;
|
||
progress.insert(sid, Progress { message: Some(mid), ..Default::default() });
|
||
}
|
||
Event::ToolStarted { name, input, .. } => {
|
||
let p = progress.entry(sid).or_default();
|
||
p.text.clear();
|
||
p.steps.push(step_line(name, input));
|
||
refresh(bot, chat, p, de, false).await;
|
||
}
|
||
Event::TextDelta { text, .. } => {
|
||
let p = progress.entry(sid).or_default();
|
||
p.text.push_str(text);
|
||
refresh(bot, chat, p, de, false).await;
|
||
}
|
||
Event::MessageAdded { message, .. } => {
|
||
let p = progress.entry(sid).or_default();
|
||
match message.role {
|
||
Role::Assistant => {
|
||
let t = text_of(message);
|
||
if !t.is_empty() {
|
||
p.answer.push(t);
|
||
}
|
||
p.text.clear();
|
||
}
|
||
// Images from tools (generated images, screenshots) go out right away.
|
||
Role::User => {
|
||
for part in &message.parts {
|
||
if let Part::ToolResult { images, name, .. } = part {
|
||
for img in images {
|
||
let _ = bot.send_photo(chat, &app.store.resolve_image(img), Some(name)).await;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Event::Error { message, .. } => {
|
||
bot.send(chat, &format!("<b>{}</b>: {}", tr(de, "Fehler", "Error"), esc(message)), None).await?;
|
||
}
|
||
Event::SessionStatus { busy: false, .. } => {
|
||
let Some(p) = progress.remove(&sid) else { return Ok(()) };
|
||
let answer = p.answer.join("\n\n");
|
||
let chunks = split_markdown(if answer.trim().is_empty() { tr(de, "(keine Textantwort)", "(no text answer)") } else { &answer });
|
||
let mut rest = chunks.iter();
|
||
// The first chunk replaces the progress message; the steps stay as a short summary.
|
||
if let (Some(mid), Some(first)) = (p.message, rest.next()) {
|
||
let mut html = String::new();
|
||
if !p.steps.is_empty() {
|
||
html.push_str(&format!("<i>{}</i>\n\n", esc(&summary(&p.steps, de))));
|
||
}
|
||
html.push_str(&md_to_html(first));
|
||
if bot.edit(chat, mid, &html, None).await.is_err() {
|
||
bot.send(chat, &md_to_html(first), None).await?;
|
||
}
|
||
}
|
||
for c in rest {
|
||
bot.send(chat, &md_to_html(c), None).await?;
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
async fn refresh(bot: &Bot, chat: i64, p: &mut Progress, de: bool, force: bool) {
|
||
let Some(mid) = p.message else { return };
|
||
if !force && p.last_edit.is_some_and(|t| t.elapsed() < EDIT_EVERY) {
|
||
return;
|
||
}
|
||
p.last_edit = Some(Instant::now());
|
||
let mut html = format!("<i>{}</i>", tr(de, "arbeitet …", "working …"));
|
||
let steps: Vec<&String> = p.steps.iter().rev().take(6).collect();
|
||
for s in steps.into_iter().rev() {
|
||
html.push_str(&format!("\n• {}", esc(s)));
|
||
}
|
||
let live: String = p.text.chars().rev().take(1200).collect::<Vec<_>>().into_iter().rev().collect();
|
||
if !live.trim().is_empty() {
|
||
html.push_str(&format!("\n\n{}", esc(&live)));
|
||
}
|
||
let _ = bot.edit(chat, mid, &html, None).await;
|
||
}
|
||
|
||
/// One line for a tool call, e.g. "bash: cargo test" or "read src/main.rs".
|
||
fn step_line(name: &str, input: &Value) -> String {
|
||
let arg = ["description", "command", "path", "file_path", "pattern", "url", "query", "prompt"].iter().find_map(|k| input[*k].as_str()).unwrap_or("");
|
||
let arg: String = arg.lines().next().unwrap_or("").chars().take(80).collect();
|
||
if arg.is_empty() { name.to_string() } else { format!("{name}: {arg}") }
|
||
}
|
||
|
||
fn summary(steps: &[String], de: bool) -> String {
|
||
let n = steps.len();
|
||
let last: Vec<&str> = steps.iter().rev().take(3).map(|s| s.as_str()).collect();
|
||
format!("{} {n} {} – {}", tr(de, "Fertig nach", "Done after"), tr(de, if n == 1 { "Schritt" } else { "Schritten" }, if n == 1 { "step" } else { "steps" }), last.into_iter().rev().collect::<Vec<_>>().join(", "))
|
||
}
|
||
|
||
fn text_of(m: &Message) -> String {
|
||
m.parts.iter().filter_map(|p| if let Part::Text { text } = p { Some(text.trim()) } else { None }).filter(|t| !t.is_empty()).collect::<Vec<_>>().join("\n\n")
|
||
}
|
||
|
||
// --- formatting ---------------------------------------------------------------------------
|
||
|
||
pub fn esc(s: &str) -> String {
|
||
s.replace('&', "&").replace('<', "<").replace('>', ">")
|
||
}
|
||
|
||
fn strip_tags(html: &str) -> String {
|
||
let re = regex::Regex::new(r"<[^>]+>").unwrap();
|
||
re.replace_all(html, "").replace("<", "<").replace(">", ">").replace("&", "&")
|
||
}
|
||
|
||
/// Split Markdown into chunks below Telegram's limit, at paragraph or line boundaries,
|
||
/// closing and reopening code fences that cross a boundary.
|
||
fn split_markdown(md: &str) -> Vec<String> {
|
||
let mut out = Vec::new();
|
||
let mut cur = String::new();
|
||
let mut fence: Option<String> = None;
|
||
for line in md.lines() {
|
||
if cur.len() + line.len() + 1 > CHUNK && !cur.is_empty() {
|
||
if fence.is_some() {
|
||
cur.push_str("```\n");
|
||
}
|
||
out.push(std::mem::take(&mut cur));
|
||
if let Some(f) = &fence {
|
||
cur.push_str(f);
|
||
cur.push('\n');
|
||
}
|
||
}
|
||
let t = line.trim_start();
|
||
if t.starts_with("```") {
|
||
fence = if fence.is_some() { None } else { Some(t.to_string()) };
|
||
}
|
||
// A single huge line: hard split.
|
||
let mut l = line;
|
||
while l.len() > CHUNK {
|
||
let cut = (0..=CHUNK).rev().find(|i| l.is_char_boundary(*i)).unwrap_or(0);
|
||
out.push(format!("{cur}{}", &l[..cut]));
|
||
cur.clear();
|
||
l = &l[cut..];
|
||
}
|
||
cur.push_str(l);
|
||
cur.push('\n');
|
||
}
|
||
if !cur.trim().is_empty() {
|
||
out.push(cur);
|
||
}
|
||
if out.is_empty() {
|
||
out.push(String::new());
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Markdown (as models write it) to Telegram's HTML subset: code blocks, inline code,
|
||
/// bold, italics, links and headings; everything else is escaped text.
|
||
pub fn md_to_html(md: &str) -> String {
|
||
let mut out = String::new();
|
||
let mut in_code = false;
|
||
for line in md.lines() {
|
||
let t = line.trim_start();
|
||
if t.starts_with("```") {
|
||
if in_code {
|
||
out.push_str("</code></pre>\n");
|
||
} else {
|
||
let lang = t.trim_start_matches('`').trim();
|
||
if lang.is_empty() {
|
||
out.push_str("<pre><code>");
|
||
} else {
|
||
out.push_str(&format!("<pre><code class=\"language-{}\">", esc(lang)));
|
||
}
|
||
}
|
||
in_code = !in_code;
|
||
continue;
|
||
}
|
||
if in_code {
|
||
out.push_str(&esc(line));
|
||
out.push('\n');
|
||
continue;
|
||
}
|
||
let heading = t.strip_prefix("### ").or(t.strip_prefix("## ")).or(t.strip_prefix("# "));
|
||
let line_html = inline(heading.unwrap_or(line));
|
||
if heading.is_some() {
|
||
out.push_str(&format!("<b>{line_html}</b>\n"));
|
||
} else {
|
||
let bullet = t.strip_prefix("- ").or(t.strip_prefix("* "));
|
||
match bullet {
|
||
Some(rest) => out.push_str(&format!("• {}\n", inline(rest))),
|
||
None => {
|
||
out.push_str(&line_html);
|
||
out.push('\n');
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if in_code {
|
||
out.push_str("</code></pre>");
|
||
}
|
||
out.trim_end().to_string()
|
||
}
|
||
|
||
fn inline(s: &str) -> String {
|
||
// Split off inline code first: nothing inside is formatted.
|
||
let mut out = String::new();
|
||
for (i, seg) in s.split('`').enumerate() {
|
||
if i % 2 == 1 {
|
||
out.push_str(&format!("<code>{}</code>", esc(seg)));
|
||
continue;
|
||
}
|
||
let mut x = esc(seg);
|
||
let link = regex::Regex::new(r"\[([^\]]+)\]\((https?://[^)\s]+)\)").unwrap();
|
||
x = link.replace_all(&x, r#"<a href="$2">$1</a>"#).into_owned();
|
||
let bold = regex::Regex::new(r"\*\*([^*]+)\*\*").unwrap();
|
||
x = bold.replace_all(&x, "<b>$1</b>").into_owned();
|
||
let italic = regex::Regex::new(r"(^|[^*\w])\*([^*\s][^*]*)\*").unwrap();
|
||
x = italic.replace_all(&x, "$1<i>$2</i>").into_owned();
|
||
out.push_str(&x);
|
||
}
|
||
// An unpaired backtick leaves an odd segment count: that last segment was code.
|
||
out
|
||
}
|
||
|
||
/// A fresh six-digit pairing code.
|
||
pub fn new_pair_code() -> String {
|
||
use rand::RngExt;
|
||
format!("{:06}", rand::rng().random_range(0..1_000_000))
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn markdown_to_telegram_html() {
|
||
let md = "# Title\nSome **bold** and *it* with `a<b>` and [link](https://x.org).\n- item\n```rust\nfn main() { 1 < 2 }\n```";
|
||
let html = md_to_html(md);
|
||
assert!(html.contains("<b>Title</b>"));
|
||
assert!(html.contains("<b>bold</b>"));
|
||
assert!(html.contains("<i>it</i>"));
|
||
assert!(html.contains("<code>a<b></code>"));
|
||
assert!(html.contains(r#"<a href="https://x.org">link</a>"#));
|
||
assert!(html.contains("• item"));
|
||
assert!(html.contains("<pre><code class=\"language-rust\">fn main() { 1 < 2 }\n</code></pre>"));
|
||
}
|
||
|
||
#[test]
|
||
fn splits_long_answers_and_keeps_code_fences_closed() {
|
||
let code: String = (0..400).map(|i| format!("line {i} of a long listing\n")).collect();
|
||
let md = format!("intro\n```\n{code}```\nend");
|
||
let chunks = split_markdown(&md);
|
||
assert!(chunks.len() > 1);
|
||
for c in &chunks {
|
||
assert!(c.len() <= CHUNK + 10, "chunk too long: {}", c.len());
|
||
assert_eq!(c.matches("```").count() % 2, 0, "unbalanced fence in chunk");
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn finds_the_token_in_pasted_text() {
|
||
let msg = "Done! Congratulations on your new bot.\nUse this token to access the HTTP API:\n7123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw4\nKeep your token secure";
|
||
assert_eq!(extract_token(msg).as_deref(), Some("7123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw4"));
|
||
assert_eq!(extract_token(" 7123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw4 ").as_deref(), Some("7123456789:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw4"));
|
||
assert_eq!(extract_token("my-password-123"), None);
|
||
}
|
||
|
||
#[test]
|
||
fn step_lines() {
|
||
assert_eq!(step_line("bash", &json!({"command": "cargo test\nmore"})), "bash: cargo test");
|
||
assert_eq!(step_line("read", &json!({"path": "src/main.rs"})), "read: src/main.rs");
|
||
assert_eq!(step_line("todo_write", &json!({})), "todo_write");
|
||
}
|
||
}
|