use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::sync::Arc;
use std::time::{Duration, Instant};
use anyhow::{bail, Context, Result};
use clap::Parser;
use rusqlite::{params, Connection};
use teloxide::net::Download;
use teloxide::prelude::*;
use teloxide::types::{
BotCommand, ChatAction, ChatId, InlineKeyboardButton, InlineKeyboardMarkup, InputFile,
MessageId, ParseMode, ReactionType, ReplyParameters,
};
use tokio::sync::Mutex;
#[derive(Parser, Clone)]
#[command(name = "team-bot", version, about = "Telegram interface for teamctl")]
struct Cli {
#[arg(long, env = "TEAMCTL_MAILBOX")]
mailbox: PathBuf,
#[arg(long, env = "TEAMCTL_TELEGRAM_TOKEN")]
token: String,
#[arg(long, env = "TEAMCTL_TELEGRAM_CHATS")]
authorized_chat_ids: Option<String>,
#[arg(long, env = "TEAMCTL_MANAGER")]
manager: Option<String>,
#[arg(long, env = "TEAMCTL_MANAGER_DISPLAY_NAME")]
manager_display_name: Option<String>,
#[arg(long, env = "TEAMCTL_TMUX_PREFIX", default_value = "t-")]
tmux_prefix: String,
#[arg(long, env = "TEAMCTL_STT_PROVIDER")]
stt_provider: Option<String>,
#[arg(long, env = "TEAMCTL_STT_API_KEY")]
stt_api_key: Option<String>,
#[arg(long, env = "TEAMCTL_STT_MODEL")]
stt_model: Option<String>,
#[arg(long, env = "TEAMCTL_STT_LANGUAGE")]
stt_language: Option<String>,
#[arg(
long,
env = "TEAMCTL_TELEGRAM_NO_RICH",
num_args = 0..=1,
default_value_t = false,
default_missing_value = "true",
value_parser = clap::builder::BoolishValueParser::new(),
)]
no_rich_messages: bool,
}
struct State {
conn: Mutex<Connection>,
allow: Vec<i64>,
manager: Option<String>,
manager_display_name: Option<String>,
tmux_prefix: String,
media_root: PathBuf,
stt: Option<SttRuntime>,
typing: Mutex<HashMap<ChatId, Instant>>,
rich: bool,
}
const TYPING_WINDOW_CEILING: Duration = Duration::from_secs(10);
const TYPING_REFRESH_INTERVAL: Duration = Duration::from_secs(4);
struct SttRuntime {
provider: String,
api_key: String,
model: String,
language: Option<String>,
http: reqwest::Client,
}
const VOICE_INBOX_PREFIX: &str = "🎙 (transcribed voice, may have misspellings):";
impl State {
fn manager_project(&self) -> Option<&str> {
self.manager
.as_deref()
.and_then(|m| m.split_once(':').map(|(p, _)| p))
}
}
impl State {
fn is_authorized(&self, chat: i64) -> bool {
self.allow.is_empty() || self.allow.contains(&chat)
}
}
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_env("TEAM_BOT_LOG")
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
.init();
let cli = Cli::parse();
let bot = Bot::new(&cli.token);
let conn = open_mailbox(&cli.mailbox)?;
let allow: Vec<i64> = cli
.authorized_chat_ids
.as_deref()
.unwrap_or("")
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.filter_map(|s| s.parse().ok())
.collect();
let media_root = cli
.mailbox
.parent()
.map(|p| p.join("inbound-media"))
.unwrap_or_else(|| PathBuf::from("inbound-media"));
let stt = match (cli.stt_provider, cli.stt_api_key, cli.stt_model) {
(Some(provider), Some(api_key), Some(model)) => Some(SttRuntime {
provider,
api_key,
model,
language: cli.stt_language,
http: reqwest::Client::new(),
}),
_ => None,
};
let state = Arc::new(State {
conn: Mutex::new(conn),
allow,
manager: cli.manager,
manager_display_name: cli.manager_display_name,
tmux_prefix: cli.tmux_prefix,
media_root,
stt,
typing: Mutex::new(HashMap::new()),
rich: !cli.no_rich_messages,
});
let runtime = if let Some(mgr) = state.manager.as_deref() {
let c = state.conn.lock().await;
agent_runtime(&c, mgr)
} else {
None
};
let commands = commands_for_runtime(runtime.as_deref());
if !commands.is_empty() {
if let Err(e) = bot.set_my_commands(commands).await {
tracing::warn!(
"set_my_commands failed (operator gets no autocomplete; \
slash-passthrough still works manually): {e}"
);
}
}
{
let bot = bot.clone();
let state = state.clone();
tokio::spawn(async move { outbound_loop(bot, state).await });
}
{
let bot = bot.clone();
let state = state.clone();
tokio::spawn(async move { typing_refresh_loop(bot, state).await });
}
let bot_inbound = bot.clone();
let handler = dptree::entry()
.branch(Update::filter_message().endpoint({
let state = state.clone();
move |bot: Bot, msg: Message| {
let state = state.clone();
async move { handle_message(bot, msg, state).await }
}
}))
.branch(Update::filter_callback_query().endpoint({
let state = state.clone();
move |bot: Bot, q: CallbackQuery| {
let state = state.clone();
async move { handle_callback(bot, q, state).await }
}
}));
Dispatcher::builder(bot_inbound, handler)
.enable_ctrlc_handler()
.build()
.dispatch()
.await;
Ok(())
}
fn peel_readnow(body: &str) -> (&str, Option<&'static str>) {
if let Some(rest) = body.strip_prefix("/readnow ") {
(rest, Some("immediate"))
} else {
(body, None)
}
}
const INBOUND_QUOTE_MAX_CHARS: usize = 500;
struct ReplyContext {
sender: String,
mailbox_id: Option<i64>,
quoted: Option<String>,
}
fn truncate_quote(text: &str) -> String {
let mut out: String = text.chars().take(INBOUND_QUOTE_MAX_CHARS).collect();
if text.chars().count() > INBOUND_QUOTE_MAX_CHARS {
out.push('…');
}
out
}
fn lookup_replied_row(conn: &Connection, telegram_msg_id: i64) -> Option<(i64, String)> {
match conn.query_row(
"SELECT id, sender FROM messages WHERE telegram_msg_id = ?1 ORDER BY id DESC LIMIT 1",
params![telegram_msg_id],
|r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?)),
) {
Ok(row) => Some(row),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => {
tracing::warn!(
telegram_msg_id,
error = %e,
"inbound reply: mailbox lookup failed; quoting without a row id"
);
None
}
}
}
fn sender_label_for(is_bot: Option<bool>) -> &'static str {
match is_bot {
Some(true) => "agent",
Some(false) => "user:telegram",
None => "unknown",
}
}
fn telegram_sender_label(replied: &Message) -> String {
sender_label_for(replied.from.as_ref().map(|u| u.is_bot)).to_string()
}
async fn resolve_reply_context(msg: &Message, state: &State) -> Option<ReplyContext> {
let replied = msg.reply_to_message();
let quote = msg.quote();
if replied.is_none() && quote.is_none() {
return None;
}
let (sender, mailbox_id) = match replied {
Some(r) => {
let row = {
let c = state.conn.lock().await;
lookup_replied_row(&c, r.id.0 as i64)
};
match row {
Some((id, sender)) => (sender, Some(id)),
None => (telegram_sender_label(r), None),
}
}
None => ("unknown".to_string(), None),
};
let quoted = quote
.map(|q| q.text.as_str())
.or_else(|| replied.and_then(|r| r.text().or_else(|| r.caption())))
.filter(|s| !s.is_empty())
.map(truncate_quote);
Some(ReplyContext {
sender,
mailbox_id,
quoted,
})
}
fn build_inbound_body(new_text: &str, ctx: Option<&ReplyContext>) -> String {
let Some(ctx) = ctx else {
return new_text.to_string();
};
let mut out = String::new();
match ctx.mailbox_id {
Some(id) => out.push_str(&format!("> [reply to {} msg {}]\n", ctx.sender, id)),
None => out.push_str(&format!("> [reply to {}]\n", ctx.sender)),
}
if let Some(quoted) = &ctx.quoted {
for line in quoted.lines() {
out.push_str("> ");
out.push_str(line);
out.push('\n');
}
}
out.push('\n');
out.push_str(new_text);
out
}
fn open_mailbox(path: &std::path::Path) -> Result<Connection> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).ok();
}
let conn = Connection::open(path).context("open mailbox")?;
conn.busy_timeout(Duration::from_secs(5))?;
conn.pragma_update(None, "journal_mode", "WAL")?;
team_core::mailbox::ensure(&conn)?;
Ok(conn)
}
async fn handle_message(bot: Bot, msg: Message, state: Arc<State>) -> ResponseResult<()> {
let chat_id = msg.chat.id.0;
let trimmed = msg.text().map(str::trim).unwrap_or("");
if !state.allow.contains(&chat_id) && trimmed == "/start" {
bot.send_message(
msg.chat.id,
format!(
"This chat isn't authorized yet.\n\n\
Your chat id: {chat_id}\n\n\
Add it to .env next to your team-compose.yaml:\n\
TEAMCTL_TELEGRAM_CHATS={chat_id}\n\n\
Then restart team-bot."
),
)
.await?;
return Ok(());
}
if !state.is_authorized(chat_id) {
return Ok(());
}
if msg.photo().is_some() || msg.document().is_some() {
return handle_inbound_media(&bot, &msg, &state).await;
}
if msg.voice().is_some() && state.stt.is_some() && state.manager.is_some() {
return handle_voice(&bot, &msg, &state).await;
}
if msg.voice().is_some() && state.stt.is_none() && state.manager.is_some() {
return handle_voice_stt_missing(&bot, &msg).await;
}
let inbound_msg_id: i64 = msg.id.0 as i64;
let reply_ctx = resolve_reply_context(&msg, &state).await;
if let Some(rest) = trimmed.strip_prefix("/dm ") {
if let Some((target, body)) = rest.split_once(' ') {
if let Some((project, _)) = target.split_once(':') {
let (body, delivery_mode) = peel_readnow(body);
let body = build_inbound_body(body, reply_ctx.as_ref());
let c = state.conn.lock().await;
let _ = c.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, telegram_msg_id, delivery_mode)
VALUES (?1, 'user:telegram', ?2, ?3, strftime('%s','now'), ?4, ?5)",
params![project, target, body, inbound_msg_id, delivery_mode],
);
drop(c);
bot.send_message(msg.chat.id, format!("→ {target}")).await?;
}
}
} else if !trimmed.is_empty() && !trimmed.starts_with('/') && state.manager.is_some() {
let target = state.manager.as_deref().unwrap();
if let Some((project, _)) = target.split_once(':') {
let body = build_inbound_body(trimmed, reply_ctx.as_ref());
let c = state.conn.lock().await;
let _ = c.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES (?1, 'user:telegram', ?2, ?3, strftime('%s','now'), ?4)",
params![project, target, body, inbound_msg_id],
);
drop(c);
bot.send_message(msg.chat.id, format!("→ {target}")).await?;
}
} else if trimmed.starts_with("/readnow ") && state.manager.is_some() {
let target = state.manager.as_deref().unwrap();
let (body, delivery_mode) = peel_readnow(trimmed);
if !body.is_empty() {
if let Some((project, _)) = target.split_once(':') {
let body = build_inbound_body(body, reply_ctx.as_ref());
let c = state.conn.lock().await;
let _ = c.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, telegram_msg_id, delivery_mode)
VALUES (?1, 'user:telegram', ?2, ?3, strftime('%s','now'), ?4, ?5)",
params![project, target, body, inbound_msg_id, delivery_mode],
);
drop(c);
bot.send_message(msg.chat.id, format!("→ {target} (now)"))
.await?;
}
}
} else if trimmed == "/pending" {
let c = state.conn.lock().await;
let rows: Vec<(i64, String, String, String)> = {
let mut stmt = c
.prepare(
"SELECT id, agent_id, action, summary FROM approvals WHERE status='pending' ORDER BY id",
)
.unwrap();
stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
.unwrap()
.flatten()
.collect()
};
drop(c);
if rows.is_empty() {
bot.send_message(msg.chat.id, "No pending approvals.")
.await?;
} else {
let mut out = String::from("Pending approvals:\n");
for (id, agent, action, summary) in rows {
out.push_str(&format!(
"#{id} {} · {}: {}\n",
html_escape_str(&agent),
html_escape_str(&action),
render_html(&summary),
));
}
bot.send_message(msg.chat.id, out)
.parse_mode(ParseMode::Html)
.await?;
}
} else if trimmed == "/start" || trimmed == "/help" {
let name = state
.manager
.as_deref()
.map(|mgr| state.manager_display_name.as_deref().unwrap_or(mgr));
let body = start_help_body(trimmed == "/help", name);
bot.send_message(msg.chat.id, body).await?;
} else if trimmed.starts_with('/') && state.manager.is_some() {
let manager = state.manager.as_deref().unwrap();
let runtime_opt = {
let c = state.conn.lock().await;
agent_runtime(&c, manager)
};
let Some(runtime) = runtime_opt else {
bot.send_message(
msg.chat.id,
format!("unknown manager `{manager}` — slash-passthrough aborted"),
)
.await?;
return Ok(());
};
match slash_outcome(manager, &runtime, &state.tmux_prefix) {
SlashOutcome::Passthrough { session } => match tmux_send_keys(&session, trimmed) {
Ok(()) => {
bot.send_message(msg.chat.id, format!("→ {manager}"))
.await?;
}
Err(err) => {
bot.send_message(msg.chat.id, format!("tmux error: {err}"))
.await?;
}
},
SlashOutcome::Reject { reason } => {
bot.send_message(msg.chat.id, reason).await?;
}
}
}
Ok(())
}
fn approval_outcome_line(approved: bool, approver_first_name: &str) -> String {
let verb = if approved {
"✅ Approved"
} else {
"❌ Rejected"
};
format!("{verb} by {approver_first_name}")
}
fn decision_outcome_line(chosen_label: &str, approver_first_name: &str) -> String {
format!("✅ {chosen_label} — chosen by {approver_first_name}")
}
fn cancel_outcome_line(approver_first_name: &str) -> String {
format!("🚫 Cancelled by {approver_first_name}")
}
#[derive(Debug, PartialEq, Eq)]
enum CbAction {
Approve,
Deny,
Opt(usize),
Cancel,
}
fn parse_callback(data: &str) -> Option<(i64, CbAction)> {
let mut it = data.split(':');
let verb = it.next()?;
let id: i64 = it.next()?.parse().ok()?;
let action = match verb {
"approve" => CbAction::Approve,
"deny" => CbAction::Deny,
"cancel" => CbAction::Cancel,
"opt" => CbAction::Opt(it.next()?.parse().ok()?),
_ => return None,
};
if it.next().is_some() {
return None;
}
Some((id, action))
}
fn decode_options(options_json: Option<&str>) -> Vec<(String, String)> {
let Some(raw) = options_json else {
return Vec::new();
};
serde_json::from_str::<Vec<serde_json::Value>>(raw)
.map(|arr| {
arr.into_iter()
.filter_map(|o| {
Some((
o.get("label")?.as_str()?.to_string(),
o.get("value")?.as_str()?.to_string(),
))
})
.collect()
})
.unwrap_or_default()
}
fn approval_keyboard(id: i64, options: &[(String, String)]) -> InlineKeyboardMarkup {
if options.is_empty() {
return InlineKeyboardMarkup::new(vec![vec![
InlineKeyboardButton::callback("Approve", format!("approve:{id}")),
InlineKeyboardButton::callback("Deny", format!("deny:{id}")),
]]);
}
let mut rows: Vec<Vec<InlineKeyboardButton>> = options
.iter()
.enumerate()
.map(|(i, (label, _))| {
vec![InlineKeyboardButton::callback(
label.clone(),
format!("opt:{id}:{i}"),
)]
})
.collect();
rows.push(vec![InlineKeyboardButton::callback(
"Cancel",
format!("cancel:{id}"),
)]);
InlineKeyboardMarkup::new(rows)
}
async fn handle_callback(bot: Bot, q: CallbackQuery, state: Arc<State>) -> ResponseResult<()> {
let chat_id = q.message.as_ref().map(|m| m.chat().id.0).unwrap_or(0);
if !state.is_authorized(chat_id) {
return Ok(());
}
let Some(data) = q.data.clone() else {
return Ok(());
};
let Some((id, action)) = parse_callback(&data) else {
return Ok(());
};
let (status, value, outcome, toast): (&str, Option<String>, String, String) = match action {
CbAction::Approve => (
"approved",
None,
approval_outcome_line(true, &q.from.first_name),
format!("✅ #{id}"),
),
CbAction::Deny => (
"denied",
None,
approval_outcome_line(false, &q.from.first_name),
format!("❌ #{id}"),
),
CbAction::Cancel => (
"denied",
None,
cancel_outcome_line(&q.from.first_name),
format!("🚫 #{id}"),
),
CbAction::Opt(idx) => {
let opts = {
let c = state.conn.lock().await;
c.query_row(
"SELECT options_json FROM approvals WHERE id=?1",
params![id],
|r| r.get::<_, Option<String>>(0),
)
.ok()
.flatten()
};
let decoded = decode_options(opts.as_deref());
let Some((label, val)) = decoded.get(idx).cloned() else {
bot.answer_callback_query(q.id)
.text(format!("#{id} option unavailable"))
.await?;
return Ok(());
};
(
"decided",
Some(val),
decision_outcome_line(&label, &q.from.first_name),
format!("✅ #{id}"),
)
}
};
let decided_now = {
let c = state.conn.lock().await;
let n = c
.execute(
"UPDATE approvals SET status=?1, decided_at=strftime('%s','now'), decided_by='user:telegram', decision_value=?2
WHERE id=?3 AND status='pending'",
params![status, value, id],
)
.map(|n| n > 0)
.unwrap_or(false);
if n {
let _ = c.execute(
"UPDATE approvals SET delivered_at=strftime('%s','now')
WHERE id=?1 AND delivered_at IS NULL",
params![id],
);
}
n
};
if !decided_now {
bot.answer_callback_query(q.id)
.text(format!("#{id} already resolved"))
.await?;
return Ok(());
}
if let Some(msg) = q.message.as_ref() {
let chat = msg.chat().id;
let mid = msg.id();
let original = msg.regular_message().and_then(|m| m.text()).unwrap_or("");
let new_text = if original.is_empty() {
outcome.clone()
} else {
format!("{original}\n\n{outcome}")
};
let _ = bot.edit_message_text(chat, mid, new_text).await;
let _ = bot
.edit_message_reply_markup(chat, mid)
.reply_markup(InlineKeyboardMarkup::new(Vec::<Vec<_>>::new()))
.await;
}
bot.answer_callback_query(q.id).text(toast).await?;
Ok(())
}
async fn outbound_loop(bot: Bot, state: Arc<State>) {
let Some(&primary) = state.allow.first() else {
tracing::warn!("no authorized_chat_ids — outbound disabled");
return;
};
let chat = ChatId(primary);
let mut last_approval_id: i64 = current_max(&state, "approvals").await;
let mut last_msg_id: i64 = current_max(&state, "messages").await;
loop {
tokio::time::sleep(Duration::from_millis(500)).await;
type ApprovalRow = (i64, String, String, String, Option<String>);
let approvals: Vec<ApprovalRow> = {
let c = state.conn.lock().await;
let rows: Vec<ApprovalRow> = match state.manager_project() {
Some(project) => {
let mut stmt = c
.prepare(
"SELECT id, agent_id, action, summary, options_json FROM approvals
WHERE status='pending' AND id > ?1 AND project_id = ?2
ORDER BY id",
)
.unwrap();
stmt.query_map(params![last_approval_id, project], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
})
.unwrap()
.flatten()
.collect()
}
None => {
let mut stmt = c
.prepare(
"SELECT id, agent_id, action, summary, options_json FROM approvals
WHERE status='pending' AND id > ?1 ORDER BY id",
)
.unwrap();
stmt.query_map(params![last_approval_id], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
})
.unwrap()
.flatten()
.collect()
}
};
rows
};
for (id, agent, action, summary, options_json) in approvals {
last_approval_id = last_approval_id.max(id);
let route_ok = {
let c = state.conn.lock().await;
should_route(state.manager.as_deref(), &agent, &c)
};
if !route_ok {
continue;
}
let kb = approval_keyboard(id, &decode_options(options_json.as_deref()));
let text = format!(
"🔐 #{id} {}\naction: {}\n{}",
html_escape_str(&agent),
html_escape_str(&action),
render_html(&summary),
);
let send_ok = bot
.send_message(chat, text)
.parse_mode(ParseMode::Html)
.reply_markup(kb)
.await
.is_ok();
if send_ok {
let c = state.conn.lock().await;
let _ = c.execute(
"UPDATE approvals SET delivered_at=strftime('%s','now')
WHERE id=?1 AND delivered_at IS NULL",
params![id],
);
}
}
let forwardable: Vec<MailboxRow> = {
let c = state.conn.lock().await;
let rows: Vec<MailboxRow> = match state.manager_project() {
Some(project) => {
let mut stmt = c
.prepare(
"SELECT m.id, m.sender, m.text, m.kind, m.structured_payload,
m.telegram_msg_id
FROM messages m
WHERE m.id > ?1
AND m.recipient = 'user:telegram'
AND m.acked_at IS NULL
AND m.project_id = ?2
ORDER BY m.id",
)
.unwrap();
stmt.query_map(params![last_msg_id, project], MailboxRow::from_row)
.unwrap()
.flatten()
.collect()
}
None => {
let mut stmt = c
.prepare(
"SELECT m.id, m.sender, m.text, m.kind, m.structured_payload,
m.telegram_msg_id
FROM messages m
WHERE m.id > ?1
AND m.recipient = 'user:telegram'
AND m.acked_at IS NULL
ORDER BY m.id",
)
.unwrap();
stmt.query_map(params![last_msg_id], MailboxRow::from_row)
.unwrap()
.flatten()
.collect()
}
};
rows
};
for row in forwardable {
last_msg_id = last_msg_id.max(row.id);
let route_ok = {
let c = state.conn.lock().await;
should_route(state.manager.as_deref(), &row.sender, &c)
};
if !route_ok {
continue;
}
let kind = classify_kind(row.kind.as_deref());
match kind {
DispatchKind::Text | DispatchKind::Image | DispatchKind::File => {
let mut map = state.typing.lock().await;
clear_typing_window(&mut map, chat);
}
DispatchKind::Typing => {
let mut map = state.typing.lock().await;
extend_typing_window(&mut map, chat, Instant::now(), TYPING_WINDOW_CEILING);
drop(map);
if let Err(e) = bot.send_chat_action(chat, ChatAction::Typing).await {
tracing::warn!("send_chat_action failed for row {}: {e}", row.id);
}
}
_ => {}
}
if !matches!(kind, DispatchKind::Typing) {
forward_row(&bot, chat, &row, state.rich).await;
}
let c = state.conn.lock().await;
let _ = c.execute(
"UPDATE messages SET acked_at = strftime('%s','now') WHERE id = ?1",
params![row.id],
);
}
}
}
async fn typing_refresh_loop(bot: Bot, state: Arc<State>) {
loop {
tokio::time::sleep(TYPING_REFRESH_INTERVAL).await;
let active: Vec<ChatId> = {
let mut map = state.typing.lock().await;
refresh_typing_windows(&mut map, Instant::now())
};
for chat in active {
if let Err(e) = bot.send_chat_action(chat, ChatAction::Typing).await {
tracing::warn!("typing refresh send_chat_action failed for {chat}: {e}");
}
}
}
}
#[derive(Debug, Clone)]
struct MailboxRow {
id: i64,
sender: String,
text: String,
kind: Option<String>,
payload: Option<String>,
telegram_msg_id: Option<i64>,
}
impl MailboxRow {
fn from_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
Ok(Self {
id: r.get(0)?,
sender: r.get(1)?,
text: r.get(2)?,
kind: r.get(3)?,
payload: r.get(4)?,
telegram_msg_id: r.get(5)?,
})
}
}
fn reply_parameters_for(telegram_msg_id: Option<i64>) -> Option<ReplyParameters> {
telegram_msg_id.map(|id| ReplyParameters::new(MessageId(id as i32)))
}
struct MediaPayload {
source: String,
value: String,
caption: Option<String>,
}
fn parse_payload(payload: &str) -> Option<MediaPayload> {
let v: serde_json::Value = serde_json::from_str(payload).ok()?;
let source = v.get("source")?.as_str()?.to_string();
let value = v.get("value")?.as_str()?.to_string();
let caption = v
.get("caption")
.and_then(|c| c.as_str())
.map(|s| s.to_string());
Some(MediaPayload {
source,
value,
caption,
})
}
fn input_file_from(payload: &MediaPayload) -> Option<InputFile> {
match payload.source.as_str() {
"path" => Some(InputFile::file(&payload.value)),
"url" => Some(InputFile::url(payload.value.parse().ok()?)),
_ => None,
}
}
#[derive(Debug, PartialEq, Eq)]
enum DispatchKind {
Text,
Image,
File,
Reaction,
Typing,
UnknownFallback,
}
fn classify_kind(kind: Option<&str>) -> DispatchKind {
match kind {
None | Some("text") | Some("") => DispatchKind::Text,
Some("image") => DispatchKind::Image,
Some("file") => DispatchKind::File,
Some("reaction") => DispatchKind::Reaction,
Some("typing") => DispatchKind::Typing,
_ => DispatchKind::UnknownFallback,
}
}
fn extend_typing_window(
map: &mut HashMap<ChatId, Instant>,
chat: ChatId,
now: Instant,
ceiling: Duration,
) -> Instant {
let deadline = now + ceiling;
map.insert(chat, deadline);
deadline
}
fn clear_typing_window(map: &mut HashMap<ChatId, Instant>, chat: ChatId) -> bool {
map.remove(&chat).is_some()
}
fn refresh_typing_windows(map: &mut HashMap<ChatId, Instant>, now: Instant) -> Vec<ChatId> {
map.retain(|_, deadline| *deadline > now);
map.keys().copied().collect()
}
struct ReactionPayload {
telegram_msg_id: i64,
emoji: String,
}
fn parse_reaction_payload(payload: &str) -> Option<ReactionPayload> {
let v: serde_json::Value = serde_json::from_str(payload).ok()?;
let telegram_msg_id = v.get("telegram_msg_id")?.as_i64()?;
let emoji = v.get("emoji")?.as_str()?.to_string();
Some(ReactionPayload {
telegram_msg_id,
emoji,
})
}
fn rich_message_body(chat_id: i64, markdown: &str) -> serde_json::Value {
serde_json::json!({
"chat_id": chat_id,
"rich_message": { "markdown": markdown },
})
}
async fn forward_rich_row(
bot: &Bot,
chat: ChatId,
markdown: &str,
row_id: i64,
) -> Result<(), String> {
let url = format!("{}bot{}/sendRichMessage", bot.api_url(), bot.token());
let body =
serde_json::to_string(&rich_message_body(chat.0, markdown)).map_err(|e| e.to_string())?;
let resp = bot
.client()
.post(url)
.header("Content-Type", "application/json")
.body(body)
.send()
.await
.map_err(|e| e.without_url().to_string())?;
let status = resp.status();
if status.is_success() {
Ok(())
} else {
let text = resp.text().await.unwrap_or_default();
Err(format!(
"row {row_id}: sendRichMessage HTTP {status}: {text}"
))
}
}
async fn send_text_html(
bot: &Bot,
chat: ChatId,
body: String,
reply: Option<ReplyParameters>,
row_id: i64,
) {
let mut req = bot.send_message(chat, body).parse_mode(ParseMode::Html);
if let Some(rp) = reply {
req = req.reply_parameters(rp);
}
if let Some(e) = req.await.err() {
tracing::warn!("send_message (text) failed for mailbox row {}: {e}", row_id);
}
}
fn use_rich_path(rich_enabled: bool, has_reply_target: bool) -> bool {
rich_enabled && !has_reply_target
}
async fn forward_row(bot: &Bot, chat: ChatId, row: &MailboxRow, rich: bool) {
let kind = classify_kind(row.kind.as_deref());
let attribution = format!("\n\n— replied by {}", html_escape_str(&row.sender));
let reply = reply_parameters_for(row.telegram_msg_id);
match kind {
DispatchKind::Text => {
let html_body = format!("{}{attribution}", render_html(&row.text));
if use_rich_path(rich, reply.is_some()) {
let md_attribution =
format!("\n\n— replied by {}", markdown_escape_str(&row.sender));
let md_body = format!("{}{md_attribution}", row.text);
match forward_rich_row(bot, chat, &md_body, row.id).await {
Ok(()) => {}
Err(e) => {
tracing::warn!("{e}; falling back to HTML send");
send_text_html(bot, chat, html_body, reply.clone(), row.id).await;
}
}
} else {
send_text_html(bot, chat, html_body, reply.clone(), row.id).await;
}
}
DispatchKind::Image | DispatchKind::File => {
let Some(payload) = row.payload.as_deref().and_then(parse_payload) else {
if let Some(e) = bot
.send_message(
chat,
format!(
"{} (media payload unparseable){attribution}",
render_html(&row.text)
),
)
.parse_mode(ParseMode::Html)
.await
.err()
{
tracing::warn!(
"send_message (media-unparseable fallback) failed for mailbox row {}: {e}",
row.id
);
}
return;
};
let Some(input) = input_file_from(&payload) else {
if let Some(e) = bot
.send_message(
chat,
format!(
"{} (unsupported media source <code>{}</code>){attribution}",
render_html(&row.text),
html_escape_str(&payload.source)
),
)
.parse_mode(ParseMode::Html)
.await
.err()
{
tracing::warn!(
"send_message (unsupported-source fallback) failed for mailbox row {}: {e}",
row.id
);
}
return;
};
let caption_text = payload
.caption
.as_deref()
.map(|c| format!("{}{attribution}", render_html(c)))
.unwrap_or_else(|| attribution.trim_start().to_string());
let result = match kind {
DispatchKind::Image => {
let mut req = bot
.send_photo(chat, input)
.caption(caption_text)
.parse_mode(ParseMode::Html);
if let Some(rp) = reply.clone() {
req = req.reply_parameters(rp);
}
req.await.err()
}
DispatchKind::File => {
let mut req = bot
.send_document(chat, input)
.caption(caption_text)
.parse_mode(ParseMode::Html);
if let Some(rp) = reply.clone() {
req = req.reply_parameters(rp);
}
req.await.err()
}
_ => unreachable!(),
};
if let Some(e) = result {
tracing::warn!(
"send_{} failed for mailbox row {}: {e}",
if kind == DispatchKind::Image {
"photo"
} else {
"document"
},
row.id
);
}
}
DispatchKind::Reaction => {
let Some(reaction) = row.payload.as_deref().and_then(parse_reaction_payload) else {
tracing::warn!(
"reaction payload unparseable for mailbox row {} (skipping)",
row.id
);
return;
};
let result = bot
.set_message_reaction(chat, MessageId(reaction.telegram_msg_id as i32))
.reaction(vec![ReactionType::Emoji {
emoji: reaction.emoji,
}])
.await
.err();
if let Some(e) = result {
tracing::warn!("set_message_reaction failed for row {}: {e}", row.id);
}
}
DispatchKind::Typing => {
}
DispatchKind::UnknownFallback => {
if let Some(e) = bot
.send_message(chat, format!("{}{attribution}", render_html(&row.text)))
.parse_mode(ParseMode::Html)
.await
.err()
{
tracing::warn!(
"send_message (unknown-kind fallback) failed for mailbox row {}: {e}",
row.id
);
}
}
}
}
async fn current_max(state: &Arc<State>, table: &str) -> i64 {
let sql = format!("SELECT COALESCE(MAX(id), 0) FROM {table}");
let c = state.conn.lock().await;
c.query_row(&sql, [], |r| r.get(0)).unwrap_or(0)
}
fn manager_of(conn: &Connection, agent_id: &str) -> Option<String> {
let row: Option<(String, i64, Option<String>)> = conn
.query_row(
"SELECT project_id, is_manager, reports_to FROM agents WHERE id = ?1",
params![agent_id],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)
.ok();
let (project, is_manager, reports_to) = row?;
if is_manager == 1 {
return Some(agent_id.to_string());
}
let role = reports_to?;
Some(format!("{project}:{role}"))
}
fn start_help_body(is_help: bool, name: Option<&str>) -> String {
match (is_help, name) {
(true, Some(name)) => format!(
"teamctl commands:\n\
/pending — show pending approvals\n\
/dm <project>:<agent> <text> — send to a different agent (rare)\n\
/<cmd> — slash-passthrough to {name}'s tmux session (Claude Code and Codex)\n\
\n\
Just type a message to chat with {name}."
),
(true, None) => "teamctl commands:\n\
/dm <project>:<agent> <message> — send a DM\n\
/pending — show pending approvals"
.into(),
(false, Some(name)) => format!("Connected to {name} via teamctl. Just type to chat."),
(false, None) => "Connected via teamctl. Send /help for commands.".into(),
}
}
fn should_route(scoped: Option<&str>, agent_id: &str, conn: &Connection) -> bool {
let Some(scoped) = scoped else {
return true;
};
let routed = manager_of(conn, agent_id).unwrap_or_else(|| agent_id.to_string());
routed == scoped
}
fn agent_runtime(conn: &Connection, agent_id: &str) -> Option<String> {
conn.query_row(
"SELECT runtime FROM agents WHERE id = ?1",
params![agent_id],
|r| r.get::<_, String>(0),
)
.ok()
}
#[derive(Debug, PartialEq, Eq)]
enum SlashOutcome {
Passthrough { session: String },
Reject { reason: String },
}
fn slash_outcome(manager: &str, runtime: &str, tmux_prefix: &str) -> SlashOutcome {
if runtime != "claude-code" && runtime != "codex" && runtime != "opencode" {
return SlashOutcome::Reject {
reason: format!(
"slash-passthrough is only supported on Claude Code, Codex, and \
OpenCode agents (this manager runs `{runtime}`)."
),
};
}
let (project, role) = match manager.split_once(':') {
Some((p, r)) => (p, r),
None => {
return SlashOutcome::Reject {
reason: format!("malformed manager id `{manager}` (expected `project:role`)."),
};
}
};
SlashOutcome::Passthrough {
session: format!("{tmux_prefix}{project}-{role}"),
}
}
fn tmux_type_body_argv<'a>(session: &'a str, body: &'a str) -> [&'a str; 5] {
["send-keys", "-t", session, "-l", body]
}
fn tmux_submit_argv(session: &str) -> [&str; 4] {
["send-keys", "-t", session, "Enter"]
}
fn tmux_send_keys(session: &str, body: &str) -> Result<(), String> {
let output = Command::new("tmux")
.args(tmux_type_body_argv(session, body))
.output()
.map_err(|e| format!("invoke tmux: {e}"))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
let trimmed = stderr.trim();
if trimmed.is_empty() {
return Err(format!("tmux exit {}", output.status));
}
return Err(format!("tmux exit {}: {trimmed}", output.status));
}
std::thread::sleep(std::time::Duration::from_millis(400));
let output = Command::new("tmux")
.args(tmux_submit_argv(session))
.output()
.map_err(|e| format!("invoke tmux: {e}"))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
let trimmed = stderr.trim();
if trimmed.is_empty() {
return Err(format!("tmux exit {}", output.status));
}
return Err(format!("tmux exit {}: {trimmed}", output.status));
}
Ok(())
}
const CC_SLASH_COMMANDS: &[(&str, &str)] = &[
("clear", "Clear conversation history"),
(
"compact",
"Compact conversation, optionally with focus instructions",
),
("cost", "Show token usage cost"),
("help", "Show available commands and shortcuts"),
("init", "Initialize a new CLAUDE.md file"),
("mcp", "Manage MCP servers"),
("model", "Set the AI model for Claude Code"),
("permissions", "View and edit permissions"),
("resume", "Resume a previous conversation"),
("review", "Review a pull request"),
("status", "Show Claude Code status"),
("vim", "Toggle between vim and default editing modes"),
];
const CODEX_SLASH_COMMANDS: &[(&str, &str)] = &[
("agent", "Inspect or switch subagent threads"),
("compact", "Compact the conversation context"),
("hooks", "Review and trust hooks"),
];
const OPENCODE_SLASH_COMMANDS: &[(&str, &str)] = &[
("models", "Switch the AI model"),
("new", "Start a new session"),
("sessions", "List and switch sessions"),
("status", "Show opencode status"),
("variants", "Cycle the model's reasoning variants"),
];
fn commands_for_runtime(runtime: Option<&str>) -> Vec<BotCommand> {
match runtime {
Some("claude-code") => CC_SLASH_COMMANDS
.iter()
.map(|(c, d)| BotCommand::new(*c, *d))
.collect(),
Some("codex") => CODEX_SLASH_COMMANDS
.iter()
.map(|(c, d)| BotCommand::new(*c, *d))
.collect(),
Some("opencode") => OPENCODE_SLASH_COMMANDS
.iter()
.map(|(c, d)| BotCommand::new(*c, *d))
.collect(),
_ => Vec::new(),
}
}
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
enum MediaKind {
Image,
File,
}
struct MediaIntent {
file_id: String,
extension: String,
mime: String,
kind: MediaKind,
}
fn classify_media_intent(msg: &Message) -> Option<MediaIntent> {
if let Some(photos) = msg.photo() {
let largest = photos
.iter()
.max_by_key(|p| (p.width as u64).saturating_mul(p.height as u64))?;
return Some(MediaIntent {
file_id: largest.file.id.clone(),
extension: "jpg".into(),
mime: "image/jpeg".into(),
kind: MediaKind::Image,
});
}
if let Some(doc) = msg.document() {
let mime = doc
.mime_type
.as_ref()
.map(|m| m.essence_str().to_string())
.unwrap_or_else(|| "application/octet-stream".to_string());
let extension = extension_for_document(doc.file_name.as_deref(), &mime);
let kind = if mime.starts_with("image/") {
MediaKind::Image
} else {
MediaKind::File
};
return Some(MediaIntent {
file_id: doc.file.id.clone(),
extension,
mime,
kind,
});
}
None
}
fn extension_for_document(filename: Option<&str>, mime: &str) -> String {
if let Some(name) = filename {
if let Some((_, ext)) = name.rsplit_once('.') {
if !ext.is_empty() && ext.len() <= 8 && ext.chars().all(|c| c.is_ascii_alphanumeric()) {
return ext.to_ascii_lowercase();
}
}
}
extension_from_mime(mime).into()
}
fn extension_from_mime(mime: &str) -> &'static str {
match mime {
"image/png" => "png",
"image/jpeg" => "jpg",
"image/webp" => "webp",
"image/gif" => "gif",
"application/pdf" => "pdf",
"text/plain" => "txt",
"text/csv" => "csv",
"application/zip" => "zip",
"application/json" => "json",
_ => "bin",
}
}
fn inbound_media_path(media_root: &Path, project: &str, row_id: i64, extension: &str) -> PathBuf {
media_root
.join(project)
.join(format!("{row_id}.{extension}"))
}
fn media_success_payload(path: &Path, caption: &str, mime: &str, size_bytes: u64) -> String {
let mut payload = serde_json::json!({
"path": path.display().to_string(),
"mime": mime,
"size_bytes": size_bytes,
});
if !caption.is_empty() {
payload["caption"] = serde_json::Value::String(caption.to_string());
}
payload.to_string()
}
fn media_error_payload(caption: &str, error: &str) -> String {
let mut payload = serde_json::json!({ "error": error });
if !caption.is_empty() {
payload["caption"] = serde_json::Value::String(caption.to_string());
}
payload.to_string()
}
async fn download_to(bot: &Bot, file_id: &str, path: &Path, dir: &Path) -> Result<u64, String> {
use tokio::io::AsyncWriteExt;
tokio::fs::create_dir_all(dir)
.await
.map_err(|e| format!("create_dir_all `{}`: {e}", dir.display()))?;
let file = bot
.get_file(file_id)
.await
.map_err(|e| format!("get_file: {e}"))?;
if (file.size as usize) > MAX_DOWNLOAD_BYTES {
return Err(media_size_pre_reject(
"media file",
file.size,
MAX_DOWNLOAD_BYTES,
));
}
let handle = tokio::fs::File::create(path)
.await
.map_err(|e| format!("create file `{}`: {e}", path.display()))?;
let mut bounded = BoundedWriter::new("media file", handle, MAX_DOWNLOAD_BYTES);
bot.download_file(&file.path, &mut bounded)
.await
.map_err(|e| format!("download_file: {e}"))?;
let mut handle = bounded.into_inner();
handle.flush().await.ok();
drop(handle);
let meta = tokio::fs::metadata(path)
.await
.map_err(|e| format!("metadata: {e}"))?;
Ok(meta.len())
}
async fn handle_inbound_media(bot: &Bot, msg: &Message, state: &State) -> ResponseResult<()> {
let Some(manager) = state.manager.as_deref() else {
bot.send_message(
msg.chat.id,
"media uploads need a manager-scoped bot. \
Run `teamctl bot up` to attach this bot to a manager.",
)
.await?;
return Ok(());
};
let Some((project, _)) = manager.split_once(':') else {
return Ok(());
};
let Some(intent) = classify_media_intent(msg) else {
return Ok(());
};
let caption = msg.caption().unwrap_or("").to_string();
let placeholder_payload = serde_json::json!({ "caption": caption }).to_string();
let row_id_opt = {
let c = state.conn.lock().await;
match c.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, kind, structured_payload)
VALUES (?1, 'user:telegram', ?2, ?3, strftime('%s','now'),
'media_pending', ?4)",
params![project, manager, &caption, placeholder_payload],
) {
Ok(_) => Some(c.last_insert_rowid()),
Err(e) => {
tracing::error!("inbound media: failed to insert placeholder row: {e}");
None
}
}
};
let Some(row_id) = row_id_opt else {
bot.send_message(
msg.chat.id,
"internal error: could not record the message; please retry.",
)
.await?;
return Ok(());
};
let path = inbound_media_path(&state.media_root, project, row_id, &intent.extension);
let dir = path.parent().unwrap_or(&state.media_root).to_path_buf();
match download_to(bot, &intent.file_id, &path, &dir).await {
Ok(size_bytes) => {
let payload = media_success_payload(&path, &caption, &intent.mime, size_bytes);
let kind = match intent.kind {
MediaKind::Image => "image",
MediaKind::File => "file",
};
let c = state.conn.lock().await;
if let Err(e) = update_message_kind(&c, row_id, kind, &payload) {
tracing::error!("inbound media: failed to finalize message {row_id}: {e}");
}
drop(c);
bot.send_message(msg.chat.id, format!("→ {manager}"))
.await?;
}
Err(err) => {
let payload = media_error_payload(&caption, &err);
let c = state.conn.lock().await;
if let Err(e) = update_message_kind(&c, row_id, "media_error", &payload) {
tracing::error!(
"inbound media: failed to record media_error on message {row_id}: {e}"
);
}
drop(c);
bot.send_message(msg.chat.id, format!("media download failed: {err}"))
.await?;
}
}
Ok(())
}
fn update_message_kind(conn: &Connection, id: i64, kind: &str, payload: &str) -> Result<usize> {
if team_core::mailbox::is_privileged_kind(kind) {
bail!(
"refusing UPDATE that would set privileged kind='{kind}' on message {id} \
(#320): the privileged kind is insert-time only"
);
}
conn.execute(
"UPDATE messages SET kind = ?1, structured_payload = ?2 WHERE id = ?3",
params![kind, payload, id],
)
.with_context(|| format!("update message {id} kind to '{kind}'"))
}
fn render_html(s: &str) -> String {
let mut out = String::with_capacity(s.len() + s.len() / 8);
let lines: Vec<&str> = s.lines().collect();
let mut i = 0;
let mut first = true;
while i < lines.len() {
if !first {
out.push('\n');
}
first = false;
let line = lines[i];
if let Some(lang) = fence_marker(line) {
let close_idx = ((i + 1)..lines.len()).find(|&j| fence_marker(lines[j]).is_some());
if let Some(close) = close_idx {
if lang.is_empty() {
out.push_str("<pre>");
} else {
out.push_str("<pre><code class=\"language-");
html_escape_into(&mut out, &lang);
out.push_str("\">");
}
for (k, body_line) in lines[(i + 1)..close].iter().enumerate() {
if k > 0 {
out.push('\n');
}
html_escape_into(&mut out, body_line);
}
if lang.is_empty() {
out.push_str("</pre>");
} else {
out.push_str("</code></pre>");
}
i = close + 1;
continue;
}
}
render_normal_line(line, &mut out);
i += 1;
}
out
}
fn fence_marker(line: &str) -> Option<String> {
let trimmed = line.trim_start();
let after = trimmed.strip_prefix("```")?;
Some(
after
.chars()
.take_while(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_'))
.collect(),
)
}
fn render_normal_line(line: &str, out: &mut String) {
let trimmed = line.trim_start();
let leading = &line[..line.len() - trimmed.len()];
let body = if let Some(rest) = trimmed
.strip_prefix("- ")
.or_else(|| trimmed.strip_prefix("* "))
.or_else(|| trimmed.strip_prefix("+ "))
{
format!("• {rest}")
} else {
trimmed.to_string()
};
out.push_str(leading);
render_inline_html(&body, out);
}
fn render_inline_html(s: &str, out: &mut String) {
let bytes = s.as_bytes();
let mut i = 0;
while i < bytes.len() {
if bytes.get(i..i + 2) == Some(b"**") {
if let Some(end) = s[i + 2..].find("**").filter(|&e| e > 0) {
let close = i + 2 + end;
out.push_str("<b>");
html_escape_into(out, &s[i + 2..close]);
out.push_str("</b>");
i = close + 2;
continue;
}
}
if bytes.get(i..i + 2) == Some(b"__") {
if let Some(end) = s[i + 2..].find("__").filter(|&e| e > 0) {
let close = i + 2 + end;
out.push_str("<b>");
html_escape_into(out, &s[i + 2..close]);
out.push_str("</b>");
i = close + 2;
continue;
}
}
if bytes[i] == b'`' {
if let Some(end) = s[i + 1..].find('`').filter(|&e| e > 0) {
let close = i + 1 + end;
out.push_str("<code>");
html_escape_into(out, &s[i + 1..close]);
out.push_str("</code>");
i = close + 1;
continue;
}
}
if bytes[i] == b'*' {
if let Some(end) = s[i + 1..].find('*').filter(|&e| e > 0) {
let close = i + 1 + end;
out.push_str("<i>");
html_escape_into(out, &s[i + 1..close]);
out.push_str("</i>");
i = close + 1;
continue;
}
}
let next = s[i..]
.chars()
.next()
.expect("byte index inside string bounds yields a char");
match next {
'<' => out.push_str("<"),
'>' => out.push_str(">"),
'&' => out.push_str("&"),
_ => out.push(next),
}
i += next.len_utf8();
}
}
fn html_escape_into(out: &mut String, s: &str) {
for c in s.chars() {
match c {
'<' => out.push_str("<"),
'>' => out.push_str(">"),
'&' => out.push_str("&"),
_ => out.push(c),
}
}
}
fn html_escape_str(s: &str) -> String {
let mut out = String::with_capacity(s.len());
html_escape_into(&mut out, s);
out
}
fn markdown_escape_str(s: &str) -> String {
let mut out = String::with_capacity(s.len());
for c in s.chars() {
if matches!(c, '_' | '*' | '`' | '[' | ']' | '(' | ')' | '~' | '\\') {
out.push('\\');
}
out.push(c);
}
out
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum SttOutcome {
Ok(String),
Skipped,
Failed(String),
}
#[derive(Debug, PartialEq, Eq)]
struct VoiceDecision {
user_reply: String,
inbox_text: Option<String>,
}
fn map_voice_outcome(outcome: &SttOutcome) -> VoiceDecision {
match outcome {
SttOutcome::Ok(transcript) => VoiceDecision {
user_reply: format!("🎙 \"{transcript}\""),
inbox_text: Some(format!("{VOICE_INBOX_PREFIX} {transcript}")),
},
SttOutcome::Skipped => VoiceDecision {
user_reply: "🎙 (couldn't capture anything. did you say something? — skipping)"
.to_string(),
inbox_text: None,
},
SttOutcome::Failed(err) => VoiceDecision {
user_reply: format!("🎙 transcription failed: {err}"),
inbox_text: None,
},
}
}
async fn handle_voice(bot: &Bot, msg: &Message, state: &State) -> ResponseResult<()> {
let manager = state.manager.as_deref().expect("checked by caller");
let stt = state.stt.as_ref().expect("checked by caller");
let Some((project, _)) = manager.split_once(':') else {
return Ok(());
};
let Some(voice) = msg.voice() else {
return Ok(());
};
let file_id = voice.file.id.clone();
let inbound_msg_id: i64 = msg.id.0 as i64;
let reply_to = ReplyParameters::new(msg.id);
let _ = bot.send_chat_action(msg.chat.id, ChatAction::Typing).await;
let audio = match download_voice_bytes(bot, &file_id).await {
Ok(bytes) => bytes,
Err(err) => {
let decision = map_voice_outcome(&SttOutcome::Failed(err));
bot.send_message(msg.chat.id, decision.user_reply)
.reply_parameters(reply_to)
.await?;
return Ok(());
}
};
let outcome = transcribe(&audio, stt).await;
let decision = map_voice_outcome(&outcome);
let mut insert_ok = false;
if let Some(inbox_text) = decision.inbox_text.as_deref() {
let c = state.conn.lock().await;
match c.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES (?1, 'user:telegram', ?2, ?3, strftime('%s','now'), ?4)",
params![project, manager, inbox_text, inbound_msg_id],
) {
Ok(_) => insert_ok = true,
Err(e) => tracing::error!(
"voice transcript INSERT failed for {manager}: {e} (operator was \
told what was heard but the agent will not receive it)"
),
}
}
if voice_should_echo_routing(&decision, insert_ok) {
bot.send_message(msg.chat.id, format!("→ {manager}"))
.await?;
}
bot.send_message(msg.chat.id, decision.user_reply)
.reply_parameters(reply_to)
.await?;
Ok(())
}
fn voice_should_echo_routing(decision: &VoiceDecision, insert_ok: bool) -> bool {
decision.inbox_text.is_some() && insert_ok
}
fn voice_stt_missing_reply() -> &'static str {
"🎙 Voice isn't configured for this agent yet.\n\n\
To enable, either run `/teamctl:adjust` in your project's Claude Code \
to configure it conversationally, or add `interfaces.telegram.speech_to_text` \
to the agent's project YAML manually.\n\n\
Docs: https://teamctl.run/guides/telegram-bot/#voice-messages-optional"
}
async fn handle_voice_stt_missing(bot: &Bot, msg: &Message) -> ResponseResult<()> {
let reply_to = ReplyParameters::new(msg.id);
bot.send_message(msg.chat.id, voice_stt_missing_reply())
.reply_parameters(reply_to)
.await?;
Ok(())
}
const MAX_VOICE_BYTES: usize = 2 * 1024 * 1024;
const MAX_DOWNLOAD_BYTES: usize = 50 * 1024 * 1024;
fn media_size_pre_reject(kind: &str, reported: u32, max: usize) -> String {
format!("{kind} too large: {reported} bytes (max {max})")
}
fn media_size_mid_reject(kind: &str, sofar: usize, incoming: usize, max: usize) -> String {
let proposed = sofar.saturating_add(incoming);
format!(
"{kind} exceeded ceiling mid-download \
(max {max} bytes; upstream sent at least {proposed})"
)
}
fn voice_size_pre_reject(reported: u32, max: usize) -> String {
media_size_pre_reject("voice file", reported, max)
}
struct BoundedWriter<W> {
kind: &'static str,
inner: W,
max: usize,
written: usize,
}
impl<W> BoundedWriter<W> {
fn new(kind: &'static str, inner: W, max: usize) -> Self {
Self {
kind,
inner,
max,
written: 0,
}
}
fn into_inner(self) -> W {
self.inner
}
}
impl<W: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for BoundedWriter<W> {
fn poll_write(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
let this = self.get_mut();
if this.written.saturating_add(buf.len()) > this.max {
let msg = media_size_mid_reject(this.kind, this.written, buf.len(), this.max);
return std::task::Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
msg,
)));
}
match std::pin::Pin::new(&mut this.inner).poll_write(cx, buf) {
std::task::Poll::Ready(Ok(n)) => {
this.written = this.written.saturating_add(n);
std::task::Poll::Ready(Ok(n))
}
other => other,
}
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let this = self.get_mut();
std::pin::Pin::new(&mut this.inner).poll_flush(cx)
}
fn poll_shutdown(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let this = self.get_mut();
std::pin::Pin::new(&mut this.inner).poll_shutdown(cx)
}
}
async fn download_voice_bytes(bot: &Bot, file_id: &str) -> Result<Vec<u8>, String> {
use tokio::io::AsyncWriteExt;
let file = bot
.get_file(file_id)
.await
.map_err(|e| format!("get_file: {e}"))?;
if (file.size as usize) > MAX_VOICE_BYTES {
return Err(voice_size_pre_reject(file.size, MAX_VOICE_BYTES));
}
let buf: Vec<u8> = Vec::with_capacity(file.size as usize);
let mut bounded = BoundedWriter::new("voice file", buf, MAX_VOICE_BYTES);
bot.download_file(&file.path, &mut bounded)
.await
.map_err(|e| format!("download_file: {e}"))?;
let mut buf = bounded.into_inner();
buf.flush().await.ok();
Ok(buf)
}
async fn transcribe(audio: &[u8], stt: &SttRuntime) -> SttOutcome {
match stt.provider.as_str() {
"groq" => transcribe_groq(audio, stt).await,
other => SttOutcome::Failed(format!("unknown stt provider `{other}`")),
}
}
async fn transcribe_groq(audio: &[u8], stt: &SttRuntime) -> SttOutcome {
let part = match reqwest::multipart::Part::bytes(audio.to_vec())
.file_name("voice.ogg")
.mime_str("audio/ogg")
{
Ok(p) => p,
Err(e) => return SttOutcome::Failed(format!("multipart: {e}")),
};
let mut form = reqwest::multipart::Form::new()
.part("file", part)
.text("model", stt.model.clone())
.text("response_format", "text");
if let Some(lang) = &stt.language {
form = form.text("language", lang.clone());
}
let resp = stt
.http
.post("https://api.groq.com/openai/v1/audio/transcriptions")
.bearer_auth(&stt.api_key)
.multipart(form)
.send()
.await;
let resp = match resp {
Ok(r) => r,
Err(e) => return SttOutcome::Failed(format!("groq request: {e}")),
};
let status = resp.status();
let body = match resp.text().await {
Ok(b) => b,
Err(e) => return SttOutcome::Failed(format!("groq read body: {e}")),
};
if !status.is_success() {
return SttOutcome::Failed(format!("groq {status}: {}", body.trim()));
}
let trimmed = body.trim();
if trimmed.is_empty() {
SttOutcome::Skipped
} else {
SttOutcome::Ok(trimmed.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
#[test]
fn peel_readnow_strips_prefix_when_present() {
assert_eq!(
peel_readnow("/readnow build broke"),
("build broke", Some("immediate")),
);
}
#[test]
fn peel_readnow_passes_through_when_prefix_absent() {
assert_eq!(peel_readnow("regular message"), ("regular message", None));
}
#[test]
fn peel_readnow_is_case_sensitive() {
assert_eq!(peel_readnow("/ReadNow x"), ("/ReadNow x", None));
assert_eq!(peel_readnow("/READNOW x"), ("/READNOW x", None));
}
#[test]
fn peel_readnow_requires_single_space_separator() {
assert_eq!(peel_readnow("/readnowfoo"), ("/readnowfoo", None));
assert_eq!(peel_readnow("/readnow x"), (" x", Some("immediate")));
}
#[test]
fn peel_readnow_with_empty_body_after_prefix() {
assert_eq!(peel_readnow("/readnow "), ("", Some("immediate")));
}
#[test]
fn truncate_quote_passes_short_text_through() {
assert_eq!(truncate_quote("hello"), "hello");
let exactly = "a".repeat(INBOUND_QUOTE_MAX_CHARS);
assert_eq!(
truncate_quote(&exactly),
exactly,
"exactly at the cap is untouched"
);
}
#[test]
fn truncate_quote_cuts_long_text_with_ellipsis() {
let long = "a".repeat(INBOUND_QUOTE_MAX_CHARS + 100);
let out = truncate_quote(&long);
assert_eq!(out.chars().count(), INBOUND_QUOTE_MAX_CHARS + 1);
assert!(out.ends_with('…'));
assert!(out.starts_with(&"a".repeat(INBOUND_QUOTE_MAX_CHARS)));
}
#[test]
fn truncate_quote_counts_chars_not_bytes() {
let multibyte = "é".repeat(INBOUND_QUOTE_MAX_CHARS + 50);
let out = truncate_quote(&multibyte);
assert_eq!(out.chars().count(), INBOUND_QUOTE_MAX_CHARS + 1);
assert!(out.ends_with('…'));
}
#[test]
fn build_inbound_body_passes_plain_message_through() {
assert_eq!(build_inbound_body("just a message", None), "just a message");
}
#[test]
fn build_inbound_body_prefixes_reply_with_mailbox_id() {
let ctx = ReplyContext {
sender: "user:telegram".to_string(),
mailbox_id: Some(730),
quoted: Some("handle the auth case".to_string()),
};
assert_eq!(
build_inbound_body("yes please", Some(&ctx)),
"> [reply to user:telegram msg 730]\n> handle the auth case\n\nyes please"
);
}
#[test]
fn build_inbound_body_omits_msg_id_when_unresolved() {
let ctx = ReplyContext {
sender: "agent".to_string(),
mailbox_id: None,
quoted: Some("earlier note".to_string()),
};
assert_eq!(
build_inbound_body("ok", Some(&ctx)),
"> [reply to agent]\n> earlier note\n\nok"
);
}
#[test]
fn build_inbound_body_blockquotes_each_line_of_a_multiline_quote() {
let ctx = ReplyContext {
sender: "p:eng_lead".to_string(),
mailbox_id: Some(12),
quoted: Some("line one\nline two".to_string()),
};
assert_eq!(
build_inbound_body("got it", Some(&ctx)),
"> [reply to p:eng_lead msg 12]\n> line one\n> line two\n\ngot it"
);
}
#[test]
fn build_inbound_body_header_only_when_nothing_quotable() {
let ctx = ReplyContext {
sender: "user:telegram".to_string(),
mailbox_id: Some(5),
quoted: None,
};
assert_eq!(
build_inbound_body("what about this", Some(&ctx)),
"> [reply to user:telegram msg 5]\n\nwhat about this"
);
}
#[test]
fn lookup_replied_row_finds_row_by_telegram_id() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
conn.execute(
"INSERT INTO messages (project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES ('p','user:telegram','p:eng_lead','hi',0,730)",
[],
)
.unwrap();
let user_row = conn.last_insert_rowid();
conn.execute(
"INSERT INTO messages (project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES ('p','p:eng_lead','user:telegram','on it',0,731)",
[],
)
.unwrap();
let agent_row = conn.last_insert_rowid();
assert_eq!(
lookup_replied_row(&conn, 730),
Some((user_row, "user:telegram".to_string())),
"a user's own message resolves to its mailbox row + sender"
);
assert_eq!(
lookup_replied_row(&conn, 731),
Some((agent_row, "p:eng_lead".to_string())),
"an agent's outbound message resolves to its <project>:<agent> sender"
);
}
#[test]
fn lookup_replied_row_returns_none_when_absent() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert_eq!(lookup_replied_row(&conn, 999), None);
}
#[test]
fn lookup_replied_row_picks_most_recent_on_telegram_id_collision() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
conn.execute(
"INSERT INTO messages (project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES ('p','user:telegram','p:eng_lead','first',0,42)",
[],
)
.unwrap();
conn.execute(
"INSERT INTO messages (project_id, sender, recipient, text, sent_at, telegram_msg_id)
VALUES ('p','p:eng_lead','user:telegram','second',0,42)",
[],
)
.unwrap();
let newer = conn.last_insert_rowid();
assert_eq!(
lookup_replied_row(&conn, 42),
Some((newer, "p:eng_lead".to_string())),
"collision resolves to the most recently inserted row"
);
}
#[test]
fn sender_label_for_classifies_bot_human_and_anonymous() {
assert_eq!(sender_label_for(Some(true)), "agent");
assert_eq!(sender_label_for(Some(false)), "user:telegram");
assert_eq!(sender_label_for(None), "unknown");
}
#[test]
fn build_inbound_body_renders_quote_only_reply() {
let ctx = ReplyContext {
sender: "unknown".to_string(),
mailbox_id: None,
quoted: Some("the part I mean".to_string()),
};
assert_eq!(
build_inbound_body("clarify this", Some(&ctx)),
"> [reply to unknown]\n> the part I mean\n\nclarify this"
);
}
#[test]
fn approval_outcome_line_uses_approver_first_name() {
assert_eq!(approval_outcome_line(true, "Hamed"), "✅ Approved by Hamed",);
assert_eq!(
approval_outcome_line(false, "Hamed"),
"❌ Rejected by Hamed",
);
}
#[test]
fn approval_outcome_line_handles_unicode_first_name() {
assert_eq!(
approval_outcome_line(true, "علیرضا"),
"✅ Approved by علیرضا",
);
}
#[test]
fn decision_and_cancel_outcome_lines_name_the_chooser() {
assert_eq!(
decision_outcome_line("Ship it", "Hamed"),
"✅ Ship it — chosen by Hamed",
);
assert_eq!(cancel_outcome_line("Hamed"), "🚫 Cancelled by Hamed");
assert_eq!(
decision_outcome_line("گزینه", "علیرضا"),
"✅ گزینه — chosen by علیرضا",
);
}
#[test]
fn parse_callback_accepts_all_four_verbs() {
assert_eq!(parse_callback("approve:7"), Some((7, CbAction::Approve)));
assert_eq!(parse_callback("deny:7"), Some((7, CbAction::Deny)));
assert_eq!(parse_callback("cancel:42"), Some((42, CbAction::Cancel)));
assert_eq!(parse_callback("opt:42:2"), Some((42, CbAction::Opt(2))));
}
#[test]
fn parse_callback_rejects_malformed() {
assert_eq!(parse_callback("approve"), None);
assert_eq!(parse_callback("approve:x"), None);
assert_eq!(parse_callback("frobnicate:7"), None);
assert_eq!(parse_callback("opt:7:notanum"), None);
assert_eq!(parse_callback("opt:7"), None);
assert_eq!(parse_callback("approve:7:8"), None);
assert_eq!(parse_callback("cancel:7:8"), None);
}
#[test]
fn decode_options_handles_null_and_garbage() {
assert!(decode_options(None).is_empty());
assert!(decode_options(Some("not json")).is_empty());
assert_eq!(
decode_options(Some(r#"[{"label":"A","value":"a"},{"value":"b"}]"#)),
vec![("A".to_string(), "a".to_string())],
);
assert_eq!(
decode_options(Some(
r#"[{"label":"Yes","value":"y"},{"label":"No","value":"n"}]"#
)),
vec![
("Yes".to_string(), "y".to_string()),
("No".to_string(), "n".to_string()),
],
);
}
fn kb_pairs(kb: &InlineKeyboardMarkup) -> Vec<(String, String)> {
use teloxide::types::InlineKeyboardButtonKind::CallbackData;
kb.inline_keyboard
.iter()
.flatten()
.map(|b| {
let data = match &b.kind {
CallbackData(d) => d.clone(),
_ => panic!("expected callback button"),
};
(b.text.clone(), data)
})
.collect()
}
#[test]
fn approval_keyboard_empty_options_is_binary_back_compat() {
let kb = approval_keyboard(7, &[]);
assert_eq!(
kb_pairs(&kb),
vec![
("Approve".to_string(), "approve:7".to_string()),
("Deny".to_string(), "deny:7".to_string()),
],
);
assert_eq!(kb.inline_keyboard.len(), 1, "binary is a single row");
}
#[test]
fn approval_keyboard_multi_renders_options_then_cancel() {
let opts = vec![
("Ship".to_string(), "ship".to_string()),
("Hold".to_string(), "hold".to_string()),
("Rework".to_string(), "rework".to_string()),
];
let kb = approval_keyboard(9, &opts);
assert_eq!(
kb_pairs(&kb),
vec![
("Ship".to_string(), "opt:9:0".to_string()),
("Hold".to_string(), "opt:9:1".to_string()),
("Rework".to_string(), "opt:9:2".to_string()),
("Cancel".to_string(), "cancel:9".to_string()),
],
);
assert_eq!(
kb.inline_keyboard.len(),
4,
"one row per option + a Cancel row"
);
}
fn seed(conn: &Connection) {
team_core::mailbox::ensure(conn).unwrap();
conn.execute(
"INSERT OR IGNORE INTO projects (id, name) VALUES ('p','P')",
[],
)
.unwrap();
conn.execute(
"INSERT OR IGNORE INTO agents (id, project_id, role, runtime, is_manager, reports_to)
VALUES ('p:eng_lead','p','eng_lead','claude-code',1,NULL)",
[],
)
.unwrap();
conn.execute(
"INSERT OR IGNORE INTO agents (id, project_id, role, runtime, is_manager, reports_to)
VALUES ('p:dev1','p','dev1','claude-code',0,'eng_lead')",
[],
)
.unwrap();
conn.execute(
"INSERT OR IGNORE INTO agents (id, project_id, role, runtime, is_manager, reports_to)
VALUES ('p:pm','p','pm','claude-code',1,NULL)",
[],
)
.unwrap();
}
#[test]
fn manager_of_returns_self_for_a_manager() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert_eq!(
manager_of(&conn, "p:eng_lead").as_deref(),
Some("p:eng_lead")
);
assert_eq!(manager_of(&conn, "p:pm").as_deref(), Some("p:pm"));
}
#[test]
fn manager_of_resolves_reports_to_for_a_worker() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert_eq!(manager_of(&conn, "p:dev1").as_deref(), Some("p:eng_lead"));
}
#[test]
fn manager_of_returns_none_for_unknown_agent() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert!(manager_of(&conn, "p:ghost").is_none());
}
#[test]
fn classify_kind_treats_null_and_empty_as_text() {
assert_eq!(classify_kind(None), DispatchKind::Text);
assert_eq!(classify_kind(Some("text")), DispatchKind::Text);
assert_eq!(classify_kind(Some("")), DispatchKind::Text);
}
#[test]
fn classify_kind_routes_image_and_file() {
assert_eq!(classify_kind(Some("image")), DispatchKind::Image);
assert_eq!(classify_kind(Some("file")), DispatchKind::File);
}
#[test]
fn classify_kind_falls_back_for_unknown_kinds() {
assert_eq!(
classify_kind(Some("garbage")),
DispatchKind::UnknownFallback
);
assert_eq!(classify_kind(Some("custom")), DispatchKind::UnknownFallback);
}
#[test]
fn parse_payload_extracts_source_value_and_caption() {
let p = parse_payload(r#"{"source":"path","value":"/tmp/x.png","caption":"hi"}"#)
.expect("payload parses");
assert_eq!(p.source, "path");
assert_eq!(p.value, "/tmp/x.png");
assert_eq!(p.caption.as_deref(), Some("hi"));
}
#[test]
fn parse_payload_handles_missing_caption() {
let p = parse_payload(r#"{"source":"url","value":"https://x.test/a.png"}"#)
.expect("payload parses");
assert_eq!(p.source, "url");
assert!(p.caption.is_none());
}
#[test]
fn parse_payload_returns_none_on_garbage() {
assert!(parse_payload("not json").is_none());
assert!(
parse_payload(r#"{"value":"x"}"#).is_none(),
"missing source"
);
assert!(
parse_payload(r#"{"source":"path"}"#).is_none(),
"missing value"
);
}
#[test]
fn input_file_from_path_and_url_both_construct() {
let p = parse_payload(r#"{"source":"path","value":"/tmp/x.png"}"#).unwrap();
assert!(input_file_from(&p).is_some());
let p = parse_payload(r#"{"source":"url","value":"https://x.test/a.png"}"#).unwrap();
assert!(input_file_from(&p).is_some());
}
#[test]
fn input_file_from_unknown_source_returns_none() {
let p = MediaPayload {
source: "bytes".into(),
value: "abc".into(),
caption: None,
};
assert!(input_file_from(&p).is_none());
}
#[allow(clippy::too_many_arguments)]
fn insert_row(
conn: &Connection,
sender: &str,
text: &str,
kind: Option<&str>,
payload: Option<&str>,
telegram_msg_id: Option<i64>,
) -> i64 {
let project = sender.split_once(':').map(|(p, _)| p).unwrap_or("p");
conn.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at,
kind, structured_payload, telegram_msg_id)
VALUES (?1, ?2, 'user:telegram', ?3, strftime('%s','now'), ?4, ?5, ?6)",
params![project, sender, text, kind, payload, telegram_msg_id],
)
.unwrap();
conn.last_insert_rowid()
}
const OUTBOUND_SELECT: &str =
"SELECT m.id, m.sender, m.text, m.kind, m.structured_payload, m.telegram_msg_id
FROM messages m
WHERE m.id > ?1
AND m.recipient = 'user:telegram'
AND m.acked_at IS NULL
ORDER BY m.id";
#[test]
fn outbound_select_returns_kind_and_payload_for_structured_rows() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_row(
&conn,
"p:eng_lead",
"shot",
Some("image"),
Some(r#"{"source":"path","value":"/tmp/a.png"}"#),
None,
);
let mut stmt = conn.prepare(OUTBOUND_SELECT).unwrap();
let rows: Vec<MailboxRow> = stmt
.query_map(params![0i64], MailboxRow::from_row)
.unwrap()
.flatten()
.collect();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].id, id);
assert_eq!(rows[0].kind.as_deref(), Some("image"));
assert!(rows[0].payload.as_deref().unwrap().contains("/tmp/a.png"));
}
#[test]
fn outbound_select_returns_null_kind_for_legacy_text_rows() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_row(&conn, "p:eng_lead", "hello", None, None, None);
let mut stmt = conn.prepare(OUTBOUND_SELECT).unwrap();
let rows: Vec<MailboxRow> = stmt
.query_map(params![0i64], MailboxRow::from_row)
.unwrap()
.flatten()
.collect();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].id, id);
assert!(rows[0].kind.is_none());
assert!(rows[0].payload.is_none());
assert!(rows[0].telegram_msg_id.is_none());
assert_eq!(classify_kind(rows[0].kind.as_deref()), DispatchKind::Text);
}
#[test]
fn outbound_select_returns_telegram_msg_id_when_set_for_threaded_rows() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_row(&conn, "p:eng_lead", "ack", None, None, Some(7777));
let mut stmt = conn.prepare(OUTBOUND_SELECT).unwrap();
let rows: Vec<MailboxRow> = stmt
.query_map(params![0i64], MailboxRow::from_row)
.unwrap()
.flatten()
.collect();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].id, id);
assert_eq!(rows[0].telegram_msg_id, Some(7777));
}
#[test]
fn render_html_paired_bold_emphasis() {
assert_eq!(render_html("**bold** text"), "<b>bold</b> text");
assert_eq!(render_html("__also bold__"), "<b>also bold</b>");
}
#[test]
fn render_html_paired_italic_and_inline_code() {
assert_eq!(render_html("*italic* text"), "<i>italic</i> text");
assert_eq!(
render_html("plain `code` here"),
"plain <code>code</code> here"
);
}
#[test]
fn render_html_translates_list_bullets() {
let input = "- one\n- two\n * nested\n+ three";
let expected = "• one\n• two\n • nested\n• three";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_preserves_emoji_and_converts_inline() {
let input = "🔐 deploy\nrouting prompt to one channel — the **right** one";
let expected = "🔐 deploy\nrouting prompt to one channel — the <b>right</b> one";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_leaves_single_underscore_alone() {
assert_eq!(render_html("thread_id"), "thread_id");
assert_eq!(render_html("snake_case_var here"), "snake_case_var here");
assert_eq!(render_html("_underscored_"), "_underscored_");
}
#[test]
fn render_html_unmatched_delimiters_pass_through() {
assert_eq!(render_html("array[i] = b * 2"), "array[i] = b * 2");
assert_eq!(render_html("unmatched `tick"), "unmatched `tick");
assert_eq!(
render_html("unmatched **bold-open"),
"unmatched **bold-open"
);
}
#[test]
fn render_html_pairing_is_per_line() {
let input = "*open\nclose*";
assert_eq!(render_html(input), "*open\nclose*");
}
#[test]
fn render_html_escapes_lt_gt_amp_in_raw_text() {
assert_eq!(
render_html("<channel source=\"team\"> & friends"),
"<channel source=\"team\"> & friends",
);
}
#[test]
fn render_html_escapes_inside_inline_code() {
assert_eq!(
render_html("see `<thing>` for more"),
"see <code><thing></code> for more",
);
}
#[test]
fn render_html_fenced_block_no_language() {
let input = "before\n```\nlet x = 1;\n```\nafter";
let expected = "before\n<pre>let x = 1;</pre>\nafter";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_fenced_block_with_language_tag() {
let input = "```rust\nfn main() {}\n```";
let expected = "<pre><code class=\"language-rust\">fn main() {}</code></pre>";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_fenced_block_escapes_html_inside() {
let input = "```\n<channel> & co\n```";
let expected = "<pre><channel> & co</pre>";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_unmatched_fence_falls_through_as_normal_line() {
let input = "```\nstray";
assert_eq!(render_html(input), "```\nstray");
}
#[test]
fn html_escape_str_escapes_the_three_html_specials_only() {
assert_eq!(
html_escape_str("<channel> & friends"),
"<channel> & friends",
);
assert_eq!(
html_escape_str("safe-text_with.no.specials"),
"safe-text_with.no.specials",
);
assert_eq!(html_escape_str("she said \"hi\""), "she said \"hi\"");
}
#[test]
fn fence_marker_takes_leading_alphanumeric_dash_underscore_run_as_lang() {
assert_eq!(fence_marker("```").as_deref(), Some(""));
assert_eq!(fence_marker("```rust").as_deref(), Some("rust"));
assert_eq!(fence_marker("```rust // example").as_deref(), Some("rust"));
assert_eq!(
fence_marker(" ```python extra junk").as_deref(),
Some("python"),
);
assert_eq!(fence_marker("not a fence").as_deref(), None);
}
#[test]
fn fence_marker_admits_dash_and_underscore_in_lang() {
assert_eq!(
fence_marker("```shell-script").as_deref(),
Some("shell-script"),
);
assert_eq!(
fence_marker("```objective-c").as_deref(),
Some("objective-c"),
);
assert_eq!(
fence_marker("```objective_c").as_deref(),
Some("objective_c"),
);
assert_eq!(fence_marker("```-rust").as_deref(), Some("-rust"));
assert_eq!(fence_marker("```_rust").as_deref(), Some("_rust"));
}
#[test]
fn fence_marker_truncates_lang_at_quote_for_attribute_injection_safety() {
assert_eq!(fence_marker("```\"x").as_deref(), Some(""));
assert_eq!(fence_marker("```rust\"injected\"").as_deref(), Some("rust"));
assert_eq!(fence_marker("```\"").as_deref(), Some(""));
}
#[test]
fn fence_marker_truncates_lang_at_other_punctuation() {
assert_eq!(fence_marker("```ru/st").as_deref(), Some("ru"));
assert_eq!(fence_marker("```py.thon").as_deref(), Some("py"));
assert_eq!(fence_marker("```rust!").as_deref(), Some("rust"));
assert_eq!(fence_marker("```c++").as_deref(), Some("c"));
}
#[test]
fn fence_marker_truncates_lang_at_non_ascii() {
assert_eq!(fence_marker("```rüst").as_deref(), Some("r"));
assert_eq!(fence_marker("```🦀rust").as_deref(), Some(""));
assert_eq!(fence_marker("```rust🦀").as_deref(), Some("rust"));
}
#[test]
fn render_html_fenced_block_drops_injected_quote_in_lang_tag() {
let input = "```\"x\nbody\n```";
let expected = "<pre>body</pre>";
assert_eq!(render_html(input), expected);
let input = "```rust\"injected\nfn main() {}\n```";
let expected = "<pre><code class=\"language-rust\">fn main() {}</code></pre>";
assert_eq!(render_html(input), expected);
}
#[test]
fn hitl_card_text_format_pins_agent_then_action_then_summary() {
let id: i64 = 42;
let agent = "pm";
let action = "approve";
let summary = "ship the **release**";
let actual = format!(
"🔐 #{id} {}\naction: {}\n{}",
html_escape_str(agent),
html_escape_str(action),
render_html(summary),
);
assert_eq!(
actual,
"🔐 #42 pm\naction: approve\nship the <b>release</b>",
);
let actual_escaped = format!(
"🔐 #{id} {}\naction: {}\n{}",
html_escape_str("ops:<bot>"),
html_escape_str("kill & restart"),
render_html(summary),
);
assert_eq!(
actual_escaped,
"🔐 #42 ops:<bot>\naction: kill & restart\nship the <b>release</b>",
);
}
#[test]
fn render_html_fenced_block_strips_trailing_lang_garbage() {
let input = "```rust // example\nfn main() {}\n```";
let expected = "<pre><code class=\"language-rust\">fn main() {}</code></pre>";
assert_eq!(render_html(input), expected);
}
#[test]
fn render_html_inline_code_is_not_re_parsed() {
assert_eq!(render_html("`**not bold**`"), "<code>**not bold**</code>",);
}
fn decide_sql(conn: &Connection, id: i64, approved: bool) -> bool {
let status = if approved { "approved" } else { "denied" };
let n = conn
.execute(
"UPDATE approvals SET status=?1, decided_at=strftime('%s','now'), decided_by='user:telegram'
WHERE id=?2 AND status='pending'",
params![status, id],
)
.map(|n| n > 0)
.unwrap_or(false);
if n {
let _ = conn.execute(
"UPDATE approvals SET delivered_at=strftime('%s','now')
WHERE id=?1 AND delivered_at IS NULL",
params![id],
);
}
n
}
fn insert_approval(conn: &Connection, status: &str, delivered_at: Option<f64>) -> i64 {
conn.execute(
"INSERT INTO approvals (project_id, agent_id, action, summary, status,
requested_at, expires_at, delivered_at)
VALUES ('p', 'eng_lead', 'publish', 's', ?1, 0.0, 999999999.0, ?2)",
params![status, delivered_at],
)
.unwrap();
conn.last_insert_rowid()
}
#[test]
fn stale_tap_on_undeliverable_does_not_flip_delivered_at() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_approval(&conn, "undeliverable", None);
let decided = decide_sql(&conn, id, true);
assert!(!decided, "stale tap should report no live decision");
let (status, delivered_at): (String, Option<f64>) = conn
.query_row(
"SELECT status, delivered_at FROM approvals WHERE id = ?1",
params![id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(status, "undeliverable");
assert!(
delivered_at.is_none(),
"delivered_at must stay NULL on undeliverable row (invariant)"
);
}
#[test]
fn live_tap_on_pending_flips_status_and_delivered_at() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_approval(&conn, "pending", None);
let decided = decide_sql(&conn, id, true);
assert!(decided, "live tap should report decision");
let (status, delivered_at): (String, Option<f64>) = conn
.query_row(
"SELECT status, delivered_at FROM approvals WHERE id = ?1",
params![id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(status, "approved");
assert!(
delivered_at.is_some(),
"live decision implies delivery acknowledgement"
);
}
#[test]
fn unscoped_bot_routes_every_approval() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert!(should_route(None, "p:dev1", &conn));
assert!(should_route(None, "p:eng_lead", &conn));
assert!(should_route(None, "p:ghost", &conn));
assert!(should_route(None, "other:agent", &conn));
}
#[test]
fn scoped_bot_routes_only_its_managers_chain() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert!(should_route(Some("p:eng_lead"), "p:dev1", &conn));
assert!(should_route(Some("p:eng_lead"), "p:eng_lead", &conn));
assert!(!should_route(Some("p:eng_lead"), "p:pm", &conn));
}
#[test]
fn scoped_bot_with_unknown_agent_falls_back_to_self_routing() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert!(!should_route(Some("p:eng_lead"), "p:ghost", &conn));
}
fn insert_reply(conn: &Connection, sender: &str, text: &str) -> i64 {
let project = sender.split_once(':').map(|(p, _)| p).unwrap_or("p");
conn.execute(
"INSERT INTO messages (project_id, sender, recipient, text, sent_at)
VALUES (?1, ?2, 'user:telegram', ?3, strftime('%s','now'))",
params![project, sender, text],
)
.unwrap();
conn.last_insert_rowid()
}
#[test]
fn reply_routes_only_to_its_senders_bot() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let pm_msg = insert_reply(&conn, "p:pm", "from pm");
let eng_msg = insert_reply(&conn, "p:eng_lead", "from eng");
let mut stmt = conn
.prepare(
"SELECT m.id, m.sender, m.text FROM messages m
WHERE m.id > 0
AND m.recipient = 'user:telegram'
AND m.acked_at IS NULL
AND m.project_id = 'p'
ORDER BY m.id",
)
.unwrap();
let rows: Vec<(i64, String, String)> = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
.unwrap()
.flatten()
.collect();
assert_eq!(rows.len(), 2, "both replies share the project pre-filter");
let pm_routed: Vec<i64> = rows
.iter()
.filter(|(_, sender, _)| should_route(Some("p:pm"), sender, &conn))
.map(|(id, _, _)| *id)
.collect();
assert_eq!(pm_routed, vec![pm_msg]);
let eng_routed: Vec<i64> = rows
.iter()
.filter(|(_, sender, _)| should_route(Some("p:eng_lead"), sender, &conn))
.map(|(id, _, _)| *id)
.collect();
assert_eq!(eng_routed, vec![eng_msg]);
let unscoped: Vec<i64> = rows
.iter()
.filter(|(_, sender, _)| should_route(None, sender, &conn))
.map(|(id, _, _)| *id)
.collect();
assert_eq!(unscoped, vec![pm_msg, eng_msg]);
}
#[test]
fn live_tap_keeps_existing_delivered_at_unchanged() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
let id = insert_approval(&conn, "pending", Some(1234.5));
let decided = decide_sql(&conn, id, false);
assert!(decided);
let delivered_at: f64 = conn
.query_row(
"SELECT delivered_at FROM approvals WHERE id = ?1",
params![id],
|r| r.get(0),
)
.unwrap();
assert!(
(delivered_at - 1234.5).abs() < 1e-6,
"previously-set delivered_at must not be overwritten ({delivered_at})"
);
}
#[test]
fn agent_runtime_returns_runtime_for_known_agent() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert_eq!(
agent_runtime(&conn, "p:eng_lead"),
Some("claude-code".into())
);
}
#[test]
fn agent_runtime_returns_runtime_when_runtime_varies() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
conn.execute(
"INSERT OR IGNORE INTO agents (id, project_id, role, runtime, is_manager, reports_to)
VALUES ('p:codex_mgr','p','codex_mgr','codex',1,NULL)",
[],
)
.unwrap();
assert_eq!(agent_runtime(&conn, "p:codex_mgr"), Some("codex".into()));
}
#[test]
fn agent_runtime_returns_none_for_unknown_agent() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
assert_eq!(agent_runtime(&conn, "p:ghost"), None);
}
#[test]
fn start_greeting_is_one_liner_using_manager_id_fallback() {
let body = start_help_body(false, Some("software-team:director"));
assert_eq!(
body,
"Connected to software-team:director via teamctl. Just type to chat."
);
assert!(
!body.contains("/pending"),
"greeting must not dump commands"
);
assert!(!body.contains("/dm"), "greeting must not dump commands");
}
#[test]
fn start_greeting_prefers_display_name() {
let body = start_help_body(false, Some("Director"));
assert_eq!(
body,
"Connected to Director via teamctl. Just type to chat."
);
}
#[test]
fn start_greeting_unscoped_points_to_help() {
let body = start_help_body(false, None);
assert_eq!(body, "Connected via teamctl. Send /help for commands.");
assert!(!body.contains("/dm"), "greeting must not dump commands");
}
#[test]
fn help_lists_advanced_commands_for_scoped_bot() {
let body = start_help_body(true, Some("Director"));
assert!(body.contains("/pending"), "help must list /pending");
assert!(body.contains("/dm"), "help must list /dm");
assert!(
body.contains("slash-passthrough to Director's tmux session (Claude Code and Codex)"),
"help must list slash-passthrough naming the manager: {body}"
);
assert!(
body.contains("chat with Director"),
"help should still tell the operator how to chat: {body}"
);
}
#[test]
fn help_lists_dm_and_pending_for_unscoped_bot() {
let body = start_help_body(true, None);
assert!(
body.contains("/dm <project>:<agent>"),
"unscoped help must list /dm"
);
assert!(
body.contains("/pending"),
"unscoped help must list /pending"
);
assert!(
!body.contains("slash-passthrough"),
"unscoped bot has no manager tmux session to pass through to: {body}"
);
}
#[test]
fn slash_outcome_passes_through_for_claude_code_runtime() {
let outcome = slash_outcome("writing:manager", "claude-code", "t-");
assert_eq!(
outcome,
SlashOutcome::Passthrough {
session: "t-writing-manager".into(),
}
);
}
#[test]
fn slash_outcome_honours_custom_tmux_prefix() {
let outcome = slash_outcome("news:head_editor", "claude-code", "a-");
assert_eq!(
outcome,
SlashOutcome::Passthrough {
session: "a-news-head_editor".into(),
}
);
}
#[test]
fn slash_outcome_passes_through_for_codex_runtime() {
let outcome = slash_outcome("writing:manager", "codex", "t-");
assert_eq!(
outcome,
SlashOutcome::Passthrough {
session: "t-writing-manager".into(),
}
);
}
#[test]
fn slash_outcome_passes_through_for_opencode_runtime() {
let outcome = slash_outcome("writing:manager", "opencode", "t-");
assert_eq!(
outcome,
SlashOutcome::Passthrough {
session: "t-writing-manager".into(),
}
);
}
#[test]
fn slash_outcome_rejects_gemini_runtime_with_named_runtime() {
let outcome = slash_outcome("writing:manager", "gemini", "t-");
let SlashOutcome::Reject { reason } = outcome else {
panic!("non-CC runtime must reject");
};
assert!(reason.contains("gemini"), "names the runtime: {reason}");
}
#[test]
fn slash_outcome_rejects_malformed_manager_id() {
let outcome = slash_outcome("not-a-manager-id", "claude-code", "t-");
let SlashOutcome::Reject { reason } = outcome else {
panic!("malformed manager id must reject");
};
assert!(reason.contains("malformed"), "names the failure: {reason}");
}
#[test]
fn tmux_type_body_argv_pins_literal_body_shape() {
let argv = tmux_type_body_argv("t-writing-manager", "/clear");
assert_eq!(
argv,
["send-keys", "-t", "t-writing-manager", "-l", "/clear"]
);
}
#[test]
fn tmux_submit_argv_pins_separate_enter_shape() {
assert_eq!(
tmux_submit_argv("t-writing-manager"),
["send-keys", "-t", "t-writing-manager", "Enter"]
);
}
#[test]
fn tmux_type_body_argv_passes_body_verbatim_no_quote_munging() {
let argv = tmux_type_body_argv("sess", "/compact focus on the cascade");
assert_eq!(argv[4], "/compact focus on the cascade");
}
#[test]
fn tmux_type_body_argv_keeps_key_names_literal() {
let argv = tmux_type_body_argv("sess", "/note press Enter then C-c");
assert_eq!(
argv,
[
"send-keys",
"-t",
"sess",
"-l",
"/note press Enter then C-c"
]
);
}
#[test]
fn commands_for_runtime_returns_full_cc_list_for_claude_code() {
let cmds = commands_for_runtime(Some("claude-code"));
assert_eq!(
cmds.len(),
CC_SLASH_COMMANDS.len(),
"CC manager registers the full curated list"
);
let names: Vec<&str> = cmds.iter().map(|c| c.command.as_str()).collect();
assert!(names.contains(&"clear"), "must include /clear: {names:?}");
assert!(
names.contains(&"compact"),
"must include /compact: {names:?}"
);
assert!(names.contains(&"help"), "must include /help: {names:?}");
}
#[test]
fn commands_for_runtime_returns_full_codex_list_for_codex() {
let cmds = commands_for_runtime(Some("codex"));
assert_eq!(
cmds.len(),
CODEX_SLASH_COMMANDS.len(),
"codex manager registers the full curated list"
);
for (cmd, (name, desc)) in cmds.iter().zip(CODEX_SLASH_COMMANDS) {
assert_eq!(cmd.command, *name);
assert_eq!(cmd.description, *desc);
}
}
#[test]
fn commands_for_runtime_returns_full_opencode_list_for_opencode() {
let cmds = commands_for_runtime(Some("opencode"));
assert_eq!(
cmds.len(),
OPENCODE_SLASH_COMMANDS.len(),
"opencode manager registers the full curated list"
);
for (cmd, (name, desc)) in cmds.iter().zip(OPENCODE_SLASH_COMMANDS) {
assert_eq!(cmd.command, *name);
assert_eq!(cmd.description, *desc);
}
}
#[test]
fn commands_for_runtime_returns_empty_for_gemini() {
assert!(commands_for_runtime(Some("gemini")).is_empty());
}
#[test]
fn commands_for_runtime_returns_empty_for_unknown_runtime() {
assert!(commands_for_runtime(Some("a-future-runtime")).is_empty());
}
#[test]
fn commands_for_runtime_returns_empty_for_unscoped_bot() {
assert!(commands_for_runtime(None).is_empty());
}
#[test]
fn cc_slash_command_names_satisfy_telegram_constraints() {
for (cmd, _desc) in CC_SLASH_COMMANDS {
assert!(
!cmd.is_empty() && cmd.len() <= 32,
"command `{cmd}` violates 1-32 char limit"
);
assert!(
cmd.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'),
"command `{cmd}` contains chars Telegram rejects (only [a-z0-9_])"
);
}
}
#[test]
fn cc_slash_command_descriptions_satisfy_telegram_constraints() {
for (cmd, desc) in CC_SLASH_COMMANDS {
assert!(
desc.len() >= 3 && desc.len() <= 256,
"description for `{cmd}` violates 3-256 char limit (got {} chars: {desc:?})",
desc.len()
);
}
}
#[test]
fn codex_slash_command_names_satisfy_telegram_constraints() {
for (cmd, _desc) in CODEX_SLASH_COMMANDS {
assert!(
!cmd.is_empty() && cmd.len() <= 32,
"command `{cmd}` violates 1-32 char limit"
);
assert!(
cmd.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'),
"command `{cmd}` contains chars Telegram rejects (only [a-z0-9_])"
);
}
}
#[test]
fn codex_slash_command_descriptions_satisfy_telegram_constraints() {
for (cmd, desc) in CODEX_SLASH_COMMANDS {
assert!(
desc.len() >= 3 && desc.len() <= 256,
"description for `{cmd}` violates 3-256 char limit (got {} chars: {desc:?})",
desc.len()
);
}
}
#[test]
fn opencode_slash_command_names_satisfy_telegram_constraints() {
for (cmd, _desc) in OPENCODE_SLASH_COMMANDS {
assert!(
!cmd.is_empty() && cmd.len() <= 32,
"command `{cmd}` violates 1-32 char limit"
);
assert!(
cmd.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'),
"command `{cmd}` contains chars Telegram rejects (only [a-z0-9_])"
);
}
}
#[test]
fn opencode_slash_command_descriptions_satisfy_telegram_constraints() {
for (cmd, desc) in OPENCODE_SLASH_COMMANDS {
assert!(
desc.len() >= 3 && desc.len() <= 256,
"description for `{cmd}` violates 3-256 char limit (got {} chars: {desc:?})",
desc.len()
);
}
}
#[test]
fn classify_kind_routes_reaction() {
assert_eq!(classify_kind(Some("reaction")), DispatchKind::Reaction);
}
#[test]
fn classify_kind_unknown_fallback_unchanged_for_other_strings() {
assert_eq!(
classify_kind(Some("reactions")),
DispatchKind::UnknownFallback
);
assert_eq!(classify_kind(Some("react")), DispatchKind::UnknownFallback);
}
#[test]
fn parse_reaction_payload_extracts_telegram_msg_id_and_emoji() {
let p = parse_reaction_payload(r#"{"telegram_msg_id":4242,"emoji":"👀"}"#)
.expect("payload parses");
assert_eq!(p.telegram_msg_id, 4242);
assert_eq!(p.emoji, "👀");
}
#[test]
fn parse_reaction_payload_returns_none_on_missing_fields() {
assert!(parse_reaction_payload("not json").is_none());
assert!(
parse_reaction_payload(r#"{"emoji":"👍"}"#).is_none(),
"missing telegram_msg_id"
);
assert!(
parse_reaction_payload(r#"{"telegram_msg_id":7}"#).is_none(),
"missing emoji"
);
}
#[test]
fn parse_reaction_payload_returns_none_on_wrong_types() {
assert!(
parse_reaction_payload(r#"{"telegram_msg_id":"oops","emoji":"👍"}"#).is_none(),
"string telegram_msg_id"
);
assert!(
parse_reaction_payload(r#"{"telegram_msg_id":7,"emoji":42}"#).is_none(),
"non-string emoji"
);
}
#[test]
fn classify_kind_routes_typing() {
assert_eq!(classify_kind(Some("typing")), DispatchKind::Typing);
}
#[test]
fn classify_kind_unknown_fallback_unchanged_for_typing_lookalikes() {
assert_eq!(classify_kind(Some("type")), DispatchKind::UnknownFallback);
assert_eq!(
classify_kind(Some("typings")),
DispatchKind::UnknownFallback
);
}
#[test]
fn extend_typing_window_inserts_new_entry_with_deadline() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
let now = Instant::now();
let ceiling = Duration::from_secs(10);
let deadline = extend_typing_window(&mut map, ChatId(42), now, ceiling);
assert_eq!(deadline, now + ceiling);
assert_eq!(map.get(&ChatId(42)), Some(&(now + ceiling)));
}
#[test]
fn extend_typing_window_resets_existing_deadline_on_second_call() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
let t0 = Instant::now();
let ceiling = Duration::from_secs(10);
extend_typing_window(&mut map, ChatId(7), t0, ceiling);
let t1 = t0 + Duration::from_secs(3);
let deadline = extend_typing_window(&mut map, ChatId(7), t1, ceiling);
assert_eq!(deadline, t1 + ceiling);
assert_eq!(map.get(&ChatId(7)), Some(&(t1 + ceiling)));
}
#[test]
fn clear_typing_window_removes_present_entry_and_reports_true() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
let now = Instant::now();
extend_typing_window(&mut map, ChatId(1), now, Duration::from_secs(10));
assert!(clear_typing_window(&mut map, ChatId(1)));
assert!(!map.contains_key(&ChatId(1)));
}
#[test]
fn clear_typing_window_returns_false_when_chat_not_tracked() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
assert!(!clear_typing_window(&mut map, ChatId(99)));
}
#[test]
fn refresh_typing_windows_drops_expired_and_returns_active() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
let now = Instant::now();
map.insert(ChatId(1), now + Duration::from_secs(2));
map.insert(ChatId(2), now - Duration::from_millis(10));
let active = refresh_typing_windows(&mut map, now);
assert_eq!(active, vec![ChatId(1)]);
assert!(map.contains_key(&ChatId(1)));
assert!(!map.contains_key(&ChatId(2)));
}
#[test]
fn refresh_typing_windows_returns_empty_on_empty_map() {
let mut map: HashMap<ChatId, Instant> = HashMap::new();
let active = refresh_typing_windows(&mut map, Instant::now());
assert!(active.is_empty());
}
#[test]
fn reply_parameters_for_returns_none_when_telegram_msg_id_is_none() {
assert!(reply_parameters_for(None).is_none());
}
#[test]
fn reply_parameters_for_returns_some_when_telegram_msg_id_is_set() {
let rp = reply_parameters_for(Some(12345)).expect("Some when set");
assert_eq!(rp.message_id, MessageId(12345));
}
#[test]
fn reply_parameters_for_safely_casts_within_i32_range() {
let id: i64 = 2_000_000_000;
let rp = reply_parameters_for(Some(id)).expect("Some when set");
assert_eq!(rp.message_id, MessageId(id as i32));
}
#[test]
fn extension_from_mime_covers_canonical_types() {
assert_eq!(extension_from_mime("image/png"), "png");
assert_eq!(extension_from_mime("image/jpeg"), "jpg");
assert_eq!(extension_from_mime("image/webp"), "webp");
assert_eq!(extension_from_mime("image/gif"), "gif");
assert_eq!(extension_from_mime("application/pdf"), "pdf");
assert_eq!(extension_from_mime("text/plain"), "txt");
assert_eq!(extension_from_mime("application/zip"), "zip");
}
#[test]
fn extension_from_mime_falls_back_to_bin_for_unknown() {
assert_eq!(extension_from_mime("application/octet-stream"), "bin");
assert_eq!(extension_from_mime("video/mp4"), "bin");
assert_eq!(extension_from_mime(""), "bin");
}
#[test]
fn extension_for_document_prefers_filename_extension() {
assert_eq!(
extension_for_document(Some("report.pdf"), "application/octet-stream"),
"pdf"
);
assert_eq!(
extension_for_document(Some("snapshot.PNG"), "application/pdf"),
"png",
"case-folded to lowercase"
);
}
#[test]
fn extension_for_document_falls_back_to_mime_when_filename_missing() {
assert_eq!(extension_for_document(None, "image/png"), "png");
assert_eq!(
extension_for_document(None, "application/octet-stream"),
"bin"
);
}
#[test]
fn extension_for_document_rejects_funky_extensions() {
assert_eq!(extension_for_document(Some("README"), "text/plain"), "txt");
assert_eq!(
extension_for_document(Some("trailing."), "text/plain"),
"txt"
);
assert_eq!(
extension_for_document(Some("name.weird/ext"), "image/png"),
"png"
);
assert_eq!(
extension_for_document(Some("name.thisistoolongatail"), "image/png"),
"png"
);
}
#[test]
fn inbound_media_path_composes_root_project_rowid_extension() {
let root = std::path::Path::new("/srv/.team/state/inbound-media");
let path = inbound_media_path(root, "writing", 42, "jpg");
assert_eq!(
path,
std::path::PathBuf::from("/srv/.team/state/inbound-media/writing/42.jpg")
);
}
#[test]
fn media_success_payload_includes_path_mime_size_and_omits_empty_caption() {
let path = std::path::Path::new("/srv/.team/state/inbound-media/p/7.jpg");
let s = media_success_payload(path, "", "image/jpeg", 1024);
let v: serde_json::Value = serde_json::from_str(&s).unwrap();
assert_eq!(v["path"], "/srv/.team/state/inbound-media/p/7.jpg");
assert_eq!(v["mime"], "image/jpeg");
assert_eq!(v["size_bytes"], 1024);
assert!(
v.get("caption").is_none(),
"empty caption omitted from payload"
);
}
#[test]
fn media_success_payload_includes_caption_when_present() {
let path = std::path::Path::new("/x.png");
let s = media_success_payload(path, "look at this", "image/png", 32);
let v: serde_json::Value = serde_json::from_str(&s).unwrap();
assert_eq!(v["caption"], "look at this");
}
#[test]
fn media_error_payload_carries_verbose_error_and_optional_caption() {
let s = media_error_payload("", "get_file: timed out");
let v: serde_json::Value = serde_json::from_str(&s).unwrap();
assert_eq!(v["error"], "get_file: timed out");
assert!(v.get("caption").is_none());
let s = media_error_payload("a screenshot", "create file: permission denied");
let v: serde_json::Value = serde_json::from_str(&s).unwrap();
assert_eq!(v["error"], "create file: permission denied");
assert_eq!(v["caption"], "a screenshot");
}
#[test]
fn placeholder_then_success_update_round_trip() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
conn.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, kind, structured_payload)
VALUES ('p', 'user:telegram', 'p:eng_lead', 'cap',
strftime('%s','now'), 'media_pending', '{}')",
[],
)
.unwrap();
let id = conn.last_insert_rowid();
let payload =
media_success_payload(std::path::Path::new("/x/p/3.jpg"), "cap", "image/jpeg", 128);
conn.execute(
"UPDATE messages SET kind = ?1, structured_payload = ?2 WHERE id = ?3",
params!["image", payload, id],
)
.unwrap();
let (kind, sp): (Option<String>, Option<String>) = conn
.query_row(
"SELECT kind, structured_payload FROM messages WHERE id = ?1",
params![id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(kind.as_deref(), Some("image"));
let v: serde_json::Value = serde_json::from_str(sp.as_deref().unwrap()).unwrap();
assert_eq!(v["mime"], "image/jpeg");
}
#[test]
fn placeholder_then_error_update_writes_media_error_kind() {
let conn = Connection::open_in_memory().unwrap();
seed(&conn);
conn.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, kind, structured_payload)
VALUES ('p', 'user:telegram', 'p:eng_lead', '',
strftime('%s','now'), 'media_pending', '{}')",
[],
)
.unwrap();
let id = conn.last_insert_rowid();
let payload = media_error_payload("", "download_file: 502 bad gateway");
conn.execute(
"UPDATE messages SET kind = 'media_error', structured_payload = ?1 WHERE id = ?2",
params![payload, id],
)
.unwrap();
let (kind, sp): (Option<String>, Option<String>) = conn
.query_row(
"SELECT kind, structured_payload FROM messages WHERE id = ?1",
params![id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(kind.as_deref(), Some("media_error"));
assert!(sp.unwrap().contains("502 bad gateway"));
}
fn seed_media_pending(conn: &Connection) -> i64 {
seed(conn);
conn.execute(
"INSERT INTO messages
(project_id, sender, recipient, text, sent_at, kind, structured_payload)
VALUES ('p', 'user:telegram', 'p:eng_lead', 'cap',
strftime('%s','now'), 'media_pending', '{}')",
[],
)
.unwrap();
conn.last_insert_rowid()
}
fn current_kind(conn: &Connection, id: i64) -> Option<String> {
conn.query_row(
"SELECT kind FROM messages WHERE id = ?1",
params![id],
|r| r.get(0),
)
.unwrap()
}
#[test]
fn update_message_kind_refuses_system_and_leaves_row_untouched() {
let conn = Connection::open_in_memory().unwrap();
let id = seed_media_pending(&conn);
let result = update_message_kind(&conn, id, "system", "{}");
assert!(result.is_err(), "setting privileged kind must be refused");
assert_eq!(
current_kind(&conn, id).as_deref(),
Some("media_pending"),
"refused UPDATE must not mutate the row"
);
}
#[test]
fn update_message_kind_writes_legitimate_media_kinds() {
let success = media_success_payload(
std::path::Path::new("/srv/.team/state/inbound-media/p/3.jpg"),
"cap",
"image/jpeg",
128,
);
let error = media_error_payload("cap", "download_file: 502 bad gateway");
let cases: [(&str, &str, &str); 3] = [
("image", success.as_str(), "image/jpeg"),
("file", success.as_str(), "image/jpeg"),
("media_error", error.as_str(), "502 bad gateway"),
];
for (kind, payload, marker) in cases {
let conn = Connection::open_in_memory().unwrap();
let id = seed_media_pending(&conn);
let changed = update_message_kind(&conn, id, kind, payload)
.unwrap_or_else(|e| panic!("kind {kind:?} should be allowed: {e}"));
assert_eq!(changed, 1, "kind {kind:?} should update exactly one row");
let (stored_kind, sp): (Option<String>, Option<String>) = conn
.query_row(
"SELECT kind, structured_payload FROM messages WHERE id = ?1",
params![id],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(stored_kind.as_deref(), Some(kind));
assert!(
sp.as_deref().unwrap().contains(marker),
"kind {kind:?} payload should carry {marker:?}, got {sp:?}"
);
}
}
#[test]
fn map_voice_outcome_ok_yields_quoted_reply_and_prefixed_inbox_row() {
let d = map_voice_outcome(&SttOutcome::Ok("hello team".into()));
assert_eq!(d.user_reply, "🎙 \"hello team\"");
assert_eq!(
d.inbox_text.as_deref(),
Some("🎙 (transcribed voice, may have misspellings): hello team")
);
assert!(d
.inbox_text
.as_deref()
.unwrap()
.starts_with(VOICE_INBOX_PREFIX));
}
#[test]
fn map_voice_outcome_skipped_yields_no_inbox_row() {
let d = map_voice_outcome(&SttOutcome::Skipped);
assert!(d.inbox_text.is_none());
assert!(d.user_reply.contains("couldn't capture anything"));
assert!(!d.user_reply.contains("failed"));
}
#[test]
fn map_voice_outcome_failed_yields_no_inbox_row_and_surfaces_error() {
let d = map_voice_outcome(&SttOutcome::Failed("network down".into()));
assert!(d.inbox_text.is_none());
assert!(d.user_reply.contains("failed"));
assert!(d.user_reply.contains("network down"));
assert!(!d.user_reply.contains("couldn't capture"));
}
#[test]
fn voice_echoes_routing_only_on_a_successful_insert() {
let ok = map_voice_outcome(&SttOutcome::Ok("hello team".into()));
assert!(
voice_should_echo_routing(&ok, true),
"success path must confirm routing"
);
assert!(
!voice_should_echo_routing(&ok, false),
"a dropped INSERT must not claim routing"
);
let failed = map_voice_outcome(&SttOutcome::Failed("network down".into()));
assert!(
!voice_should_echo_routing(&failed, true),
"a transcription failure must not confirm routing"
);
let skipped = map_voice_outcome(&SttOutcome::Skipped);
assert!(
!voice_should_echo_routing(&skipped, true),
"an empty/skipped capture must not confirm routing"
);
}
#[test]
fn voice_inbox_prefix_matches_issue_spec() {
assert_eq!(
VOICE_INBOX_PREFIX,
"🎙 (transcribed voice, may have misspellings):"
);
}
#[test]
fn voice_stt_missing_reply_carries_operator_actionable_hints() {
let body = voice_stt_missing_reply();
assert!(
body.starts_with("🎙"),
"reply must lead with the voice glyph so the operator sees \
this is about their voice message: {body}"
);
assert!(
body.contains("Voice isn't configured"),
"reply must name the cause so the operator knows what to fix: {body}"
);
assert!(
body.contains("/teamctl:adjust"),
"reply must surface the conversational fix path: {body}"
);
assert!(
body.contains("interfaces.telegram.speech_to_text"),
"reply must surface the YAML key for the manual fix path: {body}"
);
assert!(
body.contains("https://teamctl.run/"),
"reply must include a docs pointer the operator can open: {body}"
);
}
#[test]
fn voice_size_pre_reject_names_both_numbers() {
let msg = voice_size_pre_reject(5_000_000, MAX_VOICE_BYTES);
assert!(msg.contains("5000000"), "msg names reported: {msg}");
assert!(
msg.contains(&MAX_VOICE_BYTES.to_string()),
"msg names max: {msg}"
);
assert!(msg.contains("too large"), "msg flags the rejection: {msg}");
}
#[test]
fn voice_path_mid_reject_names_both_numbers_and_distinguishes() {
let msg = media_size_mid_reject("voice file", 1_000_000, 1_500_000, MAX_VOICE_BYTES);
assert!(msg.contains("voice file"), "msg carries voice kind: {msg}");
assert!(msg.contains("2500000"), "msg names cumulative: {msg}");
assert!(
msg.contains(&MAX_VOICE_BYTES.to_string()),
"msg names max: {msg}"
);
assert!(
msg.contains("mid-download"),
"msg distinguishes from pre-reject: {msg}"
);
}
#[tokio::test]
async fn bounded_writer_passes_through_when_under_max() {
use tokio::io::AsyncWriteExt;
let mut bw = BoundedWriter::new("voice file", Vec::<u8>::new(), 16);
bw.write_all(b"hello").await.unwrap();
bw.write_all(b" world").await.unwrap();
bw.flush().await.unwrap();
assert_eq!(bw.into_inner(), b"hello world");
}
#[tokio::test]
async fn bounded_writer_allows_exactly_max() {
use tokio::io::AsyncWriteExt;
let mut bw = BoundedWriter::new("voice file", Vec::<u8>::new(), 5);
bw.write_all(b"hello").await.unwrap();
assert_eq!(bw.into_inner(), b"hello");
}
#[tokio::test]
async fn bounded_writer_errors_when_chunk_would_exceed_max() {
use tokio::io::AsyncWriteExt;
let mut bw = BoundedWriter::new("voice file", Vec::<u8>::new(), 4);
bw.write_all(b"abcd").await.unwrap();
let err = bw.write_all(b"e").await.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(
err.to_string().contains("mid-download"),
"error must use media_size_mid_reject format: {err}"
);
assert_eq!(
bw.into_inner().as_slice(),
b"abcd",
"rejected chunk must not append"
);
}
#[tokio::test]
async fn bounded_writer_errors_on_single_oversize_chunk() {
use tokio::io::AsyncWriteExt;
let mut bw = BoundedWriter::new("voice file", Vec::<u8>::new(), 4);
let err = bw.write_all(b"abcde").await.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains("mid-download"));
assert!(
bw.into_inner().is_empty(),
"no bytes must land when the chunk overshoots"
);
}
#[test]
fn media_size_pre_reject_carries_kind_and_names_both_numbers() {
let msg = media_size_pre_reject("media file", 60_000_000, MAX_DOWNLOAD_BYTES);
assert!(msg.contains("media file"), "msg carries kind: {msg}");
assert!(msg.contains("60000000"), "msg names reported: {msg}");
assert!(
msg.contains(&MAX_DOWNLOAD_BYTES.to_string()),
"msg names max: {msg}"
);
assert!(msg.contains("too large"), "msg flags the rejection: {msg}");
}
#[test]
fn media_size_mid_reject_carries_kind_and_distinguishes() {
let msg = media_size_mid_reject("media file", 40_000_000, 20_000_000, MAX_DOWNLOAD_BYTES);
assert!(msg.contains("media file"), "msg carries kind: {msg}");
assert!(msg.contains("60000000"), "msg names cumulative: {msg}");
assert!(
msg.contains(&MAX_DOWNLOAD_BYTES.to_string()),
"msg names max: {msg}"
);
assert!(
msg.contains("mid-download"),
"msg distinguishes from pre-reject: {msg}"
);
}
#[tokio::test]
async fn bounded_writer_aborts_mid_stream_on_tokio_fs_file_without_excess_disk_bytes() {
use tokio::io::AsyncWriteExt;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("media");
let f = tokio::fs::File::create(&path).await.unwrap();
let mut bw = BoundedWriter::new("media file", f, 4);
bw.write_all(b"abcd").await.unwrap();
let err = bw.write_all(b"e").await.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(
err.to_string().contains("media file"),
"mid-reject must carry the `media file` kind: {err}"
);
let mut f = bw.into_inner();
f.flush().await.ok();
drop(f);
let bytes = tokio::fs::read(&path).await.unwrap();
assert_eq!(
bytes, b"abcd",
"rejected chunk must not have hit disk — bounded handle saw \
the over-cap chunk and aborted before `inner.poll_write`"
);
}
#[test]
#[allow(clippy::assertions_on_constants)]
fn max_download_bytes_strictly_exceeds_max_voice_bytes() {
assert!(
MAX_DOWNLOAD_BYTES > MAX_VOICE_BYTES,
"MAX_DOWNLOAD_BYTES ({MAX_DOWNLOAD_BYTES}) must exceed \
MAX_VOICE_BYTES ({MAX_VOICE_BYTES})"
);
}
#[test]
fn rich_message_body_has_chat_and_markdown() {
assert_eq!(
rich_message_body(123, "# Hi"),
serde_json::json!({
"chat_id": 123,
"rich_message": { "markdown": "# Hi" },
}),
);
}
#[test]
fn rich_message_body_preserves_multiline_markdown() {
let markdown = "# Heading\n\n| a | b |\n- list\n```\ncode\n```\n> quote";
let body = rich_message_body(7, markdown);
assert_eq!(body["rich_message"]["markdown"], markdown);
}
#[test]
fn markdown_escape_str_escapes_underscores_and_specials() {
assert_eq!(markdown_escape_str("team_a:bot_1"), "team\\_a:bot\\_1");
assert_eq!(markdown_escape_str("teamctl:kian"), "teamctl:kian");
}
#[test]
fn use_rich_path_three_way_precedence() {
assert!(use_rich_path(true, false));
assert!(!use_rich_path(true, true));
assert!(!use_rich_path(false, false));
assert!(!use_rich_path(false, true));
}
fn parse_cli(extra: &[&str]) -> Result<Cli, clap::Error> {
let mut argv = vec![
"bot",
"--mailbox",
"/tmp/teamctl-test-mailbox.db",
"--token",
"x",
];
argv.extend_from_slice(extra);
Cli::try_parse_from(argv)
}
#[test]
fn cli_rich_messages_default_on() {
let cli = parse_cli(&[]).expect("parse");
assert!(
!cli.no_rich_messages,
"rich must be ON by default (opt-out)"
);
}
#[test]
fn cli_no_rich_messages_opt_out() {
let cli = parse_cli(&["--no-rich-messages"]).expect("parse");
assert!(cli.no_rich_messages, "--no-rich-messages must force HTML");
}
#[test]
fn cli_no_rich_messages_accepts_boolish_values() {
let parse = |v: &str| {
let arg = format!("--no-rich-messages={v}");
parse_cli(&[arg.as_str()])
.unwrap_or_else(|_| panic!("--no-rich-messages={v} must parse"))
.no_rich_messages
};
assert!(
parse("true"),
"=true must disable rich (the documented form)"
);
assert!(parse("1"), "=1 must disable rich (forgiving boolish form)");
assert!(!parse("0"), "=0 must keep rich on");
assert!(!parse("false"), "=false must keep rich on");
}
}