gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-25 14:53:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix Telegram auto-verify flow and routing (#1273)

Browse Source 
* Fix Telegram auto-verify flow and routing

* Fix CI formatting and clippy follow-ups

* Simplify Telegram waiting state update

* Fix notification fallback scopes

* Fix message metadata routing and zh-CN copy
This commit is contained in:
Henry Park
2026-03-16 20:19:43 -07:00
committed by GitHub
parent d0cb5f0ac5
commit 4675e9618c
18 changed files with 1045 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FEATURE_PARITY.md
+1 -1
View File
@@ -68,7 +68,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O
| REPL (simple) | ✅ | ✅ | - | For testing |
| WASM channels | ❌ | ✅ | - | IronClaw innovation; host resolves owner scope vs sender identity |
| WhatsApp | ✅ | ❌ | P1 | Baileys (Web), same-phone mode with echo detection |
| Telegram | ✅ | ✅ | - | WASM channel(MTProto), DM pairing, caption, /start, bot_username, DM topics, setup-time owner verification, owner-scoped persistence |
| Telegram | ✅ | ✅ | - | WASM channel(MTProto), DM pairing, caption, /start, bot_username, DM topics, setup-time owner auto-verification, owner-scoped persistence |
| Discord | ✅ | ❌ | P2 | discord.js, thread parent binding inheritance |
| Signal | ✅ | ✅ | P2 | signal-cli daemonPC, SSE listener HTTP/JSON-R, user/group allowlists, DM pairing |
| Slack | ✅ | ✅ | - | WASM tool |
src/agent/agent_loop.rs
+111 -27
View File
@@ -54,16 +54,65 @@ pub(crate) fn truncate_for_preview(output: &str, max_chars: usize) -> String {
}
}
#[cfg(test)]
fn resolve_routine_notification_user(metadata: &serde_json::Value) -> Option<String> {
metadata
.get("notify_user")
.and_then(|value| value.as_str())
.or_else(|| metadata.get("owner_id").and_then(|value| value.as_str()))
resolve_owner_scope_notification_user(
metadata.get("notify_user").and_then(|value| value.as_str()),
metadata.get("owner_id").and_then(|value| value.as_str()),
)
}
fn trimmed_option(value: Option<&str>) -> Option<String> {
value
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn resolve_owner_scope_notification_user(
explicit_user: Option<&str>,
owner_fallback: Option<&str>,
) -> Option<String> {
trimmed_option(explicit_user).or_else(|| trimmed_option(owner_fallback))
}
async fn resolve_channel_notification_user(
extension_manager: Option<&Arc<ExtensionManager>>,
channel: Option<&str>,
explicit_user: Option<&str>,
owner_fallback: Option<&str>,
) -> Option<String> {
if let Some(user) = trimmed_option(explicit_user) {
return Some(user);
}
if let Some(channel_name) = trimmed_option(channel)
&& let Some(extension_manager) = extension_manager
&& let Some(target) = extension_manager
.notification_target_for_channel(&channel_name)
.await
{
return Some(target);
}
resolve_owner_scope_notification_user(explicit_user, owner_fallback)
}
async fn resolve_routine_notification_target(
extension_manager: Option<&Arc<ExtensionManager>>,
metadata: &serde_json::Value,
) -> Option<String> {
resolve_channel_notification_user(
extension_manager,
metadata
.get("notify_channel")
.and_then(|value| value.as_str()),
metadata.get("notify_user").and_then(|value| value.as_str()),
metadata.get("owner_id").and_then(|value| value.as_str()),
)
.await
}
fn should_fallback_routine_notification(error: &ChannelError) -> bool {
!matches!(error, ChannelError::MissingRoutingTarget { .. })
}
@@ -395,11 +444,13 @@ impl Agent {
.timezone
.clone()
.or_else(|| Some(self.config.default_timezone.clone()));
if let Some(channel) = &hb_config.notify_channel {
let user = hb_config
.notify_user
.clone()
.unwrap_or_else(|| self.owner_id().to_string());
let heartbeat_notify_user = resolve_owner_scope_notification_user(
hb_config.notify_user.as_deref(),
Some(self.owner_id()),
);
if let Some(channel) = &hb_config.notify_channel
&& let Some(user) = heartbeat_notify_user.as_deref()
{
config = config.with_notify(user, channel);
}
@@ -409,26 +460,32 @@ impl Agent {
// Spawn notification forwarder that routes through channel manager
let notify_channel = hb_config.notify_channel.clone();
let notify_user = hb_config
.notify_user
.clone()
.unwrap_or_else(|| self.owner_id().to_string());
let notify_target = resolve_channel_notification_user(
self.deps.extension_manager.as_ref(),
hb_config.notify_channel.as_deref(),
hb_config.notify_user.as_deref(),
Some(self.owner_id()),
)
.await;
let notify_user = heartbeat_notify_user;
let channels = self.channels.clone();
tokio::spawn(async move {
while let Some(response) = notify_rx.recv().await {
// Try the configured channel first, fall back to
// broadcasting on all channels.
let targeted_ok = if let Some(ref channel) = notify_channel {
let targeted_ok = if let Some(ref channel) = notify_channel
&& let Some(ref user) = notify_target
{
channels
.broadcast(channel, &notify_user, response.clone())
.broadcast(channel, user, response.clone())
.await
.is_ok()
} else {
false
};
if !targeted_ok {
let results = channels.broadcast_all(&notify_user, response).await;
if !targeted_ok && let Some(ref user) = notify_user {
let results = channels.broadcast_all(user, response).await;
for (ch, result) in results {
if let Err(e) = result {
tracing::warn!(
@@ -496,6 +553,7 @@ impl Agent {
// Spawn notification forwarder (mirrors heartbeat pattern)
let channels = self.channels.clone();
let extension_manager = self.deps.extension_manager.clone();
tokio::spawn(async move {
while let Some(response) = notify_rx.recv().await {
let notify_channel = response
@@ -503,7 +561,18 @@ impl Agent {
.get("notify_channel")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let Some(user) = resolve_routine_notification_user(&response.metadata)
let fallback_user = resolve_owner_scope_notification_user(
response
.metadata
.get("notify_user")
.and_then(|v| v.as_str()),
response.metadata.get("owner_id").and_then(|v| v.as_str()),
);
let Some(user) = resolve_routine_notification_target(
extension_manager.as_ref(),
&response.metadata,
)
.await
else {
tracing::warn!(
notify_channel = ?notify_channel,
@@ -537,7 +606,7 @@ impl Agent {
false
};
if !targeted_ok {
if !targeted_ok && let Some(user) = fallback_user {
let results = channels.broadcast_all(&user, response).await;
for (ch, result) in results {
if let Err(e) = result {
@@ -624,6 +693,29 @@ impl Agent {
// Store successfully extracted document text in workspace for indexing
self.store_extracted_documents(&message).await;
// Event-triggered routines consume plain user input before it enters
// the normal chat/tool pipeline. This avoids a duplicate turn where
// the main agent responds and the routine also fires on the same
// inbound message.
if !message.is_internal
&& matches!(
SubmissionParser::parse(&message.content),
Submission::UserInput { .. }
)
&& let Some(ref engine) = routine_engine_for_loop
{
let fired = engine.check_event_triggers(&message).await;
if fired > 0 {
tracing::debug!(
channel = %message.channel,
user = %message.user_id,
fired,
"Consumed inbound user message with matching event-triggered routine(s)"
);
continue;
}
}
match self.handle_message(&message).await {
Ok(Some(response)) if !response.is_empty() => {
// Hook: BeforeOutbound — allow hooks to modify or suppress outbound
@@ -696,14 +788,6 @@ impl Agent {
}
}
}
// Check event triggers (cheap in-memory regex, fires async if matched)
if let Some(ref engine) = routine_engine_for_loop {
let fired = engine.check_event_triggers(&message).await;
if fired > 0 {
tracing::debug!("Fired {} event-triggered routines", fired);
}
}
}
// Cleanup
src/channels/web/server.rs
+87 -8
View File
@@ -2243,14 +2243,7 @@ async fn extensions_setup_submit_handler(
resp.auth_url = result.auth_url.clone();
resp.verification = result.verification.clone();
resp.instructions = result.verification.as_ref().map(|v| v.instructions.clone());
if result.verification.is_some() {
state.sse.broadcast(SseEvent::AuthRequired {
extension_name: name.clone(),
instructions: resp.instructions.clone(),
auth_url: None,
setup_url: None,
});
} else {
if result.verification.is_none() {
// Broadcast auth_completed so the chat UI can dismiss any in-progress
// auth card or setup modal that was triggered by tool_auth/tool_activate.
state.sse.broadcast(SseEvent::AuthCompleted {
@@ -2981,6 +2974,92 @@ mod tests {
);
}
#[tokio::test]
async fn test_extensions_setup_submit_telegram_verification_does_not_broadcast_auth_required() {
use axum::body::Body;
use tokio::time::{Duration, timeout};
use tower::ServiceExt;
let secrets = test_secrets_store();
let (ext_mgr, _wasm_tools_dir, wasm_channels_dir) = test_ext_mgr(secrets);
std::fs::write(
wasm_channels_dir.path().join("telegram.wasm"),
b"\0asm fake",
)
.expect("write fake telegram wasm");
let caps = serde_json::json!({
"type": "channel",
"name": "telegram",
"setup": {
"required_secrets": [
{
"name": "telegram_bot_token",
"prompt": "Enter your Telegram Bot API token (from @BotFather)"
}
]
}
});
std::fs::write(
wasm_channels_dir.path().join("telegram.capabilities.json"),
serde_json::to_string(&caps).expect("serialize telegram caps"),
)
.expect("write telegram caps");
ext_mgr
.set_test_telegram_pending_verification("iclaw-7qk2m9", Some("test_hot_bot"))
.await;
let state = test_gateway_state(Some(ext_mgr));
let mut receiver = state.sse.sender().subscribe();
let app = Router::new()
.route(
"/api/extensions/{name}/setup",
post(extensions_setup_submit_handler),
)
.with_state(state);
let req_body = serde_json::json!({
"secrets": {
"telegram_bot_token": "123456789:ABCdefGhI"
}
});
let req = axum::http::Request::builder()
.method("POST")
.uri("/api/extensions/telegram/setup")
.header("content-type", "application/json")
.body(Body::from(req_body.to_string()))
.expect("request");
let resp = ServiceExt::<axum::http::Request<Body>>::oneshot(app, req)
.await
.expect("response");
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), 1024 * 64)
.await
.expect("body");
let parsed: serde_json::Value = serde_json::from_slice(&body).expect("json response");
assert_eq!(parsed["success"], serde_json::Value::Bool(true));
assert_eq!(parsed["activated"], serde_json::Value::Bool(false));
assert_eq!(parsed["verification"]["code"], "iclaw-7qk2m9");
let deadline = tokio::time::Instant::now() + Duration::from_millis(100);
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
break;
}
match timeout(remaining, receiver.recv()).await {
Ok(Ok(crate::channels::web::types::SseEvent::AuthRequired { .. })) => {
panic!("verification responses should not emit auth_required SSE events")
}
Ok(Ok(_)) => continue,
Ok(Err(_)) | Err(_) => break,
}
}
}
fn expired_flow_created_at() -> Option<std::time::Instant> {
std::time::Instant::now()
.checked_sub(oauth_defaults::OAUTH_FLOW_EXPIRY + std::time::Duration::from_secs(1))
src/channels/web/static/app.js
+121 -32
View File
@@ -527,7 +527,6 @@ function enableChatInput() {
const btn = document.getElementById('send-btn');
if (input) {
input.disabled = false;
input.placeholder = I18n.t('chat.inputPlaceholder');
}
if (btn) btn.disabled = false;
}
@@ -1205,11 +1204,13 @@ function showJobCard(data) {
// --- Auth card ---
function handleAuthRequired(data) {
setAuthFlowPending(true, data.instructions);
if (data.auth_url) {
setAuthFlowPending(true, data.instructions);
// OAuth flow: show the global auth prompt with an OAuth button + optional token paste field.
showAuthCard(data);
} else {
if (getConfigureOverlay(data.extension_name)) return;
setAuthFlowPending(true, data.instructions);
// Setup flow: fetch the extension's credential schema and show the multi-field
// configure modal (the same UI used by the Extensions tab "Setup" button).
showConfigureModal(data.extension_name);
@@ -1433,13 +1434,11 @@ function setAuthFlowPending(pending, instructions) {
if (authFlowPending) {
input.disabled = true;
btn.disabled = true;
input.placeholder = instructions || 'Complete extension auth to continue chatting';
return;
}
if (!currentThreadIsReadOnly) {
input.disabled = false;
btn.disabled = false;
input.placeholder = I18n.t('chat.inputPlaceholder');
}
}
@@ -2712,8 +2711,11 @@ function renderConfigureModal(name, secrets) {
const overlay = document.createElement('div');
overlay.className = 'configure-overlay';
overlay.setAttribute('data-extension-name', name);
overlay.dataset.telegramVerificationState = 'idle';
overlay.addEventListener('click', (e) => {
if (e.target === overlay) closeConfigureModal();
if (e.target !== overlay) return;
if (name === 'telegram' && overlay.dataset.telegramVerificationState === 'waiting') return;
closeConfigureModal();
});
const modal = document.createElement('div');
@@ -2737,6 +2739,7 @@ function renderConfigureModal(name, secrets) {
for (const secret of secrets) {
const field = document.createElement('div');
field.className = 'configure-field';
field.dataset.secretName = secret.name;
const label = document.createElement('label');
label.textContent = secret.prompt;
@@ -2781,6 +2784,16 @@ function renderConfigureModal(name, secrets) {
modal.appendChild(form);
const error = document.createElement('div');
error.className = 'configure-inline-error';
error.style.display = 'none';
modal.appendChild(error);
const status = document.createElement('div');
status.className = 'configure-inline-status';
status.style.display = 'none';
modal.appendChild(status);
const actions = document.createElement('div');
actions.className = 'configure-actions';
@@ -2807,12 +2820,20 @@ function renderTelegramVerificationChallenge(overlay, verification) {
if (!overlay || !verification) return;
const modal = overlay.querySelector('.configure-modal');
if (!modal) return;
const telegramField = modal.querySelector('.configure-field[data-secret-name="telegram_bot_token"]');
let panel = modal.querySelector('.configure-verification');
if (!panel) {
panel = document.createElement('div');
panel.className = 'configure-verification';
modal.insertBefore(panel, modal.querySelector('.configure-actions'));
}
if (telegramField && telegramField.parentNode) {
telegramField.insertAdjacentElement('afterend', panel);
} else {
modal.insertBefore(
panel,
modal.querySelector('.configure-inline-error') || modal.querySelector('.configure-actions')
);
}
panel.innerHTML = '';
@@ -2827,10 +2848,15 @@ function renderTelegramVerificationChallenge(overlay, verification) {
instructions.textContent = verification.instructions;
panel.appendChild(instructions);
const code = document.createElement('code');
code.className = 'configure-verification-code';
code.textContent = verification.code;
panel.appendChild(code);
const commandLabel = document.createElement('div');
commandLabel.className = 'configure-verification-instructions';
commandLabel.textContent = I18n.t('config.telegramCommandLabel');
panel.appendChild(commandLabel);
const command = document.createElement('code');
command.className = 'configure-verification-code';
command.textContent = '/start ' + verification.code;
panel.appendChild(command);
if (verification.deep_link) {
const link = document.createElement('a');
@@ -2843,7 +2869,57 @@ function renderTelegramVerificationChallenge(overlay, verification) {
}
}
function submitConfigureModal(name, fields) {
function getConfigurePrimaryButton(overlay) {
return overlay && overlay.querySelector('.configure-actions button.btn-ext.activate');
}
function getConfigureCancelButton(overlay) {
return overlay && overlay.querySelector('.configure-actions button.btn-ext.remove');
}
function setConfigureInlineError(overlay, message) {
const error = overlay && overlay.querySelector('.configure-inline-error');
if (!error) return;
error.textContent = message || '';
error.style.display = message ? 'block' : 'none';
}
function clearConfigureInlineError(overlay) {
setConfigureInlineError(overlay, '');
}
function setConfigureInlineStatus(overlay, message) {
const status = overlay && overlay.querySelector('.configure-inline-status');
if (!status) return;
status.textContent = message || '';
status.style.display = message ? 'block' : 'none';
}
function setTelegramConfigureState(overlay, fields, state) {
if (!overlay) return;
overlay.dataset.telegramVerificationState = state;
const primaryBtn = getConfigurePrimaryButton(overlay);
const cancelBtn = getConfigureCancelButton(overlay);
const waiting = state === 'waiting';
const retry = state === 'retry';
setConfigureInlineStatus(overlay, waiting ? I18n.t('config.telegramOwnerWaiting') : '');
if (primaryBtn) {
primaryBtn.style.display = waiting ? 'none' : '';
primaryBtn.disabled = false;
primaryBtn.textContent = retry ? I18n.t('config.telegramStartOver') : I18n.t('config.save');
}
if (cancelBtn) cancelBtn.disabled = waiting;
}
function startTelegramAutoVerify(name, fields) {
window.setTimeout(() => submitConfigureModal(name, fields, { telegramAutoVerify: true }), 0);
}
function submitConfigureModal(name, fields, options) {
options = options || {};
const secrets = {};
for (const f of fields) {
if (f.input.value.trim()) {
@@ -2851,13 +2927,15 @@ function submitConfigureModal(name, fields) {
}
}
// Disable buttons to prevent double-submit
const overlay = getConfigureOverlay(name) || document.querySelector('.configure-overlay');
const isTelegram = name === 'telegram';
clearConfigureInlineError(overlay);
// Disable buttons to prevent double-submit
var btns = overlay ? overlay.querySelectorAll('.configure-actions button') : [];
btns.forEach(function(b) { b.disabled = true; });
if (overlay && name === 'telegram') {
const submitBtn = overlay.querySelector('.configure-actions button.btn-ext.activate');
if (submitBtn) submitBtn.textContent = I18n.t('config.telegramOwnerWaiting');
if (overlay && isTelegram) {
setTelegramConfigureState(overlay, fields, 'waiting');
}
apiFetch('/api/extensions/' + encodeURIComponent(name) + '/setup', {
@@ -2866,13 +2944,20 @@ function submitConfigureModal(name, fields) {
})
.then((res) => {
if (res.success) {
if (res.verification && name === 'telegram') {
btns.forEach(function(b) { b.disabled = false; });
if (res.verification && isTelegram) {
renderTelegramVerificationChallenge(overlay, res.verification);
fields.forEach(function(f) { f.input.value = ''; });
const submitBtn = overlay.querySelector('.configure-actions button.btn-ext.activate');
if (submitBtn) submitBtn.textContent = I18n.t('config.telegramVerifyOwner');
showToast(res.message || res.verification.instructions, 'info');
setTelegramConfigureState(overlay, fields, 'waiting');
// Once the verification challenge is rendered inline, the global auth lock
// should not keep the chat composer disabled for this setup-driven flow.
setAuthFlowPending(false);
enableChatInput();
if (!options.telegramAutoVerify) {
startTelegramAutoVerify(name, fields);
return;
}
setTelegramConfigureState(overlay, fields, 'retry');
setConfigureInlineError(overlay, I18n.t('config.telegramStartOverHint'));
return;
}
@@ -2891,13 +2976,13 @@ function submitConfigureModal(name, fields) {
} else {
// Keep modal open so the user can correct their input and retry.
btns.forEach(function(b) { b.disabled = false; });
if (name === 'telegram') {
const submitBtn = overlay && overlay.querySelector('.configure-actions button.btn-ext.activate');
setConfigureInlineError(overlay, res.message || 'Configuration failed');
if (isTelegram) {
const hasVerification = overlay && overlay.querySelector('.configure-verification');
if (submitBtn) {
submitBtn.textContent = hasVerification
? I18n.t('config.telegramVerifyOwner')
: I18n.t('config.save');
if (options.telegramAutoVerify || hasVerification) {
setTelegramConfigureState(overlay, fields, 'retry');
} else {
setTelegramConfigureState(overlay, fields, 'idle');
}
}
showToast(res.message || 'Configuration failed', 'error');
@@ -2905,13 +2990,13 @@ function submitConfigureModal(name, fields) {
})
.catch((err) => {
btns.forEach(function(b) { b.disabled = false; });
if (name === 'telegram') {
const submitBtn = overlay && overlay.querySelector('.configure-actions button.btn-ext.activate');
setConfigureInlineError(overlay, 'Configuration failed: ' + err.message);
if (isTelegram) {
const hasVerification = overlay && overlay.querySelector('.configure-verification');
if (submitBtn) {
submitBtn.textContent = hasVerification
? I18n.t('config.telegramVerifyOwner')
: I18n.t('config.save');
if (options.telegramAutoVerify || hasVerification) {
setTelegramConfigureState(overlay, fields, 'retry');
} else {
setTelegramConfigureState(overlay, fields, 'idle');
}
}
showToast('Configuration failed: ' + err.message, 'error');
@@ -2922,6 +3007,10 @@ function closeConfigureModal(extensionName) {
if (typeof extensionName !== 'string') extensionName = null;
const existing = getConfigureOverlay(extensionName);
if (existing) existing.remove();
if (!document.querySelector('.configure-overlay') && !document.querySelector('.auth-card')) {
setAuthFlowPending(false);
enableChatInput();
}
}
// Validate that a server-supplied OAuth URL is HTTPS before opening a popup.
src/channels/web/static/i18n/en.js
+4 -2
View File
@@ -342,10 +342,12 @@ I18n.register('en', {
// Configure
'config.title': 'Configure {name}',
'config.telegramOwnerHint': 'After saving, IronClaw will show a one-time code. Send `/start CODE` to your bot in Telegram, then click Verify owner.',
'config.telegramOwnerHint': 'After saving, IronClaw will show a one-time code. Send `/start CODE` to your bot in Telegram and IronClaw will finish setup automatically.',
'config.telegramChallengeTitle': 'Telegram owner verification',
'config.telegramOwnerWaiting': 'Waiting for Telegram owner verification...',
'config.telegramVerifyOwner': 'Verify owner',
'config.telegramCommandLabel': 'Send this in Telegram:',
'config.telegramStartOver': 'Start over',
'config.telegramStartOverHint': 'Telegram verification did not complete. Click Start over to generate a new code and try again.',
'config.telegramOpenBot': 'Open bot in Telegram',
'config.optional': ' (optional)',
'config.alreadySet': '(already set — leave empty to keep)',
src/channels/web/static/i18n/zh-CN.js
+6
View File
@@ -342,6 +342,12 @@ I18n.register('zh-CN', {
// 配置
'config.title': '配置 {name}',
'config.telegramOwnerHint': '保存后，IronClaw 会显示一次性验证码。将 `/start CODE` 发送给你的 Telegram 机器人，IronClaw 会自动完成设置。',
'config.telegramChallengeTitle': 'Telegram 所有者验证',
'config.telegramOwnerWaiting': '正在等待 Telegram 所有者验证...',
'config.telegramCommandLabel': '请在 Telegram 中发送：',
'config.telegramStartOver': '重新开始',
'config.telegramStartOverHint': 'Telegram 验证未完成。点击“重新开始”以生成新的验证码并重试。',
'config.optional': '（可选）',
'config.alreadySet': '（已设置 — 留空以保持不变）',
'config.alreadyConfigured': '已配置',
src/channels/web/static/style.css
+22
View File
@@ -2952,6 +2952,28 @@ body {
text-decoration: underline;
}
.configure-inline-error {
margin: 16px 0 0 0;
padding: 10px 12px;
border-radius: 8px;
background: rgba(220, 38, 38, 0.12);
border: 1px solid rgba(220, 38, 38, 0.35);
color: #fca5a5;
font-size: 13px;
line-height: 1.5;
}
.configure-inline-status {
margin: 16px 0 0 0;
padding: 10px 12px;
border-radius: 8px;
background: var(--bg-secondary);
border: 1px solid var(--border);
color: var(--text-secondary);
font-size: 13px;
line-height: 1.5;
}
.configure-form {
display: flex;
flex-direction: column;
src/extensions/manager.rs
+165 -11
View File
@@ -140,6 +140,13 @@ struct TelegramGetUpdatesResponse {
description: Option<String>,
}
#[derive(Debug, serde::Deserialize)]
struct TelegramApiOkResponse {
ok: bool,
#[serde(default)]
description: Option<String>,
}
#[derive(Debug, serde::Deserialize)]
struct TelegramUpdate {
update_id: i64,
@@ -204,7 +211,7 @@ fn channel_auth_instructions(
) -> String {
if channel_name == TELEGRAM_CHANNEL_NAME && secret.name == "telegram_bot_token" {
return format!(
"{} After you submit it, IronClaw will show a one-time verification code. Send `/start CODE` to your bot in Telegram, then verify again to bind the owner.",
"{} After you submit it, IronClaw will show a one-time verification code. Send `/start CODE` to your bot in Telegram and IronClaw will finish setup automatically.",
secret.prompt
);
}
@@ -237,10 +244,12 @@ fn telegram_verification_deep_link(bot_username: Option<&str>, code: &str) -> Op
fn telegram_verification_instructions(bot_username: Option<&str>, code: &str) -> String {
if let Some(username) = bot_username.filter(|username| !username.trim().is_empty()) {
return format!("Send `/start {code}` to @{username}, then click Verify owner.");
return format!(
"Send `/start {code}` to @{username} in Telegram. IronClaw will finish setup automatically."
);
}
format!("Send `/start {code}` to your Telegram bot, then click Verify owner.")
format!("Send `/start {code}` to your Telegram bot. IronClaw will finish setup automatically.")
}
fn telegram_message_matches_verification_code(text: &str, code: &str) -> bool {
@@ -253,6 +262,42 @@ fn telegram_message_matches_verification_code(text: &str, code: &str) -> bool {
.any(|token| token == code)
}
async fn send_telegram_text_message(
client: &reqwest::Client,
endpoint: &str,
chat_id: i64,
text: &str,
) -> Result<(), ExtensionError> {
let response = client
.post(endpoint)
.json(&serde_json::json!({
"chat_id": chat_id,
"text": text,
}))
.send()
.await
.map_err(|e| telegram_request_error("sendMessage", &e))?;
if !response.status().is_success() {
return Err(ExtensionError::Other(format!(
"Telegram sendMessage failed (HTTP {})",
response.status()
)));
}
let payload: TelegramApiOkResponse = response
.json()
.await
.map_err(|e| telegram_response_parse_error("sendMessage", &e))?;
if !payload.ok {
return Err(ExtensionError::Other(payload.description.unwrap_or_else(
|| "Telegram sendMessage returned ok=false".to_string(),
)));
}
Ok(())
}
/// Central manager for extension lifecycle operations.
///
/// # Initialization Order
@@ -421,6 +466,29 @@ impl ExtensionManager {
*self.test_telegram_binding_resolver.write().await = Some(resolver);
}
#[cfg(test)]
pub(crate) async fn set_test_telegram_pending_verification(
&self,
code: &str,
bot_username: Option<&str>,
) {
let code = code.to_string();
let bot_username = bot_username.map(str::to_string);
self.set_test_telegram_binding_resolver(Arc::new(move |_token, existing_owner_id| {
if existing_owner_id.is_some() {
return Err(ExtensionError::Other(
"unexpected existing owner binding".to_string(),
));
}
Ok(TelegramBindingResult::Pending(VerificationChallenge {
code: code.clone(),
instructions: telegram_verification_instructions(bot_username.as_deref(), &code),
deep_link: telegram_verification_deep_link(bot_username.as_deref(), &code),
}))
}))
.await;
}
/// Enable gateway mode so OAuth flows return auth URLs to the frontend
/// instead of calling `open::that()` on the server.
///
@@ -597,6 +665,12 @@ impl ExtensionManager {
self.current_channel_owner_id(name).await.is_some()
}
pub(crate) async fn notification_target_for_channel(&self, name: &str) -> Option<String> {
self.current_channel_owner_id(name)
.await
.map(|owner_id| owner_id.to_string())
}
async fn get_pending_telegram_verification(
&self,
name: &str,
@@ -1074,7 +1148,7 @@ impl ExtensionManager {
active,
tools: Vec::new(),
needs_setup: auth_state == ToolAuthState::NeedsSetup,
has_auth: false,
has_auth: auth_state != ToolAuthState::NoAuth,
installed: true,
activation_error,
version,
@@ -4336,6 +4410,22 @@ impl ExtensionManager {
}
if let Some(owner_id) = bound_owner_id {
if let Err(err) = send_telegram_text_message(
&client,
&format!("https://api.telegram.org/bot{bot_token}/sendMessage"),
owner_id,
"Verification received. Finishing setup...",
)
.await
{
tracing::warn!(
channel = name,
owner_id,
error = %err,
"Failed to send Telegram verification acknowledgment"
);
}
self.clear_pending_telegram_verification(name).await;
if offset > 0 {
let _ = client
@@ -4355,10 +4445,10 @@ impl ExtensionManager {
}
}
Err(ExtensionError::ValidationFailed(format!(
"Telegram owner verification timed out. Send `/start {}` to your bot, then click Verify owner again.",
challenge.code
)))
self.clear_pending_telegram_verification(name).await;
Err(ExtensionError::ValidationFailed(
"Telegram owner verification timed out. Request a new code and try again.".to_string(),
))
}
async fn notify_telegram_owner_verified(
@@ -5120,7 +5210,7 @@ mod tests {
use crate::extensions::manager::{
ChannelRuntimeState, FallbackDecision, TelegramBindingData, TelegramBindingResult,
TelegramOwnerBindingState, build_wasm_channel_runtime_config_updates,
combine_install_errors, fallback_decision, infer_kind_from_url,
combine_install_errors, fallback_decision, infer_kind_from_url, send_telegram_text_message,
telegram_message_matches_verification_code,
};
use crate::extensions::{
@@ -5923,7 +6013,7 @@ mod tests {
Ok(TelegramBindingResult::Pending(VerificationChallenge {
code: "iclaw-7qk2m9".to_string(),
instructions:
"Send `/start iclaw-7qk2m9` to @test_hot_bot, then click Verify owner."
"Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically."
.to_string(),
deep_link: Some("https://t.me/test_hot_bot?start=iclaw-7qk2m9".to_string()),
}))
@@ -6875,11 +6965,75 @@ mod tests {
)?;
require(
instructions.contains("one-time verification code")
&& instructions.contains("/start CODE"),
&& instructions.contains("/start CODE")
&& instructions.contains("finish setup automatically"),
"telegram auth instructions should explain the owner verification step",
)
}
#[tokio::test]
async fn test_send_telegram_text_message_posts_expected_payload() -> Result<(), String> {
use axum::{Json, Router, extract::State, routing::post};
let payloads = Arc::new(tokio::sync::Mutex::new(Vec::<serde_json::Value>::new()));
async fn handler(
State(payloads): State<Arc<tokio::sync::Mutex<Vec<serde_json::Value>>>>,
Json(payload): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
payloads.lock().await.push(payload);
Json(serde_json::json!({ "ok": true, "result": {} }))
}
let app = Router::new()
.route("/sendMessage", post(handler))
.with_state(Arc::clone(&payloads));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.map_err(|err| format!("bind listener: {err}"))?;
let addr = listener
.local_addr()
.map_err(|err| format!("listener addr: {err}"))?;
let server = tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
let client = reqwest::Client::new();
send_telegram_text_message(
&client,
&format!("http://{addr}/sendMessage"),
424242,
"Verification received. Finishing setup...",
)
.await
.map_err(|err| format!("send message: {err}"))?;
let captured = tokio::time::timeout(std::time::Duration::from_secs(1), async {
loop {
let maybe_payload = { payloads.lock().await.first().cloned() };
if let Some(payload) = maybe_payload {
break payload;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.map_err(|_| "timed out waiting for sendMessage payload".to_string())?;
server.abort();
require_eq(
captured["chat_id"].clone(),
serde_json::json!(424242),
"chat_id",
)?;
require_eq(
captured["text"].clone(),
serde_json::json!("Verification received. Finishing setup..."),
"text",
)
}
#[test]
fn test_telegram_message_matches_verification_code_variants() -> Result<(), String> {
require(
src/main.rs
+1 -1
View File
@@ -620,7 +620,7 @@ async fn async_main() -> anyhow::Result<()> {
// Register message tool for sending messages to connected channels
components
.tools
.register_message_tools(Arc::clone(&channels))
.register_message_tools(Arc::clone(&channels), components.extension_manager.clone())
.await;
// Wire up channel runtime for hot-activation of WASM channels.
src/tools/builtin/message.rs
+90 -23
View File
@@ -10,6 +10,7 @@ use async_trait::async_trait;
use crate::bootstrap::ironclaw_base_dir;
use crate::channels::{ChannelManager, OutgoingResponse};
use crate::context::JobContext;
use crate::extensions::ExtensionManager;
use crate::tools::tool::{
ApprovalRequirement, Tool, ToolError, ToolOutput, ToolRateLimitConfig, require_str,
};
@@ -17,6 +18,7 @@ use crate::tools::tool::{
/// Tool for sending messages to channels.
pub struct MessageTool {
channel_manager: Arc<ChannelManager>,
extension_manager: Option<Arc<ExtensionManager>>,
/// Default channel for current conversation (set per-turn).
/// Uses std::sync::RwLock because requires_approval() is sync and called from async context.
default_channel: Arc<RwLock<Option<String>>>,
@@ -32,12 +34,18 @@ impl MessageTool {
Self {
channel_manager,
extension_manager: None,
default_channel: Arc::new(RwLock::new(None)),
default_target: Arc::new(RwLock::new(None)),
base_dir,
}
}
pub fn with_extension_manager(mut self, extension_manager: Arc<ExtensionManager>) -> Self {
self.extension_manager = Some(extension_manager);
self
}
/// Set the base directory for attachment validation.
/// This is primarily used for testing or future configuration.
pub fn with_base_dir(mut self, dir: PathBuf) -> Self {
@@ -111,39 +119,69 @@ impl Tool for MessageTool {
let content = require_str(&params, "content")?;
let explicit_channel = params
.get("channel")
.and_then(|v| v.as_str())
.map(|value| value.to_string());
let default_channel = self
.default_channel
.read()
.unwrap_or_else(|e| e.into_inner())
.clone();
let metadata_channel = ctx
.metadata
.get("notify_channel")
.and_then(|v| v.as_str())
.map(|value| value.to_string());
// Get channel: use param → conversation default → job metadata → None (broadcast all)
let channel: Option<String> =
if let Some(c) = params.get("channel").and_then(|v| v.as_str()) {
Some(c.to_string())
} else if let Some(c) = self
.default_channel
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
{
Some(c)
} else {
ctx.metadata
.get("notify_channel")
.and_then(|v| v.as_str())
.map(|c| c.to_string())
};
let channel: Option<String> = explicit_channel
.clone()
.or_else(|| default_channel.clone())
.or_else(|| metadata_channel.clone());
let can_use_default_target = match (explicit_channel.as_deref(), default_channel.as_deref())
{
(None, _) => true,
(Some(explicit), Some(current)) if explicit == current => true,
_ => false,
};
let can_use_metadata_target = match (channel.as_deref(), metadata_channel.as_deref()) {
(None, _) => true,
(Some(resolved), Some(current)) if resolved == current => true,
_ => false,
};
// Get target: use param → conversation default → job metadata → owner scope
// fallback when a specific channel is known.
let target = if let Some(t) = params.get("target").and_then(|v| v.as_str()) {
Some(t.to_string())
} else if let Some(t) = self
.default_target
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
} else if can_use_default_target
&& let Some(t) = self
.default_target
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
{
Some(t)
} else if let Some(t) = ctx.metadata.get("notify_user").and_then(|v| v.as_str()) {
} else if can_use_metadata_target
&& let Some(t) = ctx.metadata.get("notify_user").and_then(|v| v.as_str())
{
Some(t.to_string())
} else if channel.is_some() {
Some(ctx.user_id.clone())
if let Some(channel_name) = channel.as_deref() {
if let Some(extension_manager) = self.extension_manager.as_ref()
&& let Some(target) = extension_manager
.notification_target_for_channel(channel_name)
.await
{
Some(target)
} else {
Some(ctx.user_id.clone())
}
} else {
Some(ctx.user_id.clone())
}
} else {
None
};
@@ -742,4 +780,33 @@ mod tests {
err
);
}
#[tokio::test]
async fn message_tool_does_not_apply_metadata_target_to_different_default_channel() {
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
tool.set_context(Some("telegram".to_string()), None).await;
let mut ctx = crate::context::JobContext::with_user("owner-scope", "test", "test");
ctx.metadata = serde_json::json!({
"notify_channel": "signal",
"notify_user": "metadata-user",
});
let result = tool
.execute(serde_json::json!({"content": "hello"}), &ctx)
.await;
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(
!err.contains("metadata-user"),
"metadata target should not be applied to a different default channel: {}",
err
);
assert!(
err.contains("owner-scope"),
"expected owner-scope fallback target when metadata channel differs: {}",
err
);
}
}
src/tools/registry.rs
+6 -1
View File
@@ -501,9 +501,14 @@ impl ToolRegistry {
pub async fn register_message_tools(
&self,
channel_manager: Arc<crate::channels::ChannelManager>,
extension_manager: Option<Arc<crate::extensions::ExtensionManager>>,
) {
use crate::tools::builtin::MessageTool;
let tool = Arc::new(MessageTool::new(channel_manager));
let mut tool = MessageTool::new(channel_manager);
if let Some(extension_manager) = extension_manager {
tool = tool.with_extension_manager(extension_manager);
}
let tool = Arc::new(tool);
*self.message_tool.write().await = Some(Arc::clone(&tool));
self.tools
.write()
tests/e2e/scenarios/test_extensions.py
+30
View File
@@ -885,6 +885,36 @@ async def test_auth_and_configure_helpers_escape_selector_sensitive_extension_na
assert result["configureStillPresent"] is False
async def test_auth_required_does_not_reopen_existing_configure_modal(page):
"""Regression: auth_required SSE should not clobber an already-open configure modal."""
result = await page.evaluate(
"""() => {
const overlay = document.createElement('div');
overlay.className = 'configure-overlay';
overlay.setAttribute('data-extension-name', 'telegram');
document.body.appendChild(overlay);
const originalShowConfigureModal = window.showConfigureModal;
const originalSetAuthFlowPending = window.setAuthFlowPending;
let showCalls = 0;
let pendingCalls = 0;
window.showConfigureModal = () => { showCalls += 1; };
window.setAuthFlowPending = () => { pendingCalls += 1; };
handleAuthRequired({ extension_name: 'telegram', instructions: 'pending', auth_url: null });
window.showConfigureModal = originalShowConfigureModal;
window.setAuthFlowPending = originalSetAuthFlowPending;
overlay.remove();
return { showCalls, pendingCalls };
}"""
)
assert result["showCalls"] == 0
assert result["pendingCalls"] == 0
async def test_auth_completed_sse_dismisses_card(page):
"""Simulating the auth_completed SSE event removes the auth card."""
await _show_auth_card(page, extension_name="myext")
tests/e2e/scenarios/test_telegram_hot_activation.py
+14 -7
View File
@@ -125,6 +125,8 @@ async def test_telegram_hot_activation_transitions_installed_to_active(page):
phase = {"value": "installed"}
captured_setup_payloads = []
post_count = {"value": 0}
second_request_started = asyncio.Event()
allow_second_response = asyncio.Event()
async def handle_ext_list(route):
extensions = {
@@ -170,16 +172,18 @@ async def test_telegram_hot_activation_transitions_installed_to_active(page):
{
"success": True,
"activated": False,
"message": "Configuration saved for 'telegram'. Send `/start iclaw-7qk2m9` to @test_hot_bot, then click Verify owner.",
"message": "Configuration saved for 'telegram'. Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
"verification": {
"code": "iclaw-7qk2m9",
"instructions": "Send `/start iclaw-7qk2m9` to @test_hot_bot, then click Verify owner.",
"instructions": "Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
"deep_link": "https://t.me/test_hot_bot?start=iclaw-7qk2m9",
},
}
),
)
else:
second_request_started.set()
await allow_second_response.wait()
await route.fulfill(
status=200,
content_type="application/json",
@@ -203,16 +207,19 @@ async def test_telegram_hot_activation_transitions_installed_to_active(page):
await modal.wait_for(state="visible", timeout=5000)
await modal.locator(_CONFIGURE_SECRET_INPUT).fill("123456789:ABCdefGhI")
await modal.locator(_CONFIGURE_SAVE_BUTTON).click()
await modal.locator(_CONFIGURE_SAVE_BUTTON, has_text="Verify owner").wait_for(
await second_request_started.wait()
await modal.locator(".configure-inline-status", has_text="Waiting for Telegram owner verification...").wait_for(
state="visible", timeout=5000
)
assert "Verify owner" in (
await modal.locator(_CONFIGURE_SAVE_BUTTON).text_content()
)
assert "iclaw-7qk2m9" in (await modal.text_content())
assert "/start iclaw-7qk2m9" in (await modal.text_content())
assert await modal.locator(".configure-verification-link").count() == 1
await modal.locator(_CONFIGURE_SAVE_BUTTON).wait_for(state="hidden", timeout=5000)
await modal.locator(_CONFIGURE_SAVE_BUTTON).click()
await page.locator(SEL["configure_overlay"]).click(position={"x": 1, "y": 1})
assert await page.locator(SEL["configure_overlay"]).is_visible()
allow_second_response.set()
await page.locator(SEL["configure_overlay"]).wait_for(state="hidden", timeout=5000)
phase["value"] = "active"
tests/e2e_advanced_traces.rs
+33 -1
View File
@@ -442,6 +442,9 @@ mod advanced {
other => panic!("expected event trigger, got {other:?}"),
}
rig.clear().await;
let llm_calls_before = rig.llm_call_count();
rig.send_incoming(IncomingMessage::new(
"telegram",
"test-user",
@@ -451,8 +454,18 @@ mod advanced {
let runs = wait_for_routine_run(rig.database(), routine.id, TIMEOUT).await;
assert_eq!(runs[0].trigger_type, "event");
assert_eq!(
rig.llm_call_count(),
llm_calls_before + 1,
"matching event message should only trigger the routine LLM call"
);
let responses = rig.wait_for_responses(3, TIMEOUT).await;
let responses = rig.wait_for_responses(1, TIMEOUT).await;
assert_eq!(
responses.len(),
1,
"expected only the routine notification after the matching event"
);
assert!(
responses.iter().any(|response| {
response
@@ -505,6 +518,9 @@ mod advanced {
other => panic!("expected event trigger, got {other:?}"),
}
rig.clear().await;
let llm_calls_before = rig.llm_call_count();
rig.send_incoming(IncomingMessage::new(
"telegram",
"test-user",
@@ -514,6 +530,22 @@ mod advanced {
let runs = wait_for_routine_run(rig.database(), routine.id, TIMEOUT).await;
assert_eq!(runs[0].trigger_type, "event");
assert_eq!(
rig.llm_call_count(),
llm_calls_before + 1,
"matching event message should only trigger the routine LLM call"
);
let responses = rig.wait_for_responses(1, TIMEOUT).await;
assert_eq!(
responses.len(),
1,
"expected only the routine notification after the matching event"
);
assert!(
responses[0].content.contains("Bug report detected"),
"expected routine notification, got: {responses:?}"
);
rig.shutdown();
}
tests/e2e_telegram_message_routing.rs
+353
View File
@@ -0,0 +1,353 @@
//! E2E tests for Telegram message routing through the real agent + message tool.
#[cfg(feature = "libsql")]
mod support;
#[cfg(feature = "libsql")]
mod tests {
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures::StreamExt;
use ironclaw::agent::{Agent, AgentDeps};
use ironclaw::app::{AppBuilder, AppBuilderFlags};
use ironclaw::channels::web::log_layer::LogBroadcaster;
use ironclaw::channels::{
Channel, ChannelManager, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate,
};
use ironclaw::config::Config;
use ironclaw::db::{Database, libsql::LibSqlBackend};
use ironclaw::error::ChannelError;
use ironclaw::llm::{LlmProvider, SessionConfig, SessionManager};
use tokio::sync::{Mutex, mpsc};
use tokio_stream::wrappers::ReceiverStream;
use crate::support::test_channel::{TestChannel, TestChannelHandle};
use crate::support::trace_llm::{LlmTrace, TraceLlm, TraceResponse, TraceStep, TraceToolCall};
type TelegramCaptures = Arc<Mutex<Vec<(String, OutgoingResponse)>>>;
struct RecordingTelegramChannel {
captures: TelegramCaptures,
}
impl RecordingTelegramChannel {
fn new() -> (Self, TelegramCaptures) {
let captures = Arc::new(Mutex::new(Vec::new()));
(
Self {
captures: Arc::clone(&captures),
},
captures,
)
}
}
#[async_trait]
impl Channel for RecordingTelegramChannel {
fn name(&self) -> &str {
"telegram"
}
async fn start(&self) -> Result<MessageStream, ChannelError> {
let (_tx, rx) = mpsc::channel::<IncomingMessage>(1);
Ok(ReceiverStream::new(rx).boxed())
}
async fn respond(
&self,
_msg: &IncomingMessage,
response: OutgoingResponse,
) -> Result<(), ChannelError> {
self.captures
.lock()
.await
.push(("respond".to_string(), response));
Ok(())
}
async fn send_status(
&self,
_status: StatusUpdate,
_metadata: &serde_json::Value,
) -> Result<(), ChannelError> {
Ok(())
}
async fn broadcast(
&self,
user_id: &str,
response: OutgoingResponse,
) -> Result<(), ChannelError> {
self.captures
.lock()
.await
.push((user_id.to_string(), response));
Ok(())
}
async fn health_check(&self) -> Result<(), ChannelError> {
Ok(())
}
}
struct Harness {
gateway: Arc<TestChannel>,
telegram_captures: Arc<Mutex<Vec<(String, OutgoingResponse)>>>,
db: Arc<dyn Database>,
owner_id: String,
_temp_dir: tempfile::TempDir,
agent_handle: Option<tokio::task::JoinHandle<()>>,
}
impl Harness {
async fn store_telegram_owner_binding(&self, owner_id: i64) {
for scope in [&self.owner_id, "test-user"] {
self.db
.set_setting(
scope,
"channels.wasm_channel_owner_ids.telegram",
&serde_json::json!(owner_id),
)
.await
.expect("failed to store telegram owner binding");
}
}
async fn wait_for_telegram_broadcasts(
&self,
expected: usize,
timeout: Duration,
) -> Vec<(String, OutgoingResponse)> {
let deadline = tokio::time::Instant::now() + timeout;
loop {
let snapshot = self.telegram_captures.lock().await.clone();
if snapshot.len() >= expected || tokio::time::Instant::now() >= deadline {
return snapshot;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
}
impl Drop for Harness {
fn drop(&mut self) {
self.gateway.signal_shutdown();
if let Some(handle) = self.agent_handle.take() {
handle.abort();
}
}
}
async fn build_harness(trace: LlmTrace) -> Harness {
let temp_dir = tempfile::tempdir().expect("failed to create temp dir");
let db_path = temp_dir.path().join("telegram_message_routing.db");
let backend = LibSqlBackend::new_local(&db_path)
.await
.expect("failed to create test LibSqlBackend");
backend
.run_migrations()
.await
.expect("failed to run migrations");
let db: Arc<dyn Database> = Arc::new(backend);
let skills_dir = temp_dir.path().join("skills");
let installed_skills_dir = temp_dir.path().join("installed_skills");
let _ = std::fs::create_dir_all(&skills_dir);
let _ = std::fs::create_dir_all(&installed_skills_dir);
let mut config = Config::for_testing(db_path, skills_dir, installed_skills_dir);
config.agent.auto_approve_tools = true;
let session = Arc::new(SessionManager::new(SessionConfig::default()));
let log_broadcaster = Arc::new(LogBroadcaster::new());
let llm: Arc<dyn LlmProvider> = Arc::new(TraceLlm::from_trace(trace));
let mut builder = AppBuilder::new(
config,
AppBuilderFlags::default(),
None,
session,
log_broadcaster,
);
builder.with_database(Arc::clone(&db));
builder.with_llm(llm);
let mut components = builder
.build_all()
.await
.expect("AppBuilder::build_all() failed");
components.config.agent.auto_approve_tools = true;
components.config.agent.allow_local_tools = true;
let deps = AgentDeps {
owner_id: components.config.owner_id.clone(),
store: components.db.clone(),
llm: components.llm.clone(),
cheap_llm: components.cheap_llm.clone(),
safety: components.safety.clone(),
tools: components.tools.clone(),
workspace: components.workspace.clone(),
extension_manager: components.extension_manager.clone(),
skill_registry: components.skill_registry.clone(),
skill_catalog: components.skill_catalog.clone(),
skills_config: components.config.skills.clone(),
hooks: components.hooks.clone(),
cost_guard: components.cost_guard.clone(),
sse_tx: None,
http_interceptor: None,
transcription: None,
document_extraction: None,
};
let gateway = Arc::new(TestChannel::new());
let gateway_handle = TestChannelHandle::new(Arc::clone(&gateway));
let (telegram_channel, telegram_captures) = RecordingTelegramChannel::new();
let channel_manager = ChannelManager::new();
channel_manager.add(Box::new(gateway_handle)).await;
channel_manager.add(Box::new(telegram_channel)).await;
let channels = Arc::new(channel_manager);
deps.tools
.register_message_tools(Arc::clone(&channels), deps.extension_manager.clone())
.await;
let agent = Agent::new(
components.config.agent.clone(),
deps,
channels,
None,
None,
None,
Some(Arc::clone(&components.context_manager)),
None,
);
let agent_handle = tokio::spawn(async move {
if let Err(err) = agent.run().await {
eprintln!("[telegram routing e2e] Agent exited with error: {err}");
}
});
if let Some(rx) = gateway.take_ready_rx().await {
let _ = tokio::time::timeout(Duration::from_secs(5), rx).await;
}
Harness {
gateway,
telegram_captures,
db,
owner_id: components.config.owner_id.clone(),
_temp_dir: temp_dir,
agent_handle: Some(agent_handle),
}
}
fn single_message_trace(arguments: serde_json::Value, final_text: &str) -> LlmTrace {
LlmTrace::single_turn(
"telegram-message-routing",
"send a reminder",
vec![
TraceStep {
request_hint: None,
response: TraceResponse::ToolCalls {
tool_calls: vec![TraceToolCall {
id: "call_message_1".to_string(),
name: "message".to_string(),
arguments,
}],
input_tokens: 32,
output_tokens: 12,
},
expected_tool_results: Vec::new(),
},
TraceStep {
request_hint: None,
response: TraceResponse::Text {
content: final_text.to_string(),
input_tokens: 24,
output_tokens: 8,
},
expected_tool_results: Vec::new(),
},
],
)
}
#[tokio::test]
async fn telegram_message_tool_uses_bound_owner_target_when_target_omitted() {
let harness = build_harness(single_message_trace(
serde_json::json!({
"content": "Walk Conan",
"channel": "telegram",
}),
"Sent on Telegram.",
))
.await;
harness.store_telegram_owner_binding(424242).await;
harness
.gateway
.send_message("remind me to walk conan")
.await;
let responses = harness
.gateway
.wait_for_responses(1, Duration::from_secs(10))
.await;
assert!(
responses
.iter()
.any(|response| response.content.contains("Sent on Telegram")),
"expected assistant confirmation, got: {:?}",
responses
.iter()
.map(|response| &response.content)
.collect::<Vec<_>>()
);
let broadcasts = harness
.wait_for_telegram_broadcasts(1, Duration::from_secs(10))
.await;
assert_eq!(
broadcasts.len(),
1,
"expected exactly one telegram broadcast"
);
assert_eq!(broadcasts[0].0, "424242");
assert_eq!(broadcasts[0].1.content, "Walk Conan");
}
#[tokio::test]
async fn telegram_message_tool_prefers_explicit_target_over_bound_owner_target() {
let harness = build_harness(single_message_trace(
serde_json::json!({
"content": "Walk Conan",
"channel": "telegram",
"target": "999999",
}),
"Sent on Telegram.",
))
.await;
harness.store_telegram_owner_binding(424242).await;
harness.gateway.send_message("send the reminder").await;
let _ = harness
.gateway
.wait_for_responses(1, Duration::from_secs(10))
.await;
let broadcasts = harness
.wait_for_telegram_broadcasts(1, Duration::from_secs(10))
.await;
assert_eq!(
broadcasts.len(),
1,
"expected exactly one telegram broadcast"
);
assert_eq!(broadcasts[0].0, "999999");
assert_eq!(broadcasts[0].1.content, "Walk Conan");
}
}
tests/fixtures/llm_traces/advanced/routine_event_any_channel.json
Vendored
-8
View File
@@ -34,14 +34,6 @@
"output_tokens": 18
}
},
{
"response": {
"type": "text",
"content": "I saw the Telegram message.",
"input_tokens": 90,
"output_tokens": 12
}
},
{
"response": {
"type": "text",
tests/fixtures/llm_traces/advanced/routine_event_telegram.json
Vendored
-8
View File
@@ -35,14 +35,6 @@
"output_tokens": 20
}
},
{
"response": {
"type": "text",
"content": "I saw the Telegram message.",
"input_tokens": 90,
"output_tokens": 12
}
},
{
"response": {
"type": "text",
tests/support/test_rig.rs
+1 -1
View File
@@ -653,7 +653,7 @@ impl TestRigBuilder {
// 7b. Register message tool so routines can send messages to channels.
deps.tools
.register_message_tools(Arc::clone(&channels))
.register_message_tools(Arc::clone(&channels), deps.extension_manager.clone())
.await;
// 8. Create Agent.