diff --git a/.env.example b/.env.example
index 55c3adb5..8fd44c5a 100644
--- a/.env.example
+++ b/.env.example
@@ -78,7 +78,7 @@ NEARAI_AUTH_URL=https://private.near.ai
# === MiniMax ===
# LLM_BACKEND=minimax
# MINIMAX_API_KEY=...
-# MINIMAX_MODEL=MiniMax-M2.5
+# MINIMAX_MODEL=MiniMax-M2.7
# MINIMAX_BASE_URL=https://api.minimax.io/v1 # default (global); use https://api.minimaxi.com/v1 for China
# === Anthropic Direct ===
diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml
index e7371677..2f885b16 100644
--- a/.github/workflows/coverage.yml
+++ b/.github/workflows/coverage.yml
@@ -174,7 +174,7 @@ jobs:
- name: Run E2E tests
run: |
- pytest tests/e2e/ -v -x --timeout=120
+ pytest tests/e2e/ -v --timeout=120
env:
RUST_LOG: ironclaw=info
RUST_BACKTRACE: "1"
diff --git a/.github/workflows/regression-test-check.yml b/.github/workflows/regression-test-check.yml
index 6d97c4ce..ef1a4d92 100644
--- a/.github/workflows/regression-test-check.yml
+++ b/.github/workflows/regression-test-check.yml
@@ -43,12 +43,42 @@ jobs:
fi
fi
- if [ "$IS_FIX" = false ]; then
- echo "Not a fix PR — skipping regression test check."
+ # --- 1b. Does this PR touch high-risk state machine or resilience code? ---
+ CHANGED_FILES=$(git diff --name-only "${BASE_REF}...${HEAD_REF}")
+
+ TOUCHES_HIGH_RISK=false
+ HIGH_RISK_PATTERNS=(
+ "src/context/state.rs"
+ "src/agent/session.rs"
+ "src/llm/circuit_breaker.rs"
+ "src/llm/retry.rs"
+ "src/llm/failover.rs"
+ "src/agent/self_repair.rs"
+ "src/agent/agentic_loop.rs"
+ "src/tools/execute.rs"
+ "crates/ironclaw_safety/src/"
+ )
+
+ for pattern in "${HIGH_RISK_PATTERNS[@]}"; do
+ if echo "$CHANGED_FILES" | grep -q "$pattern"; then
+ TOUCHES_HIGH_RISK=true
+ echo "High-risk file matched: $pattern"
+ break
+ fi
+ done
+
+ # Skip only if NEITHER condition holds — no double-firing on fix PRs
+ if [ "$IS_FIX" = false ] && [ "$TOUCHES_HIGH_RISK" = false ]; then
+ echo "Not a fix PR and no high-risk files changed — skipping."
exit 0
fi
- echo "Fix PR detected."
+ if [ "$IS_FIX" = true ]; then
+ echo "Fix PR detected."
+ fi
+ if [ "$TOUCHES_HIGH_RISK" = true ]; then
+ echo "High-risk state machine or resilience code modified."
+ fi
# --- 2. Skip label or commit message marker ---
if grep -qF ',skip-regression-check,' <<< ",$PR_LABELS,"; then
@@ -63,8 +93,6 @@ jobs:
fi
# --- 3. Exempt static-only / docs-only changes ---
- CHANGED_FILES=$(git diff --name-only "${BASE_REF}...${HEAD_REF}")
-
if [ -z "$CHANGED_FILES" ]; then
echo "No changed files — skipping."
exit 0
@@ -110,5 +138,12 @@ jobs:
fi
# --- 5. No tests found ---
- echo "::warning::This PR looks like a bug fix but contains no test changes. Every fix should include a regression test. Add a #[test] or #[tokio::test], or apply the 'skip-regression-check' label if not feasible."
+ if [ "$IS_FIX" = true ]; then
+ echo "::warning::This PR looks like a bug fix but contains no test changes."
+ fi
+ if [ "$TOUCHES_HIGH_RISK" = true ]; then
+ echo "::warning::This PR modifies high-risk state machine or resilience code but includes no test changes."
+ fi
+ echo "::warning::Please add tests exercising the changed behavior, or apply the 'skip-regression-check' label if not feasible."
exit 1
+
diff --git a/README.ja.md b/README.ja.md
new file mode 100644
index 00000000..887cf67e
--- /dev/null
+++ b/README.ja.md
@@ -0,0 +1,330 @@
+
+
+
+
+IronClaw
+
+
+ あなたの味方になる、安全なパーソナルAIアシスタント
+
+
+
+
+
+
+
+
+
+ English |
+ 简体中文 |
+ Русский |
+ 日本語
+
+
+
+ フィロソフィー •
+ 機能 •
+ インストール •
+ 設定 •
+ セキュリティ •
+ アーキテクチャ
+
+
+---
+
+## フィロソフィー
+
+IronClawはシンプルな原則に基づいて構築されています:**あなたのAIアシスタントは、あなたのために働くべきであり、あなたに不利益をもたらすべきではありません。**
+
+AIシステムがデータの取り扱いについて不透明になり、企業の利益に沿って調整されることが増えている世界で、IronClawは異なるアプローチを取ります:
+
+- **あなたのデータはあなたのもの** - すべての情報はローカルに保存・暗号化され、あなたの管理下から離れることはありません
+- **設計段階からの透明性** - オープンソース、監査可能、隠れたテレメトリやデータ収集なし
+- **自己拡張する能力** - ベンダーのアップデートを待たずに、新しいツールをその場で構築
+- **多層防御** - 複数のセキュリティレイヤーがプロンプトインジェクションやデータ流出から保護
+
+IronClawは、個人生活にも仕事にも本当に信頼できるAIアシスタントです。
+
+## 機能
+
+### セキュリティファースト
+
+- **WASMサンドボックス** - 信頼されていないツールは、機能ベースの権限を持つ隔離されたWebAssemblyコンテナで実行
+- **認証情報の保護** - シークレットはツールに公開されず、リーク検出付きでホスト境界で注入
+- **プロンプトインジェクション防御** - パターン検出、コンテンツサニタイズ、ポリシー適用
+- **エンドポイントの許可リスト** - HTTPリクエストは明示的に許可されたホストとパスのみに制限
+
+### 常時利用可能
+
+- **マルチチャネル** - REPL、HTTPウェブフック、WASMチャネル(Telegram、Slack)、Webゲートウェイ
+- **Dockerサンドボックス** - ジョブごとのトークンとオーケストレーター/ワーカーパターンによる隔離されたコンテナ実行
+- **Webゲートウェイ** - リアルタイムSSE/WebSocketストリーミング対応のブラウザUI
+- **ルーティン** - cronスケジュール、イベントトリガー、ウェブフックハンドラーによるバックグラウンド自動化
+- **ハートビートシステム** - 監視・保守タスクのためのプロアクティブなバックグラウンド実行
+- **並列ジョブ** - 隔離されたコンテキストで複数のリクエストを同時に処理
+- **自己修復** - スタックした操作の自動検出と復旧
+
+### 自己拡張
+
+- **動的ツール構築** - 必要なものを説明すると、IronClawがWASMツールとして構築
+- **MCPプロトコル** - Model Context Protocolサーバーに接続して追加機能を利用
+- **プラグインアーキテクチャ** - 再起動なしで新しいWASMツールやチャネルを追加
+
+### 永続メモリ
+
+- **ハイブリッド検索** - Reciprocal Rank Fusionを使用した全文検索+ベクトル検索
+- **ワークスペースファイルシステム** - メモ、ログ、コンテキストのための柔軟なパスベースストレージ
+- **アイデンティティファイル** - セッション間で一貫した人格と設定を維持
+
+## インストール
+
+### 前提条件
+
+- Rust 1.85+
+- PostgreSQL 15+ ([pgvector](https://github.com/pgvector/pgvector)拡張機能を含む)
+- NEAR AIアカウント(セットアップウィザードで認証を処理)
+
+## ダウンロードまたはビルド
+
+最新のアップデートは[リリースページ](https://github.com/nearai/ironclaw/releases/)をご覧ください。
+
+
+ Windowsインストーラーでインストール(Windows)
+
+[Windowsインストーラー](https://github.com/nearai/ironclaw/releases/latest/download/ironclaw-x86_64-pc-windows-msvc.msi)をダウンロードして実行してください。
+
+
+
+
+ PowerShellスクリプトでインストール(Windows)
+
+```sh
+irm https://github.com/nearai/ironclaw/releases/latest/download/ironclaw-installer.ps1 | iex
+```
+
+
+
+
+ シェルスクリプトでインストール(macOS、Linux、Windows/WSL)
+
+```sh
+curl --proto '=https' --tlsv1.2 -LsSf https://github.com/nearai/ironclaw/releases/latest/download/ironclaw-installer.sh | sh
+```
+
+
+
+ Homebrewでインストール(macOS/Linux)
+
+```sh
+brew install ironclaw
+```
+
+
+
+
+ ソースコードからコンパイル(Windows、Linux、macOSでCargo)
+
+`cargo`でインストールします。コンピューターに[Rust](https://rustup.rs)がインストールされていることを確認してください。
+
+```bash
+# リポジトリをクローン
+git clone https://github.com/nearai/ironclaw.git
+cd ironclaw
+
+# ビルド
+cargo build --release
+
+# テストを実行
+cargo test
+```
+
+**フルリリース**(チャネルソースを変更した後)の場合、まず`./scripts/build-all.sh`を実行してチャネルを再ビルドしてください。
+
+
+
+### データベースのセットアップ
+
+```bash
+# データベースを作成
+createdb ironclaw
+
+# pgvectorを有効化
+psql ironclaw -c "CREATE EXTENSION IF NOT EXISTS vector;"
+```
+
+## 設定
+
+セットアップウィザードを実行してIronClawを設定します:
+
+```bash
+ironclaw onboard
+```
+
+ウィザードは、データベース接続、NEAR AI認証(ブラウザOAuth経由)、シークレットの暗号化(システムキーチェーンを使用)を処理します。設定は接続されたデータベースに永続化されます。ブートストラップ変数(例:`DATABASE_URL`、`LLM_BACKEND`)は、データベース接続前に利用できるよう`~/.ironclaw/.env`に書き込まれます。
+
+### 代替LLMプロバイダー
+
+IronClawはデフォルトでNEAR AIを使用しますが、多くのLLMプロバイダーをすぐに利用できます。組み込みプロバイダーには**Anthropic**、**OpenAI**、**Google Gemini**、**MiniMax**、**Mistral**、**Ollama**(ローカル)が含まれます。**OpenRouter**(300以上のモデル)、**Together AI**、**Fireworks AI**、セルフホストサーバー(**vLLM**、**LiteLLM**)などのOpenAI互換サービスもサポートされています。
+
+ウィザードでプロバイダーを選択するか、環境変数を直接設定してください:
+
+```env
+# 例:MiniMax(組み込み、204Kコンテキスト)
+LLM_BACKEND=minimax
+MINIMAX_API_KEY=...
+
+# 例:OpenAI互換エンドポイント
+LLM_BACKEND=openai_compatible
+LLM_BASE_URL=https://openrouter.ai/api/v1
+LLM_API_KEY=sk-or-...
+LLM_MODEL=anthropic/claude-sonnet-4
+```
+
+完全なプロバイダーガイドは[docs/LLM_PROVIDERS.md](docs/LLM_PROVIDERS.md)をご覧ください。
+
+## セキュリティ
+
+IronClawは、データを保護し悪用を防ぐために多層防御を実装しています。
+
+### WASMサンドボックス
+
+すべての信頼されていないツールは、隔離されたWebAssemblyコンテナで実行されます:
+
+- **機能ベースの権限** - HTTP、シークレット、ツール呼び出しの明示的なオプトイン
+- **エンドポイントの許可リスト** - 許可されたホスト/パスへのHTTPリクエストのみ
+- **認証情報の注入** - シークレットはホスト境界で注入され、WASMコードに公開されない
+- **リーク検出** - リクエストとレスポンスのシークレット流出試行をスキャン
+- **レート制限** - 悪用防止のためのツールごとのリクエスト制限
+- **リソース制限** - メモリ、CPU、実行時間の制約
+
+```
+WASM ──► 許可リスト ──► リーク ──► 認証情報 ──► リクエスト ──► リーク ──► WASM
+ バリデーター スキャン 注入 実行 スキャン
+ (リクエスト) (レスポンス)
+```
+
+### プロンプトインジェクション防御
+
+外部コンテンツは複数のセキュリティレイヤーを通過します:
+
+- パターンベースのインジェクション試行検出
+- コンテンツのサニタイズとエスケープ
+- 重要度レベル付きポリシールール(ブロック/警告/レビュー/サニタイズ)
+- 安全なLLMコンテキスト注入のためのツール出力ラッピング
+
+### データ保護
+
+- すべてのデータはローカルのPostgreSQLデータベースに保存
+- AES-256-GCMでシークレットを暗号化
+- テレメトリ、分析、データ共有なし
+- すべてのツール実行の完全な監査ログ
+
+## アーキテクチャ
+
+```
+┌────────────────────────────────────────────────────────────────┐
+│ チャネル │
+│ ┌──────┐ ┌──────┐ ┌─────────────┐ ┌─────────────┐ │
+│ │ REPL │ │ HTTP │ │WASMチャネル │ │ Web │ │
+│ └──┬───┘ └──┬───┘ └──────┬──────┘ │ ゲートウェイ│ │
+│ │ │ │ │(SSE + WS) │ │
+│ │ │ │ └──────┬──────┘ │
+│ └─────────┴──────────────┴────────────────┘ │
+│ │ │
+│ ┌─────────▼─────────┐ │
+│ │ エージェントループ │ インテントルーティング│
+│ └────┬──────────┬───┘ │
+│ │ │ │
+│ ┌──────────▼────┐ ┌──▼───────────────┐ │
+│ │ スケジューラー │ │ ルーティン │ │
+│ │ (並列ジョブ) │ │ エンジン │ │
+│ └──────┬────────┘ │(cron,event,wh) │ │
+│ │ └────────┬─────────┘ │
+│ ┌─────────────┼────────────────────┘ │
+│ │ │ │
+│ ┌───▼─────┐ ┌────▼────────────────┐ │
+│ │ ローカル │ │ オーケストレーター │ │
+│ │ ワーカー │ │ ┌───────────────┐ │ │
+│ │(プロセス │ │ │ Docker │ │ │
+│ │ 内) │ │ │ サンドボックス│ │ │
+│ └───┬─────┘ │ │ コンテナ │ │ │
+│ │ │ │ ┌───────────┐ │ │ │
+│ │ │ │ │Worker / CC│ │ │ │
+│ │ │ │ └───────────┘ │ │ │
+│ │ │ └───────────────┘ │ │
+│ │ └─────────┬───────────┘ │
+│ └──────────────────┤ │
+│ │ │
+│ ┌───────────▼──────────┐ │
+│ │ ツールレジストリ │ │
+│ │ 組み込み, MCP, WASM │ │
+│ └──────────────────────┘ │
+└────────────────────────────────────────────────────────────────┘
+```
+
+### コアコンポーネント
+
+| コンポーネント | 目的 |
+|---------------|------|
+| **エージェントループ** | メインのメッセージ処理とジョブの調整 |
+| **ルーター** | ユーザーの意図を分類(コマンド、クエリ、タスク) |
+| **スケジューラー** | 優先度付きの並列ジョブ実行を管理 |
+| **ワーカー** | LLM推論とツール呼び出しでジョブを実行 |
+| **オーケストレーター** | コンテナのライフサイクル、LLMプロキシ、ジョブごとの認証 |
+| **Webゲートウェイ** | チャット、メモリ、ジョブ、ログ、拡張機能、ルーティンのブラウザUI |
+| **ルーティンエンジン** | スケジュール(cron)とリアクティブ(イベント、ウェブフック)のバックグラウンドタスク |
+| **ワークスペース** | ハイブリッド検索付き永続メモリ |
+| **セーフティレイヤー** | プロンプトインジェクション防御とコンテンツサニタイズ |
+
+## 使い方
+
+```bash
+# 初回セットアップ(データベース、認証などを設定)
+ironclaw onboard
+
+# インタラクティブREPLを起動
+cargo run
+
+# デバッグログ付き
+RUST_LOG=ironclaw=debug cargo run
+```
+
+## 開発
+
+```bash
+# コードフォーマット
+cargo fmt
+
+# リント
+cargo clippy --all --benches --tests --examples --all-features
+
+# テスト実行
+createdb ironclaw_test
+cargo test
+
+# 特定のテストを実行
+cargo test test_name
+```
+
+- **Telegramチャネル**: セットアップとDMペアリングについては[docs/TELEGRAM_SETUP.md](docs/TELEGRAM_SETUP.md)を参照してください。
+- **チャネルソースの変更**: `cargo build`の前に`./channels-src/telegram/build.sh`を実行して、更新されたWASMをバンドルしてください。
+
+## OpenClawの系譜
+
+IronClawは[OpenClaw](https://github.com/openclaw/openclaw)にインスパイアされたRust再実装です。完全な対応表は[FEATURE_PARITY.md](FEATURE_PARITY.md)をご覧ください。
+
+主な違い:
+
+- **Rust vs TypeScript** - ネイティブパフォーマンス、メモリ安全性、シングルバイナリ
+- **WASMサンドボックス vs Docker** - 軽量、機能ベースのセキュリティ
+- **PostgreSQL vs SQLite** - 本番環境対応の永続化
+- **セキュリティファースト設計** - 複数の防御レイヤー、認証情報の保護
+
+## ライセンス
+
+以下のいずれかのライセンスの下で提供されています:
+
+- Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE))
+- MIT License ([LICENSE-MIT](LICENSE-MIT))
+
+お好みに応じて選択してください。
diff --git a/README.md b/README.md
index 9684ee4d..fa73dc45 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,8 @@
English |
简体中文 |
- Русский
+ Русский |
+ 日本語
diff --git a/README.ru.md b/README.ru.md
index c64770a9..0546e7f4 100644
--- a/README.ru.md
+++ b/README.ru.md
@@ -17,7 +17,8 @@
English |
简体中文 |
- Русский
+ Русский |
+ 日本語
diff --git a/README.zh-CN.md b/README.zh-CN.md
index 34023822..a337d713 100644
--- a/README.zh-CN.md
+++ b/README.zh-CN.md
@@ -17,7 +17,8 @@
English |
简体中文 |
- Русский
+ Русский |
+ 日本語
diff --git a/channels-src/feishu/src/lib.rs b/channels-src/feishu/src/lib.rs
index 2e7261d8..3094eaa0 100644
--- a/channels-src/feishu/src/lib.rs
+++ b/channels-src/feishu/src/lib.rs
@@ -206,9 +206,17 @@ struct FeishuApiResponse {
data: Option,
}
-/// Tenant access token response.
-#[derive(Debug, Default, Deserialize)]
-struct TenantAccessTokenData {
+/// Tenant access token response (flat format).
+///
+/// Unlike most Feishu APIs that nest results under `data`, the
+/// `/auth/v3/tenant_access_token/internal` endpoint returns `code`, `msg`,
+/// `tenant_access_token`, and `expire` at the top level.
+#[derive(Debug, Deserialize)]
+struct TenantAccessTokenResponse {
+ #[serde(default)]
+ code: i32,
+ #[serde(default)]
+ msg: String,
tenant_access_token: String,
expire: i64,
}
@@ -770,9 +778,8 @@ fn obtain_tenant_token(api_base: &str) -> Result {
));
}
- let token_resp: FeishuApiResponse =
- serde_json::from_slice(&response.body)
- .map_err(|e| format!("Failed to parse token response: {}", e))?;
+ let token_resp: TenantAccessTokenResponse = serde_json::from_slice(&response.body)
+ .map_err(|e| format!("Failed to parse token response: {}", e))?;
if token_resp.code != 0 {
return Err(format!(
@@ -781,23 +788,33 @@ fn obtain_tenant_token(api_base: &str) -> Result {
));
}
- let data = token_resp
- .data
- .ok_or_else(|| "Token response missing data".to_string())?;
+ if token_resp.tenant_access_token.is_empty() {
+ return Err("Token response missing tenant_access_token".to_string());
+ }
+
+ if token_resp.expire <= 0 {
+ return Err(format!(
+ "Token response has invalid expire value: {}",
+ token_resp.expire
+ ));
+ }
// Cache the token with expiry.
let now = channel_host::now_millis();
- let expiry = now + (data.expire as u64) * 1000;
+ let expiry = now.saturating_add((token_resp.expire as u64).saturating_mul(1000));
- let _ = channel_host::workspace_write(TOKEN_PATH, &data.tenant_access_token);
+ let _ = channel_host::workspace_write(TOKEN_PATH, &token_resp.tenant_access_token);
let _ = channel_host::workspace_write(TOKEN_EXPIRY_PATH, &expiry.to_string());
channel_host::log(
channel_host::LogLevel::Debug,
- &format!("Tenant access token refreshed, expires in {}s", data.expire),
+ &format!(
+ "Tenant access token refreshed, expires in {}s",
+ token_resp.expire
+ ),
);
- Ok(data.tenant_access_token)
+ Ok(token_resp.tenant_access_token)
}
Err(e) => Err(format!("Token exchange request failed: {}", e)),
}
@@ -819,3 +836,60 @@ fn json_response(status: u16, body: serde_json::Value) -> OutgoingHttpResponse {
body: body_bytes,
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parse_flat_token_response() {
+ let json = r#"{
+ "code": 0,
+ "msg": "ok",
+ "tenant_access_token": "t-abc123",
+ "expire": 7200
+ }"#;
+ let resp: TenantAccessTokenResponse = serde_json::from_str(json).unwrap();
+ assert_eq!(resp.code, 0);
+ assert_eq!(resp.msg, "ok");
+ assert_eq!(resp.tenant_access_token, "t-abc123");
+ assert_eq!(resp.expire, 7200);
+ }
+
+ #[test]
+ fn parse_token_response_rejects_missing_token() {
+ let json = r#"{"code": 0, "msg": "ok", "expire": 7200}"#;
+ let result: Result = serde_json::from_str(json);
+ assert!(result.is_err(), "should fail when tenant_access_token is missing");
+ }
+
+ #[test]
+ fn parse_token_response_rejects_missing_expire() {
+ let json = r#"{"code": 0, "msg": "ok", "tenant_access_token": "t-abc"}"#;
+ let result: Result = serde_json::from_str(json);
+ assert!(result.is_err(), "should fail when expire is missing");
+ }
+
+ #[test]
+ fn parse_token_response_defaults_code_and_msg() {
+ let json = r#"{"tenant_access_token": "t-abc", "expire": 3600}"#;
+ let resp: TenantAccessTokenResponse = serde_json::from_str(json).unwrap();
+ assert_eq!(resp.code, 0);
+ assert_eq!(resp.msg, "");
+ assert_eq!(resp.tenant_access_token, "t-abc");
+ assert_eq!(resp.expire, 3600);
+ }
+
+ #[test]
+ fn parse_token_error_response() {
+ let json = r#"{
+ "code": 10003,
+ "msg": "invalid app_id",
+ "tenant_access_token": "",
+ "expire": 0
+ }"#;
+ let resp: TenantAccessTokenResponse = serde_json::from_str(json).unwrap();
+ assert_eq!(resp.code, 10003);
+ assert!(resp.tenant_access_token.is_empty());
+ }
+}
diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs
index a095ccb3..f34ed68a 100644
--- a/channels-src/telegram/src/lib.rs
+++ b/channels-src/telegram/src/lib.rs
@@ -360,6 +360,8 @@ enum TelegramStatusAction {
}
const TELEGRAM_STATUS_MAX_CHARS: usize = 600;
+/// Telegram's hard limit for message text length.
+const TELEGRAM_MAX_MESSAGE_LEN: usize = 4096;
fn truncate_status_message(input: &str, max_chars: usize) -> String {
let mut iter = input.chars();
@@ -371,6 +373,73 @@ fn truncate_status_message(input: &str, max_chars: usize) -> String {
}
}
+/// Split a long message into chunks that fit within Telegram's 4096-char limit.
+///
+/// Tries to split at the most natural boundary available (in priority order):
+/// 1. Double newline (paragraph break)
+/// 2. Single newline
+/// 3. Sentence end (`. `, `! `, `? `)
+/// 4. Word boundary (space)
+/// 5. Hard cut at the limit (last resort for pathological input)
+fn split_message(text: &str) -> Vec {
+ if text.chars().count() <= TELEGRAM_MAX_MESSAGE_LEN {
+ return vec![text.to_string()];
+ }
+
+ let mut chunks: Vec = Vec::new();
+ let mut remaining = text;
+
+ while !remaining.is_empty() {
+ // Count chars to find the byte offset for our window.
+ let window_bytes = remaining
+ .char_indices()
+ .take(TELEGRAM_MAX_MESSAGE_LEN)
+ .last()
+ .map(|(byte_idx, ch)| byte_idx + ch.len_utf8())
+ .unwrap_or(remaining.len());
+
+ if window_bytes >= remaining.len() {
+ // Remainder fits entirely.
+ chunks.push(remaining.to_string());
+ break;
+ }
+
+ let window = &remaining[..window_bytes];
+
+ // 1. Double newline — best paragraph boundary
+ let split_at = window.rfind("\n\n")
+ // 2. Single newline
+ .or_else(|| window.rfind('\n'))
+ // 3. Sentence-ending punctuation followed by space.
+ // Note: this only detects ASCII punctuation (. ! ?), not CJK
+ // sentence-ending marks (。!?). CJK text falls through to
+ // word-boundary or hard-cut splitting.
+ .or_else(|| {
+ let bytes = window.as_bytes();
+ // Search backwards for '. ', '! ', '? '
+ (1..bytes.len()).rev().find(|&i| {
+ matches!(bytes[i - 1], b'.' | b'!' | b'?') && bytes[i] == b' '
+ })
+ })
+ // 4. Word boundary (last space)
+ .or_else(|| window.rfind(' '))
+ // 5. Hard cut
+ .unwrap_or(window_bytes);
+
+ // Avoid empty chunks (e.g. text starting with \n\n).
+ let split_at = if split_at == 0 { window_bytes } else { split_at };
+
+ // Trim whitespace at chunk boundaries for clean Telegram display.
+ // Note: this drops leading/trailing spaces at split points, which is
+ // acceptable for chat messages but means the concatenation of chunks
+ // may not exactly equal the original text when split at spaces.
+ chunks.push(remaining[..split_at].trim_end().to_string());
+ remaining = remaining[split_at..].trim_start();
+ }
+
+ chunks
+}
+
fn status_message_for_user(update: &StatusUpdate) -> Option {
let message = update.message.trim();
if message.is_empty() {
@@ -1242,26 +1311,64 @@ fn send_response(
return Ok(());
}
- // Try Markdown, fall back to plain text on parse errors
- match send_message(
- chat_id,
- &response.content,
- reply_to_message_id,
- Some("Markdown"),
- message_thread_id,
- ) {
- Ok(_) => Ok(()),
- Err(SendError::ParseEntities(_)) => send_message(
- chat_id,
- &response.content,
- reply_to_message_id,
- None,
- message_thread_id,
- )
- .map(|_| ())
- .map_err(|e| format!("Plain-text retry also failed: {}", e)),
- Err(e) => Err(e.to_string()),
+ // Split large messages into chunks that fit Telegram's limit.
+ let chunks = split_message(&response.content);
+ let total = chunks.len();
+
+ // The first chunk replies to the original message; subsequent chunks
+ // reply to the previously sent chunk so they form a visual thread.
+ let mut reply_to = reply_to_message_id;
+
+ for (i, chunk) in chunks.into_iter().enumerate() {
+ // Try Markdown, fall back to plain text on parse errors
+ let result = send_message(chat_id, &chunk, reply_to, Some("Markdown"), message_thread_id);
+
+ let msg_id = match result {
+ Ok(id) => {
+ channel_host::log(
+ channel_host::LogLevel::Debug,
+ &format!(
+ "Sent message chunk {}/{} to chat {}: message_id={}",
+ i + 1,
+ total,
+ chat_id,
+ id,
+ ),
+ );
+ id
+ }
+ Err(SendError::ParseEntities(detail)) => {
+ channel_host::log(
+ channel_host::LogLevel::Warn,
+ &format!(
+ "Markdown parse failed on chunk {}/{} ({}), retrying as plain text",
+ i + 1,
+ total,
+ detail
+ ),
+ );
+ let id = send_message(chat_id, &chunk, reply_to, None, message_thread_id)
+ .map_err(|e| format!("Plain-text retry also failed: {}", e))?;
+ channel_host::log(
+ channel_host::LogLevel::Debug,
+ &format!(
+ "Sent plain-text chunk {}/{} to chat {}: message_id={}",
+ i + 1,
+ total,
+ chat_id,
+ id,
+ ),
+ );
+ id
+ }
+ Err(e) => return Err(e.to_string()),
+ };
+
+ // Each subsequent chunk threads off the previous sent message.
+ reply_to = Some(msg_id);
}
+
+ Ok(())
}
/// Send a single attachment, choosing sendPhoto or sendDocument based on MIME type.
@@ -2043,6 +2150,102 @@ export!(TelegramChannel);
mod tests {
use super::*;
+ #[test]
+ fn test_split_message_short() {
+ let text = "Hello, world!";
+ let chunks = split_message(text);
+ assert_eq!(chunks, vec![text]);
+ }
+
+ #[test]
+ fn test_split_message_paragraph_boundary() {
+ let para_a = "A".repeat(3000);
+ let para_b = "B".repeat(3000);
+ let text = format!("{}\n\n{}", para_a, para_b);
+ let chunks = split_message(&text);
+ assert_eq!(chunks.len(), 2);
+ assert_eq!(chunks[0], para_a);
+ assert_eq!(chunks[1], para_b);
+ }
+
+ #[test]
+ fn test_split_message_word_boundary() {
+ // Build a string well over the limit with no newlines.
+ let words: Vec = (0..1000).map(|i| format!("word{:04}", i)).collect();
+ let text = words.join(" ");
+ assert!(text.len() > TELEGRAM_MAX_MESSAGE_LEN);
+ let chunks = split_message(&text);
+ assert!(chunks.len() > 1, "expected multiple chunks");
+ for chunk in &chunks {
+ assert!(chunk.chars().count() <= TELEGRAM_MAX_MESSAGE_LEN);
+ }
+ // Rejoined chunks must equal the original text exactly.
+ let rejoined = chunks.join(" ");
+ assert_eq!(rejoined, text);
+ }
+
+ #[test]
+ fn test_split_message_each_chunk_fits() {
+ // Stress-test: 20 000 chars of mixed text.
+ let text: String = (0..500)
+ .map(|i| format!("Sentence number {}. ", i))
+ .collect();
+ assert!(text.len() > TELEGRAM_MAX_MESSAGE_LEN);
+ let chunks = split_message(&text);
+ for chunk in &chunks {
+ assert!(chunk.chars().count() <= TELEGRAM_MAX_MESSAGE_LEN);
+ }
+ }
+
+ #[test]
+ fn test_split_message_sentence_boundary() {
+ // Build text that exceeds the limit, with sentence boundaries inside.
+ let sentence = "This is a test sentence. ";
+ let repeat_count = TELEGRAM_MAX_MESSAGE_LEN / sentence.len() + 5;
+ let text: String = sentence.repeat(repeat_count);
+ assert!(text.chars().count() > TELEGRAM_MAX_MESSAGE_LEN);
+
+ let chunks = split_message(&text);
+ assert!(chunks.len() > 1);
+ // First chunk should end at a sentence boundary (trimmed)
+ let first = &chunks[0];
+ assert!(
+ first.ends_with('.'),
+ "First chunk should end at a sentence boundary, got: ...{}",
+ &first[first.len().saturating_sub(20)..]
+ );
+ }
+
+ #[test]
+ fn test_split_message_hard_cut_no_spaces() {
+ // Pathological input: a single huge "word" with no spaces or newlines.
+ let text = "x".repeat(TELEGRAM_MAX_MESSAGE_LEN * 2 + 100);
+ let chunks = split_message(&text);
+ assert!(chunks.len() >= 2);
+ for chunk in &chunks {
+ assert!(chunk.chars().count() <= TELEGRAM_MAX_MESSAGE_LEN);
+ }
+ // Rejoined must preserve all characters
+ let rejoined: String = chunks.concat();
+ assert_eq!(rejoined, text);
+ }
+
+ #[test]
+ fn test_split_message_multibyte_chars() {
+ // Emoji are 4 bytes each. Ensure we don't panic or split mid-character.
+ let emoji = "\u{1F600}"; // 😀
+ let text: String = emoji.repeat(TELEGRAM_MAX_MESSAGE_LEN + 100);
+ assert!(text.chars().count() > TELEGRAM_MAX_MESSAGE_LEN);
+
+ let chunks = split_message(&text);
+ assert!(chunks.len() >= 2);
+ for chunk in &chunks {
+ assert!(chunk.chars().count() <= TELEGRAM_MAX_MESSAGE_LEN);
+ // Every char should be a complete emoji
+ assert!(chunk.chars().all(|c| c == '\u{1F600}'));
+ }
+ }
+
#[test]
fn test_clean_message_text() {
// Without bot_username: strips any leading @mention
diff --git a/codecov.yml b/codecov.yml
index 3e31b00a..723c1175 100644
--- a/codecov.yml
+++ b/codecov.yml
@@ -2,9 +2,13 @@ coverage:
status:
project:
default:
- target: auto
- threshold: 1%
+ target: 80%
+ threshold: 2%
patch:
default:
- target: 80%
- threshold: 5%
\ No newline at end of file
+ target: 90%
+
+comment:
+ layout: "reach,diff,flags"
+ behavior: default
+ require_changes: true
diff --git a/crates/ironclaw_safety/src/policy.rs b/crates/ironclaw_safety/src/policy.rs
index f731d687..d1784b98 100644
--- a/crates/ironclaw_safety/src/policy.rs
+++ b/crates/ironclaw_safety/src/policy.rs
@@ -324,7 +324,7 @@ mod tests {
let violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"excessive_urls pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -349,7 +349,7 @@ mod tests {
let violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"obfuscated_string pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -370,7 +370,7 @@ mod tests {
let _violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"shell_injection pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -387,7 +387,7 @@ mod tests {
let _violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"sql_pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -405,7 +405,7 @@ mod tests {
let _violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"crypto_private_key pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -423,7 +423,7 @@ mod tests {
let _violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"system_file_access pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
@@ -441,7 +441,7 @@ mod tests {
let _violations = policy.check(&payload);
let elapsed = start.elapsed();
assert!(
- elapsed.as_millis() < 100,
+ elapsed.as_millis() < 500,
"encoded_exploit pattern took {}ms on 100KB near-miss",
elapsed.as_millis()
);
diff --git a/docs/LLM_PROVIDERS.md b/docs/LLM_PROVIDERS.md
index 02016890..793f0682 100644
--- a/docs/LLM_PROVIDERS.md
+++ b/docs/LLM_PROVIDERS.md
@@ -15,7 +15,7 @@ the most common configurations.
| io.net | `ionet` | `IONET_API_KEY` | Intelligence API |
| Mistral | `mistral` | `MISTRAL_API_KEY` | Mistral models |
| Yandex AI Studio | `yandex` | `YANDEX_API_KEY` | YandexGPT models |
-| MiniMax | `minimax` | `MINIMAX_API_KEY` | MiniMax-M2.5 models |
+| MiniMax | `minimax` | `MINIMAX_API_KEY` | MiniMax-M2.7 models |
| Cloudflare Workers AI | `cloudflare` | `CLOUDFLARE_API_KEY` | Access to Workers AI |
| Ollama | `ollama` | No | Local inference |
| AWS Bedrock | `bedrock` | AWS credentials | Native Converse API |
@@ -129,7 +129,7 @@ LLM_BACKEND=minimax
MINIMAX_API_KEY=...
```
-Available models: `MiniMax-M2.5` (default), `MiniMax-M2.5-highspeed`
+Available models: `MiniMax-M2.7` (default), `MiniMax-M2.7-highspeed`, `MiniMax-M2.5`, `MiniMax-M2.5-highspeed`
To use the China mainland endpoint, set:
diff --git a/providers.json b/providers.json
index 12723a6f..550edd64 100644
--- a/providers.json
+++ b/providers.json
@@ -393,8 +393,8 @@
"api_key_required": true,
"base_url_env": "MINIMAX_BASE_URL",
"model_env": "MINIMAX_MODEL",
- "default_model": "MiniMax-M2.5",
- "description": "MiniMax API (MiniMax-M2.5 and MiniMax-M2.5-highspeed models)",
+ "default_model": "MiniMax-M2.7",
+ "description": "MiniMax API (MiniMax-M2.7, MiniMax-M2.7-highspeed, MiniMax-M2.5 and MiniMax-M2.5-highspeed models)",
"setup": {
"kind": "api_key",
"secret_name": "llm_minimax_api_key",
diff --git a/registry/channels/telegram.json b/registry/channels/telegram.json
index bd07208f..85d793ed 100644
--- a/registry/channels/telegram.json
+++ b/registry/channels/telegram.json
@@ -2,7 +2,7 @@
"name": "telegram",
"display_name": "Telegram Channel",
"kind": "channel",
- "version": "0.2.4",
+ "version": "0.2.5",
"wit_version": "0.3.0",
"description": "Talk to your agent through a Telegram bot",
"keywords": [
diff --git a/skills/ironclaw-workflow-orchestrator/references/workflow-routines.md b/skills/ironclaw-workflow-orchestrator/references/workflow-routines.md
index 8afa857d..5e64a2b2 100644
--- a/skills/ironclaw-workflow-orchestrator/references/workflow-routines.md
+++ b/skills/ironclaw-workflow-orchestrator/references/workflow-routines.md
@@ -8,15 +8,21 @@ Replace `{{...}}` placeholders before use.
{
"name": "wf-issue-plan",
"description": "Create implementation plan when a new issue arrives",
- "trigger_type": "system_event",
- "event_source": "github",
- "event_type": "issue.opened",
- "event_filters": {
- "repository_name": "{{repository}}"
- },
- "action_type": "full_job",
"prompt": "For issue #{{issue_number}} in {{repository}}, produce a concrete implementation plan with milestones, edge cases, and tests. Post/update an issue comment with the plan.",
- "cooldown_secs": 30
+ "request": {
+ "kind": "system_event",
+ "source": "github",
+ "event_type": "issue.opened",
+ "filters": {
+ "repository_name": "{{repository}}"
+ }
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 30
+ }
}
```
@@ -28,16 +34,22 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
{
"name": "wf-maintainer-comment-gate-{{maintainer}}",
"description": "React to maintainer guidance comments on issues/PRs",
- "trigger_type": "system_event",
- "event_source": "github",
- "event_type": "pr.comment.created",
- "event_filters": {
- "repository_name": "{{repository}}",
- "comment_author": "{{maintainer}}"
- },
- "action_type": "full_job",
"prompt": "Read the maintainer comment and decide: update plan or start/continue implementation. If plan changes are requested, edit the plan artifact first. If implementation is requested, continue on the feature branch and update PR status/comment.",
- "cooldown_secs": 20
+ "request": {
+ "kind": "system_event",
+ "source": "github",
+ "event_type": "pr.comment.created",
+ "filters": {
+ "repository_name": "{{repository}}",
+ "comment_author": "{{maintainer}}"
+ }
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 20
+ }
}
```
@@ -47,15 +59,21 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
{
"name": "wf-pr-monitor-loop",
"description": "Keep PR healthy: address review comments and refresh branch",
- "trigger_type": "system_event",
- "event_source": "github",
- "event_type": "pr.synchronize",
- "event_filters": {
- "repository_name": "{{repository}}"
- },
- "action_type": "full_job",
"prompt": "For PR #{{pr_number}}, collect open review comments and unresolved threads, apply fixes, push branch updates, and summarize remaining blockers. If conflict with {{main_branch}}, rebase/merge from origin/{{main_branch}} and resolve safely.",
- "cooldown_secs": 20
+ "request": {
+ "kind": "system_event",
+ "source": "github",
+ "event_type": "pr.synchronize",
+ "filters": {
+ "repository_name": "{{repository}}"
+ }
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 20
+ }
}
```
@@ -65,16 +83,22 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
{
"name": "wf-ci-fix-loop",
"description": "Fix failing CI checks on active PRs",
- "trigger_type": "system_event",
- "event_source": "github",
- "event_type": "ci.check_run.completed",
- "event_filters": {
- "repository_name": "{{repository}}",
- "ci_conclusion": "failure"
- },
- "action_type": "full_job",
"prompt": "Find failing check details for PR #{{pr_number}}, implement minimal safe fixes, rerun or await CI, and post concise status updates. Prioritize deterministic and test-backed fixes.",
- "cooldown_secs": 20
+ "request": {
+ "kind": "system_event",
+ "source": "github",
+ "event_type": "ci.check_run.completed",
+ "filters": {
+ "repository_name": "{{repository}}",
+ "ci_conclusion": "failure"
+ }
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 20
+ }
}
```
@@ -84,11 +108,17 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
{
"name": "wf-staging-batch-review",
"description": "Batch correctness review through staging, then merge to main",
- "trigger_type": "cron",
- "schedule": "0 0 */{{batch_interval_hours}} * * *",
- "action_type": "full_job",
"prompt": "Every cycle: list ready PRs, merge ready ones into {{staging_branch}}, run deep correctness analysis in batch, fix discovered issues on affected branches, ensure CI green, then merge {{staging_branch}} into {{main_branch}} if clean.",
- "cooldown_secs": 120
+ "request": {
+ "kind": "cron",
+ "schedule": "0 0 */{{batch_interval_hours}} * * *"
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 120
+ }
}
```
@@ -98,16 +128,22 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
{
"name": "wf-learning-memory",
"description": "Capture merge learnings into shared memory",
- "trigger_type": "system_event",
- "event_source": "github",
- "event_type": "pr.closed",
- "event_filters": {
- "repository_name": "{{repository}}",
- "pr_merged": "true"
- },
- "action_type": "full_job",
"prompt": "From merged PR #{{pr_number}}, extract preventable mistakes, reviewer themes, CI failure causes, and successful patterns. Write/update a shared memory doc with actionable rules to reduce cycle time and regressions.",
- "cooldown_secs": 30
+ "request": {
+ "kind": "system_event",
+ "source": "github",
+ "event_type": "pr.closed",
+ "filters": {
+ "repository_name": "{{repository}}",
+ "pr_merged": "true"
+ }
+ },
+ "execution": {
+ "mode": "full_job"
+ },
+ "advanced": {
+ "cooldown_secs": 30
+ }
}
```
@@ -115,7 +151,7 @@ Trigger per-maintainer by creating one routine per handle, or maintain a shared
```json
{
- "source": "github",
+ "event_source": "github",
"event_type": "issue.opened",
"payload": {
"repository_name": "{{repository}}",
diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs
index 83d971ef..1780ba9d 100644
--- a/src/agent/agent_loop.rs
+++ b/src/agent/agent_loop.rs
@@ -146,6 +146,8 @@ pub struct AgentDeps {
pub transcription: Option>,
/// Document text extraction middleware for PDF, DOCX, PPTX, etc.
pub document_extraction: Option>,
+ /// Software builder for self-repair tool rebuilding.
+ pub builder: Option>,
}
/// The main agent that coordinates all components.
@@ -161,9 +163,10 @@ pub struct Agent {
pub(super) heartbeat_config: Option,
pub(super) hygiene_config: Option,
pub(super) routine_config: Option,
- /// Optional slot to expose the routine engine to the gateway for manual triggering.
+ /// Shared routine-engine slot used for internal event matching and for exposing
+ /// the engine to gateway/manual trigger entry points.
pub(super) routine_engine_slot:
- Option>>>>,
+ Arc>>>,
}
impl Agent {
@@ -228,16 +231,21 @@ impl Agent {
heartbeat_config,
hygiene_config,
routine_config,
- routine_engine_slot: None,
+ routine_engine_slot: Arc::new(tokio::sync::RwLock::new(None)),
}
}
- /// Set the routine engine slot for exposing the engine to the gateway.
+ /// Replace the routine-engine slot with a shared one so the gateway and
+ /// agent reference the same engine.
pub fn set_routine_engine_slot(
&mut self,
slot: Arc>>>,
) {
- self.routine_engine_slot = Some(slot);
+ self.routine_engine_slot = slot;
+ }
+
+ async fn routine_engine(&self) -> Option> {
+ self.routine_engine_slot.read().await.clone()
}
// Convenience accessors
@@ -334,11 +342,18 @@ impl Agent {
let mut message_stream = self.channels.start_all().await?;
// Start self-repair task with notification forwarding
- let repair = Arc::new(DefaultSelfRepair::new(
+ let mut self_repair = DefaultSelfRepair::new(
self.context_manager.clone(),
self.config.stuck_threshold,
self.config.max_repair_attempts,
- ));
+ );
+ if let Some(ref store) = self.deps.store {
+ self_repair = self_repair.with_store(Arc::clone(store));
+ }
+ if let Some(ref builder) = self.deps.builder {
+ self_repair = self_repair.with_builder(Arc::clone(builder), Arc::clone(self.tools()));
+ }
+ let repair = Arc::new(self_repair);
let repair_interval = self.config.repair_check_interval;
let repair_channels = self.channels.clone();
let repair_owner_id = self.owner_id().to_string();
@@ -633,9 +648,7 @@ impl Agent {
// via a local to use in the message loop below.
// Expose engine to gateway for manual triggering
- if let Some(ref slot) = self.routine_engine_slot {
- *slot.write().await = Some(Arc::clone(&engine));
- }
+ *self.routine_engine_slot.write().await = Some(Arc::clone(&engine));
tracing::debug!(
"Routines enabled: cron ticker every {}s, max {} concurrent",
@@ -655,9 +668,6 @@ impl Agent {
None
};
- // Extract engine ref for use in message loop
- let routine_engine_for_loop = routine_handle.as_ref().map(|(_, e)| Arc::clone(e));
-
// Main message loop
tracing::debug!("Agent {} ready and listening", self.config.name);
@@ -693,29 +703,6 @@ 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
@@ -1032,6 +1019,24 @@ impl Agent {
message.content.len()
);
+ if !message.is_internal
+ && let Submission::UserInput { ref content } = submission
+ && let Some(engine) = self.routine_engine().await
+ {
+ let fired = engine
+ .check_event_triggers(&message.user_id, &message.channel, content)
+ .await;
+ if fired > 0 {
+ tracing::debug!(
+ channel = %message.channel,
+ user = %message.user_id,
+ fired,
+ "Consumed inbound user message with matching event-triggered routine(s)"
+ );
+ return Ok(Some(String::new()));
+ }
+ }
+
// Process based on submission type
let result = match submission {
Submission::UserInput { content } => {
diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs
index 9be0d654..d3825b2f 100644
--- a/src/agent/dispatcher.rs
+++ b/src/agent/dispatcher.rs
@@ -29,7 +29,7 @@ pub(super) enum AgenticLoopResult {
/// A tool requires approval before continuing.
NeedApproval {
/// The pending approval request to store.
- pending: PendingApproval,
+ pending: Box,
},
}
@@ -217,9 +217,7 @@ impl Agent {
reason: format!("Exceeded maximum tool iterations ({max_tool_iterations})"),
}
.into()),
- LoopOutcome::NeedApproval(pending) => {
- Ok(AgenticLoopResult::NeedApproval { pending: *pending })
- }
+ LoopOutcome::NeedApproval(pending) => Ok(AgenticLoopResult::NeedApproval { pending }),
}
}
@@ -482,6 +480,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
usize,
crate::llm::ToolCall,
Arc,
+ bool, // allow_always
)> = None;
for (idx, original_tc) in tool_calls.iter().enumerate() {
@@ -551,7 +550,8 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
&& let Some(tool) = tool_opt
{
use crate::tools::ApprovalRequirement;
- let needs_approval = match tool.requires_approval(&tc.arguments) {
+ let requirement = tool.requires_approval(&tc.arguments);
+ let needs_approval = match requirement {
ApprovalRequirement::Never => false,
ApprovalRequirement::UnlessAutoApproved => {
let sess = self.session.lock().await;
@@ -586,7 +586,8 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
continue;
}
- approval_needed = Some((idx, tc, tool));
+ let allow_always = !matches!(requirement, ApprovalRequirement::Always);
+ approval_needed = Some((idx, tc, tool, allow_always));
break;
}
}
@@ -887,7 +888,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
}
// Handle approval if a tool needed it
- if let Some((approval_idx, tc, tool)) = approval_needed {
+ if let Some((approval_idx, tc, tool, allow_always)) = approval_needed {
let display_params = redact_params(&tc.arguments, tool.sensitive_params());
let pending = PendingApproval {
request_id: Uuid::new_v4(),
@@ -899,6 +900,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
context_messages: reason_ctx.messages.clone(),
deferred_tool_calls: tool_calls[approval_idx + 1..].to_vec(),
user_timezone: Some(self.user_tz.name().to_string()),
+ allow_always,
};
return Ok(Some(LoopOutcome::NeedApproval(Box::new(pending))));
@@ -1197,6 +1199,7 @@ mod tests {
http_interceptor: None,
transcription: None,
document_extraction: None,
+ builder: None,
};
Agent::new(
@@ -1364,6 +1367,35 @@ mod tests {
assert!(always_needs, "Always must always require approval");
}
+ /// Regression test: `allow_always` must be `false` for `Always` and
+ /// `true` for `UnlessAutoApproved`, so the UI hides the "always" button
+ /// for tools that truly cannot be auto-approved.
+ #[test]
+ fn test_allow_always_matches_approval_requirement() {
+ use crate::tools::ApprovalRequirement;
+
+ // Mirrors the expression used in dispatcher.rs and thread_ops.rs:
+ // let allow_always = !matches!(requirement, ApprovalRequirement::Always);
+
+ // UnlessAutoApproved → allow_always = true
+ let req = ApprovalRequirement::UnlessAutoApproved;
+ let allow_always = !matches!(req, ApprovalRequirement::Always);
+ assert!(
+ allow_always,
+ "UnlessAutoApproved should set allow_always = true"
+ );
+
+ // Always → allow_always = false
+ let req = ApprovalRequirement::Always;
+ let allow_always = !matches!(req, ApprovalRequirement::Always);
+ assert!(!allow_always, "Always should set allow_always = false");
+
+ // Never → allow_always = true (approval is never needed, but if it were, always would be ok)
+ let req = ApprovalRequirement::Never;
+ let allow_always = !matches!(req, ApprovalRequirement::Always);
+ assert!(allow_always, "Never should set allow_always = true");
+ }
+
#[test]
fn test_pending_approval_serialization_backcompat_without_deferred_calls() {
// PendingApproval from before the deferred_tool_calls field was added
@@ -1409,6 +1441,7 @@ mod tests {
},
],
user_timezone: None,
+ allow_always: true,
};
let json = serde_json::to_string(&pending).expect("serialize");
@@ -2037,6 +2070,7 @@ mod tests {
http_interceptor: None,
transcription: None,
document_extraction: None,
+ builder: None,
};
Agent::new(
@@ -2155,6 +2189,7 @@ mod tests {
http_interceptor: None,
transcription: None,
document_extraction: None,
+ builder: None,
};
Agent::new(
diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs
index 519f16c2..2487ac05 100644
--- a/src/agent/routine_engine.rs
+++ b/src/agent/routine_engine.rs
@@ -10,6 +10,7 @@
//! Lightweight routines execute inline (single LLM call, no scheduler slot).
//! Full-job routines are delegated to the existing `Scheduler`.
+use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
@@ -23,19 +24,19 @@ use crate::agent::Scheduler;
use crate::agent::routine::{
NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire,
};
-use crate::channels::{IncomingMessage, OutgoingResponse};
+use crate::channels::OutgoingResponse;
use crate::config::RoutineConfig;
-use crate::context::JobContext;
+use crate::context::{JobContext, JobState};
use crate::db::Database;
use crate::error::RoutineError;
use crate::llm::{
ChatMessage, CompletionRequest, FinishReason, LlmProvider, ToolCall, ToolCompletionRequest,
};
-use crate::safety::SafetyLayer;
use crate::tools::{
ApprovalContext, ApprovalRequirement, ToolError, ToolRegistry, prepare_tool_params,
};
use crate::workspace::Workspace;
+use ironclaw_safety::SafetyLayer;
enum EventMatcher {
Message { routine: Routine, regex: Regex },
@@ -60,6 +61,10 @@ pub struct RoutineEngine {
tools: Arc,
/// Safety layer for tool output sanitization.
safety: Arc,
+ /// Timestamp when this engine instance was created. Used by
+ /// `sync_dispatched_runs` to distinguish orphaned runs (from a previous
+ /// process) from actively-watched runs (from this process).
+ boot_time: chrono::DateTime,
}
impl RoutineEngine {
@@ -85,9 +90,16 @@ impl RoutineEngine {
scheduler,
tools,
safety,
+ boot_time: Utc::now(),
}
}
+ /// Expose the running count for integration tests.
+ #[doc(hidden)]
+ pub fn running_count_for_test(&self) -> &Arc {
+ &self.running_count
+ }
+
/// Refresh the in-memory event trigger cache from DB.
pub async fn refresh_event_cache(&self) {
match self.store.list_event_routines().await {
@@ -135,10 +147,19 @@ impl RoutineEngine {
/// Check incoming message against event triggers. Returns number of routines fired.
///
- /// Called synchronously from the main loop after handle_message(). The actual
- /// execution is spawned async so this returns quickly.
- pub async fn check_event_triggers(&self, message: &IncomingMessage) -> usize {
+ /// Accepts only the three fields needed for matching (user scope, channel,
+ /// message content) so callers never need to clone a full `IncomingMessage`.
+ pub async fn check_event_triggers(&self, user_id: &str, channel: &str, content: &str) -> usize {
let cache = self.event_cache.read().await;
+
+ // Early return if there are no message matchers at all.
+ if !cache
+ .iter()
+ .any(|m| matches!(m, EventMatcher::Message { .. }))
+ {
+ return 0;
+ }
+
let mut fired = 0;
// Collect routine IDs for batch query
@@ -155,16 +176,9 @@ impl RoutineEngine {
}
// Single batch query instead of N queries
- let concurrent_counts = match self
- .store
- .count_running_routine_runs_batch(&routine_ids)
- .await
- {
- Ok(counts) => counts,
- Err(e) => {
- tracing::error!("Failed to batch-load concurrent counts: {}", e);
- return 0;
- }
+ let concurrent_counts = match self.batch_concurrent_counts(&routine_ids).await {
+ Some(counts) => counts,
+ None => return 0,
};
for matcher in cache.iter() {
@@ -173,7 +187,7 @@ impl RoutineEngine {
EventMatcher::System { .. } => continue,
};
- if routine.user_id != message.user_id {
+ if routine.user_id != user_id {
continue;
}
@@ -181,13 +195,13 @@ impl RoutineEngine {
if let Trigger::Event {
channel: Some(ch), ..
} = &routine.trigger
- && ch != &message.channel
+ && ch != channel
{
continue;
}
// Regex match
- if !re.is_match(&message.content) {
+ if !re.is_match(content) {
continue;
}
@@ -210,7 +224,7 @@ impl RoutineEngine {
continue;
}
- let detail = truncate(&message.content, 200);
+ let detail = truncate(content, 200);
self.spawn_fire(routine.clone(), "event", Some(detail));
fired += 1;
}
@@ -229,6 +243,15 @@ impl RoutineEngine {
user_id: Option<&str>,
) -> usize {
let cache = self.event_cache.read().await;
+
+ // Early return if there are no system-event matchers at all.
+ if !cache
+ .iter()
+ .any(|m| matches!(m, EventMatcher::System { .. }))
+ {
+ return 0;
+ }
+
let mut fired = 0;
// Collect routine IDs for batch query
@@ -245,19 +268,9 @@ impl RoutineEngine {
}
// Single batch query instead of N queries
- let concurrent_counts = match self
- .store
- .count_running_routine_runs_batch(&routine_ids)
- .await
- {
- Ok(counts) => counts,
- Err(e) => {
- tracing::error!(
- "Failed to batch-load concurrent counts for system events: {}",
- e
- );
- return 0;
- }
+ let concurrent_counts = match self.batch_concurrent_counts(&routine_ids).await {
+ Some(counts) => counts,
+ None => return 0,
};
for matcher in cache.iter() {
@@ -331,6 +344,23 @@ impl RoutineEngine {
fired
}
+ /// Batch-load concurrent run counts for a set of routine IDs.
+ ///
+ /// Returns `None` on database error (already logged).
+ async fn batch_concurrent_counts(&self, routine_ids: &[Uuid]) -> Option> {
+ match self
+ .store
+ .count_running_routine_runs_batch(routine_ids)
+ .await
+ {
+ Ok(counts) => Some(counts),
+ Err(e) => {
+ tracing::error!("Failed to batch-load concurrent counts: {}", e);
+ None
+ }
+ }
+ }
+
/// Check all due cron routines and fire them. Called by the cron ticker.
pub async fn check_cron_triggers(&self) {
let routines = match self.store.list_due_cron_routines().await {
@@ -365,6 +395,230 @@ impl RoutineEngine {
}
}
+ /// Reconcile orphaned full_job routine runs with their linked job outcomes.
+ ///
+ /// Called on each cron tick. Finds routine runs that are still `running`
+ /// with a linked `job_id`, checks the job state, and finalizes the run
+ /// when the job reaches a completed or terminal state.
+ ///
+ /// Only processes runs started **before** this engine's boot time, so it
+ /// never races with `FullJobWatcher` instances from the current process.
+ /// This makes it safe to call on every tick as a crash-recovery mechanism.
+ pub async fn sync_dispatched_runs(&self) {
+ let runs = match self.store.list_dispatched_routine_runs().await {
+ Ok(r) => r,
+ Err(e) => {
+ tracing::error!("Failed to list dispatched routine runs: {}", e);
+ return;
+ }
+ };
+
+ // Only process runs from a previous process instance. Runs started
+ // after boot_time are actively watched by a FullJobWatcher in this
+ // process and should not be finalized here.
+ let orphaned: Vec<_> = runs
+ .into_iter()
+ .filter(|r| r.started_at < self.boot_time)
+ .collect();
+
+ if orphaned.is_empty() {
+ return;
+ }
+
+ tracing::info!(
+ "Recovering {} orphaned dispatched routine runs",
+ orphaned.len()
+ );
+
+ for run in orphaned {
+ let job_id = match run.job_id {
+ Some(id) => id,
+ None => continue, // Should not happen (query filters), but guard anyway
+ };
+
+ // Fetch the linked job
+ let job = match self.store.get_job(job_id).await {
+ Ok(Some(j)) => j,
+ Ok(None) => {
+ // Orphaned: job record was deleted or never persisted
+ tracing::warn!(
+ run_id = %run.id,
+ job_id = %job_id,
+ "Linked job not found, marking routine run as failed"
+ );
+ self.complete_dispatched_run(
+ &run,
+ RunStatus::Failed,
+ &format!("Linked job {job_id} not found (orphaned)"),
+ )
+ .await;
+ continue;
+ }
+ Err(e) => {
+ tracing::error!(
+ run_id = %run.id,
+ job_id = %job_id,
+ "Failed to fetch linked job: {}", e
+ );
+ continue;
+ }
+ };
+
+ // Map job state to final run status
+ let final_status = match job.state {
+ JobState::Completed | JobState::Submitted | JobState::Accepted => {
+ Some(RunStatus::Ok)
+ }
+ JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed),
+ // Pending, InProgress, Stuck — still running
+ _ => None,
+ };
+
+ let status = match final_status {
+ Some(s) => s,
+ None => continue, // Job still active, check again next tick
+ };
+
+ // Build summary
+ let summary = if status == RunStatus::Failed {
+ match self.store.get_agent_job_failure_reason(job_id).await {
+ Ok(Some(reason)) => format!("Job {job_id} failed: {reason}"),
+ _ => format!("Job {job_id} {}", job.state),
+ }
+ } else {
+ format!("Job {job_id} completed successfully")
+ };
+
+ self.complete_dispatched_run(&run, status, &summary).await;
+ }
+ }
+
+ /// Finalize a dispatched routine run: update DB, update routine runtime,
+ /// persist to conversation thread, and send notification.
+ async fn complete_dispatched_run(&self, run: &RoutineRun, status: RunStatus, summary: &str) {
+ // Complete the run record in DB
+ if let Err(e) = self
+ .store
+ .complete_routine_run(run.id, status, Some(summary), None)
+ .await
+ {
+ tracing::error!(
+ run_id = %run.id,
+ "Failed to complete dispatched routine run: {}", e
+ );
+ return;
+ }
+
+ tracing::info!(
+ run_id = %run.id,
+ status = %status,
+ "Finalized dispatched routine run"
+ );
+
+ // Load the routine to update consecutive_failures and send notification
+ let routine = match self.store.get_routine(run.routine_id).await {
+ Ok(Some(r)) => r,
+ Ok(None) => {
+ tracing::warn!(
+ run_id = %run.id,
+ routine_id = %run.routine_id,
+ "Routine not found for dispatched run finalization"
+ );
+ return;
+ }
+ Err(e) => {
+ tracing::error!(
+ run_id = %run.id,
+ "Failed to load routine for dispatched run: {}", e
+ );
+ return;
+ }
+ };
+
+ // Update runtime fields. In crash recovery, execute_routine() never
+ // reached its normal runtime update, so we must advance all fields here.
+ let new_failures = if status == RunStatus::Failed {
+ routine.consecutive_failures + 1
+ } else {
+ 0
+ };
+
+ let now = Utc::now();
+ let next_fire = if let Trigger::Cron {
+ ref schedule,
+ ref timezone,
+ } = routine.trigger
+ {
+ next_cron_fire(schedule, timezone.as_deref()).unwrap_or(None)
+ } else {
+ None
+ };
+
+ if let Err(e) = self
+ .store
+ .update_routine_runtime(
+ routine.id,
+ now,
+ next_fire,
+ routine.run_count + 1,
+ new_failures,
+ &routine.state,
+ )
+ .await
+ {
+ tracing::error!(
+ routine = %routine.name,
+ "Failed to update routine runtime after dispatched run: {}", e
+ );
+ }
+
+ // Persist result to the routine's conversation thread
+ let thread_id = match self
+ .store
+ .get_or_create_routine_conversation(routine.id, &routine.name, &routine.user_id)
+ .await
+ {
+ Ok(conv_id) => {
+ let msg = format!("[dispatched] {}: {}", status, summary);
+ if let Err(e) = self
+ .store
+ .add_conversation_message(conv_id, "assistant", &msg)
+ .await
+ {
+ tracing::error!(
+ routine = %routine.name,
+ "Failed to persist dispatched run message: {}", e
+ );
+ }
+ Some(conv_id.to_string())
+ }
+ Err(e) => {
+ tracing::error!(
+ routine = %routine.name,
+ "Failed to get routine conversation: {}", e
+ );
+ None
+ }
+ };
+
+ // Send notification
+ send_notification(
+ &self.notify_tx,
+ &routine.notify,
+ &routine.user_id,
+ &routine.name,
+ status,
+ Some(summary),
+ thread_id.as_deref(),
+ )
+ .await;
+
+ // Note: we do NOT decrement running_count here. In normal flow,
+ // execute_routine() handles that after FullJobWatcher returns.
+ // This sync path only runs for crash recovery (process restarted),
+ // where running_count was already reset to 0.
+ }
+
/// Fire a routine manually (from tool call or CLI).
///
/// Bypasses cooldown checks (those only apply to cron/event triggers).
@@ -508,6 +762,92 @@ impl RoutineEngine {
}
}
+/// Watches a dispatched full_job until the linked scheduler job completes.
+///
+/// Polls `store.get_job(job_id)` at a fixed interval until the job leaves
+/// an active state (Pending/InProgress/Stuck). Maps the final `JobState` to
+/// a `RunStatus` for the routine run.
+struct FullJobWatcher {
+ store: Arc,
+ job_id: Uuid,
+ routine_name: String,
+}
+
+impl FullJobWatcher {
+ /// Poll interval between DB checks.
+ const POLL_INTERVAL: Duration = Duration::from_secs(5);
+ /// Safety ceiling: 24 hours, derived from POLL_INTERVAL.
+ const MAX_POLLS: u32 = (24 * 60 * 60) / Self::POLL_INTERVAL.as_secs() as u32;
+
+ fn new(store: Arc, job_id: Uuid, routine_name: String) -> Self {
+ Self {
+ store,
+ job_id,
+ routine_name,
+ }
+ }
+
+ /// Block until the linked job finishes and return the mapped status + summary.
+ async fn wait_for_completion(&self) -> (RunStatus, Option) {
+ let mut polls = 0u32;
+
+ let final_status = loop {
+ // Check job state before sleeping so we finalize promptly
+ // if the job is already done (e.g. fast-failing jobs).
+ match self.store.get_job(self.job_id).await {
+ Ok(Some(job_ctx)) => {
+ // Use is_parallel_blocking (Pending/InProgress/Stuck) instead
+ // of is_active (!is_terminal) because routine jobs typically
+ // stop at Completed — which is NOT terminal but IS finished
+ // from an execution standpoint.
+ if !job_ctx.state.is_parallel_blocking() {
+ break Self::map_job_state(&job_ctx.state);
+ }
+ }
+ Ok(None) => {
+ tracing::warn!(
+ routine = %self.routine_name,
+ job_id = %self.job_id,
+ "full_job disappeared from DB while polling"
+ );
+ break RunStatus::Failed;
+ }
+ Err(e) => {
+ tracing::error!(
+ routine = %self.routine_name,
+ job_id = %self.job_id,
+ "Error polling full_job state: {}", e
+ );
+ break RunStatus::Failed;
+ }
+ }
+
+ polls += 1;
+ if polls >= Self::MAX_POLLS {
+ tracing::error!(
+ routine = %self.routine_name,
+ job_id = %self.job_id,
+ "full_job timed out after 24 hours, treating as failed"
+ );
+ break RunStatus::Failed;
+ }
+
+ tokio::time::sleep(Self::POLL_INTERVAL).await;
+ };
+
+ let summary = format!("Job {} finished ({})", self.job_id, final_status);
+ (final_status, Some(summary))
+ }
+
+ fn map_job_state(state: &crate::context::JobState) -> RunStatus {
+ use crate::context::JobState;
+ match state {
+ JobState::Failed | JobState::Cancelled => RunStatus::Failed,
+ _ => RunStatus::Ok, // Completed / Submitted / Accepted
+ }
+ }
+}
+
/// Shared context passed to the execution function.
struct EngineContext {
config: RoutineConfig,
@@ -682,8 +1022,10 @@ fn sanitize_routine_name(name: &str) -> String {
///
/// Fire-and-forget: creates a job via `Scheduler::dispatch_job` (which handles
/// creation, metadata, persistence, and scheduling), links the routine run to
-/// the job, and returns immediately. The job runs independently via the
-/// existing Worker/Scheduler with full tool access.
+/// the job, then watches it via `FullJobWatcher` until it reaches a
+/// non-active state (not Pending/InProgress/Stuck). Returns the final
+/// `RunStatus` mapped from the job outcome. This keeps the routine run
+/// active for the full job lifetime so concurrency guardrails apply.
async fn execute_full_job(
ctx: &EngineContext,
routine: &Routine,
@@ -726,25 +1068,30 @@ async fn execute_full_job(
reason: format!("failed to dispatch job: {e}"),
})?;
- // Link the routine run to the dispatched job
- if let Err(e) = ctx.store.link_routine_run_to_job(run.id, job_id).await {
- tracing::error!(
- routine = %routine.name,
- "Failed to link run to job: {}", e
- );
- }
+ // Link the routine run to the dispatched job.
+ // This MUST succeed — if it fails, sync_dispatched_runs() will never find
+ // this run (it filters on job_id IS NOT NULL), leaving it stuck as 'running'
+ // with running_count permanently elevated.
+ ctx.store
+ .link_routine_run_to_job(run.id, job_id)
+ .await
+ .map_err(|e| RoutineError::Database {
+ reason: format!("failed to link run to job: {e}"),
+ })?;
tracing::info!(
routine = %routine.name,
job_id = %job_id,
max_iterations = max_iterations,
- "Dispatched full job for routine"
+ "Dispatched full job for routine, watching for completion"
);
- let summary = format!(
- "Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations})"
- );
- Ok((RunStatus::Ok, Some(summary), None))
+ // Watch the job until it finishes — keeps the routine run active
+ // so concurrency guardrails (running_count, routine_runs status)
+ // remain enforced for the full job lifetime.
+ let watcher = FullJobWatcher::new(ctx.store.clone(), job_id, routine.name.clone());
+ let (status, summary) = watcher.wait_for_completion().await;
+ Ok((status, summary, None))
}
/// Execute a lightweight routine with optional tool support.
@@ -784,23 +1131,12 @@ async fn execute_lightweight(
Err(_) => None,
};
- // Build the user-facing prompt
- let mut full_prompt = String::new();
- full_prompt.push_str(prompt);
-
- if !context_parts.is_empty() {
- full_prompt.push_str("\n\n---\n\n# Context\n\n");
- full_prompt.push_str(&context_parts.join("\n\n"));
- }
-
- if let Some(state) = &state_content {
- full_prompt.push_str("\n\n---\n\n# Previous State\n\n");
- full_prompt.push_str(state);
- }
-
- full_prompt.push_str(
- "\n\n---\n\nIf nothing needs attention, reply EXACTLY with: ROUTINE_OK\n\
- If something needs attention, provide a concise summary.",
+ let full_prompt = build_lightweight_prompt(
+ prompt,
+ &context_parts,
+ state_content.as_deref(),
+ &routine.notify,
+ use_tools,
);
// Get system prompt
@@ -844,6 +1180,65 @@ async fn execute_lightweight(
}
}
+fn build_lightweight_prompt(
+ prompt: &str,
+ context_parts: &[String],
+ state_content: Option<&str>,
+ notify: &NotifyConfig,
+ use_tools: bool,
+) -> String {
+ let mut full_prompt = String::new();
+ full_prompt.push_str(prompt);
+
+ if notify.on_attention {
+ full_prompt.push_str("\n\n---\n\n# Delivery\n\n");
+ full_prompt.push_str(
+ "If you reply with anything other than ROUTINE_OK, the host will deliver your \
+ reply as the routine notification. Return the message exactly as it should be sent.\n",
+ );
+
+ if let Some(channel) = notify.channel.as_deref() {
+ full_prompt.push_str(&format!(
+ "The configured delivery channel for this routine is `{channel}`.\n"
+ ));
+ }
+
+ if let Some(user) = notify.user.as_deref() {
+ full_prompt.push_str(&format!(
+ "The configured delivery target for this routine is `{user}`.\n"
+ ));
+ }
+
+ full_prompt.push_str(
+ "Do not claim you lack messaging integrations or ask the user to set one up when \
+ a plain reply is sufficient.\n",
+ );
+ }
+
+ if !use_tools {
+ full_prompt.push_str(
+ "\nTools are disabled for this routine run. Do not ask to call tools or describe tool limitations unless they prevent a necessary external action.\n",
+ );
+ }
+
+ if !context_parts.is_empty() {
+ full_prompt.push_str("\n\n---\n\n# Context\n\n");
+ full_prompt.push_str(&context_parts.join("\n\n"));
+ }
+
+ if let Some(state) = state_content {
+ full_prompt.push_str("\n\n---\n\n# Previous State\n\n");
+ full_prompt.push_str(state);
+ }
+
+ full_prompt.push_str(
+ "\n\n---\n\nIf nothing needs attention, reply EXACTLY with: ROUTINE_OK\n\
+ If something needs attention, provide a concise summary.",
+ );
+
+ full_prompt
+}
+
/// Execute a lightweight routine without tool support (original single-call behavior).
async fn execute_lightweight_no_tools(
ctx: &EngineContext,
@@ -901,8 +1296,8 @@ fn handle_text_response(
};
}
- // Check for the "nothing to do" sentinel
- if content == "ROUTINE_OK" || content.contains("ROUTINE_OK") {
+ // Check for the "nothing to do" sentinel (exact match on trimmed content).
+ if content == "ROUTINE_OK" {
let total_tokens = Some((total_input_tokens + total_output_tokens) as i32);
return Ok((RunStatus::Ok, None, total_tokens));
}
@@ -1268,14 +1663,22 @@ pub fn spawn_cron_ticker(
interval: Duration,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
- // Run one check immediately so routines due at startup don't wait
- // an extra full polling interval.
+ // Recover orphaned runs from a previous process crash before
+ // dispatching any new work, so we don't confuse fresh dispatches
+ // with crash orphans.
+ engine.sync_dispatched_runs().await;
+
+ // Run one cron check immediately so routines due at startup don't
+ // wait an extra full polling interval.
engine.check_cron_triggers().await;
let mut ticker = tokio::time::interval(interval);
loop {
ticker.tick().await;
+ // Sync first: only processes runs from before boot_time, so it
+ // never races with FullJobWatcher instances from this process.
+ engine.sync_dispatched_runs().await;
engine.check_cron_triggers().await;
}
})
@@ -1385,22 +1788,78 @@ mod tests {
}
}
+ #[test]
+ fn test_build_lightweight_prompt_explains_delivery_and_disabled_tools() {
+ let notify = NotifyConfig {
+ channel: Some("telegram".to_string()),
+ user: Some("default".to_string()),
+ on_attention: true,
+ on_failure: true,
+ on_success: false,
+ };
+
+ let prompt = super::build_lightweight_prompt(
+ "Send a Telegram reminder message to the user.",
+ &[],
+ None,
+ ¬ify,
+ false,
+ );
+
+ assert!(
+ prompt.contains("the host will deliver your reply as the routine notification"),
+ "delivery guidance should explain host delivery: {prompt}",
+ );
+ assert!(
+ prompt.contains("configured delivery channel for this routine is `telegram`"),
+ "delivery guidance should mention telegram channel: {prompt}",
+ );
+ assert!(
+ prompt.contains("Do not claim you lack messaging integrations"),
+ "delivery guidance should suppress fake setup chatter: {prompt}",
+ );
+ assert!(
+ prompt.contains("Tools are disabled for this routine run"),
+ "prompt should explain that tools are disabled: {prompt}",
+ );
+ }
+
+ #[test]
+ fn test_build_lightweight_prompt_skips_delivery_block_when_attention_notifications_disabled() {
+ let notify = NotifyConfig {
+ on_attention: false,
+ ..NotifyConfig::default()
+ };
+
+ let prompt = super::build_lightweight_prompt("Check inbox.", &[], None, ¬ify, true);
+
+ assert!(
+ !prompt.contains("# Delivery"),
+ "prompt should not include delivery guidance when attention notifications are off: {prompt}",
+ );
+ assert!(
+ !prompt.contains("Tools are disabled for this routine run"),
+ "prompt should not claim tools are disabled when they are enabled: {prompt}",
+ );
+ }
+
#[test]
fn test_routine_sentinel_detection_exact_match() {
- // The execute_lightweight_no_tools checks: content == "ROUTINE_OK" || content.contains("ROUTINE_OK")
- // After trim(), whitespace is removed
+ // Sentinel detection uses exact match on trimmed content to avoid
+ // false positives from substrings like "NOT_ROUTINE_OK".
let test_cases = vec![
("ROUTINE_OK", true),
(" ROUTINE_OK ", true), // After trim, whitespace is removed so matches
- ("something ROUTINE_OK something", true),
- ("ROUTINE_OK is done", true),
- ("done ROUTINE_OK", true),
+ ("something ROUTINE_OK something", false), // substring no longer matches
+ ("ROUTINE_OK is done", false), // substring no longer matches
+ ("done ROUTINE_OK", false), // substring no longer matches
+ ("NOT_ROUTINE_OK", false), // exact match prevents this
("no sentinel here", false),
];
for (content, should_match) in test_cases {
let trimmed = content.trim();
- let matches = trimmed == "ROUTINE_OK" || trimmed.contains("ROUTINE_OK");
+ let matches = trimmed == "ROUTINE_OK";
assert_eq!(
matches, should_match,
"Content '{}' sentinel detection should be {}, got {}",
@@ -1514,4 +1973,86 @@ mod tests {
assert_eq!(snapshot[1].content, "a"); // safety: test-only no-panics CI false positive
assert_eq!(snapshot[2].content, "b"); // safety: test-only no-panics CI false positive
}
+
+ /// Regression test for #1317: FullJobWatcher maps terminal job states correctly.
+ #[test]
+ fn test_full_job_watcher_state_mapping() {
+ use crate::context::JobState;
+
+ // Failed/Cancelled → RunStatus::Failed
+ assert_eq!(
+ super::FullJobWatcher::map_job_state(&JobState::Failed),
+ RunStatus::Failed
+ );
+ assert_eq!(
+ super::FullJobWatcher::map_job_state(&JobState::Cancelled),
+ RunStatus::Failed
+ );
+
+ // All other non-active states → RunStatus::Ok
+ assert_eq!(
+ super::FullJobWatcher::map_job_state(&JobState::Completed),
+ RunStatus::Ok
+ );
+ assert_eq!(
+ super::FullJobWatcher::map_job_state(&JobState::Accepted),
+ RunStatus::Ok
+ );
+ }
+
+ /// Verify that job state to run status mapping covers all expected cases.
+ #[test]
+ fn test_job_state_to_run_status_mapping() {
+ use crate::context::JobState;
+
+ // Success states
+ for state in [JobState::Completed, JobState::Submitted, JobState::Accepted] {
+ let status = match state {
+ JobState::Completed | JobState::Submitted | JobState::Accepted => {
+ Some(RunStatus::Ok)
+ }
+ JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed),
+ _ => None,
+ };
+ assert_eq!(
+ status,
+ Some(RunStatus::Ok),
+ "{:?} should map to RunStatus::Ok",
+ state
+ );
+ }
+
+ // Failure states
+ for state in [JobState::Failed, JobState::Cancelled] {
+ let status = match state {
+ JobState::Completed | JobState::Submitted | JobState::Accepted => {
+ Some(RunStatus::Ok)
+ }
+ JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed),
+ _ => None,
+ };
+ assert_eq!(
+ status,
+ Some(RunStatus::Failed),
+ "{:?} should map to RunStatus::Failed",
+ state
+ );
+ }
+
+ // Active states (should not finalize)
+ for state in [JobState::Pending, JobState::InProgress, JobState::Stuck] {
+ let status = match state {
+ JobState::Completed | JobState::Submitted | JobState::Accepted => {
+ Some(RunStatus::Ok)
+ }
+ JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed),
+ _ => None,
+ };
+ assert_eq!(
+ status, None,
+ "{:?} should not finalize the routine run",
+ state
+ );
+ }
+ }
}
diff --git a/src/agent/self_repair.rs b/src/agent/self_repair.rs
index a67fe23e..db491194 100644
--- a/src/agent/self_repair.rs
+++ b/src/agent/self_repair.rs
@@ -66,14 +66,10 @@ pub trait SelfRepair: Send + Sync {
/// Default self-repair implementation.
pub struct DefaultSelfRepair {
context_manager: Arc,
- // TODO: use for time-based stuck detection (currently only max_repair_attempts is checked)
- #[allow(dead_code)]
stuck_threshold: Duration,
max_repair_attempts: u32,
store: Option>,
builder: Option>,
- // TODO: use for tool hot-reload after repair
- #[allow(dead_code)]
tools: Option>,
}
@@ -95,15 +91,13 @@ impl DefaultSelfRepair {
}
/// Add a Store for tool failure tracking.
- #[allow(dead_code)] // TODO: wire up in main.rs when persistence is needed
- pub(crate) fn with_store(mut self, store: Arc) -> Self {
+ pub fn with_store(mut self, store: Arc) -> Self {
self.store = Some(store);
self
}
/// Add a Builder and ToolRegistry for automatic tool repair.
- #[allow(dead_code)] // TODO: wire up in main.rs when auto-repair is needed
- pub(crate) fn with_builder(
+ pub fn with_builder(
mut self,
builder: Arc,
tools: Arc,
@@ -124,18 +118,30 @@ impl SelfRepair for DefaultSelfRepair {
if let Ok(ctx) = self.context_manager.get_context(job_id).await
&& ctx.state == JobState::Stuck
{
- let stuck_duration = ctx
- .started_at
- .map(|start| {
- let now = Utc::now();
- let duration = now.signed_duration_since(start);
+ // Measure stuck_duration from the most recent Stuck transition,
+ // not from started_at (which reflects when the job first ran).
+ let stuck_since = ctx
+ .transitions
+ .iter()
+ .rev()
+ .find(|t| t.to == JobState::Stuck)
+ .map(|t| t.timestamp);
+
+ let stuck_duration = stuck_since
+ .map(|ts| {
+ let duration = Utc::now().signed_duration_since(ts);
Duration::from_secs(duration.num_seconds().max(0) as u64)
})
.unwrap_or_default();
+ // Only report jobs that have been stuck long enough
+ if stuck_duration < self.stuck_threshold {
+ continue;
+ }
+
stuck_jobs.push(StuckJob {
job_id,
- last_activity: ctx.started_at.unwrap_or(ctx.created_at),
+ last_activity: stuck_since.unwrap_or(ctx.created_at),
stuck_duration,
last_error: None,
repair_attempts: ctx.repair_attempts,
@@ -273,9 +279,8 @@ impl SelfRepair for DefaultSelfRepair {
tracing::warn!("Failed to mark tool as repaired: {}", e);
}
- // Log if the tool was auto-registered
if result.registered {
- tracing::info!("Repaired tool '{}' auto-registered", tool.name);
+ tracing::info!("Repaired tool '{}' auto-registered by builder", tool.name);
}
Ok(RepairResult::Success {
@@ -417,7 +422,8 @@ mod tests {
.unwrap()
.unwrap();
- let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 3);
+ // Use zero threshold so the just-stuck job is detected immediately.
+ let repair = DefaultSelfRepair::new(cm, Duration::from_secs(0), 3);
let stuck = repair.detect_stuck_jobs().await;
assert_eq!(stuck.len(), 1);
assert_eq!(stuck[0].job_id, job_id);
@@ -483,6 +489,98 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn detect_stuck_jobs_filters_by_threshold() {
+ let cm = Arc::new(ContextManager::new(10));
+ let job_id = cm.create_job("Stuck job", "desc").await.unwrap();
+
+ // Transition to InProgress, then to Stuck.
+ cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
+ .await
+ .unwrap()
+ .unwrap();
+ cm.update_context(job_id, |ctx| {
+ ctx.transition_to(JobState::Stuck, Some("timed out".to_string()))
+ })
+ .await
+ .unwrap()
+ .unwrap();
+
+ // Use a very large threshold (1 hour). Job just became stuck, so
+ // stuck_duration < threshold. It should be filtered out.
+ let repair = DefaultSelfRepair::new(cm, Duration::from_secs(3600), 3);
+ let stuck = repair.detect_stuck_jobs().await;
+ assert!(
+ stuck.is_empty(),
+ "Job stuck for <1s should be filtered by 1h threshold"
+ );
+ }
+
+ #[tokio::test]
+ async fn detect_stuck_jobs_includes_when_over_threshold() {
+ let cm = Arc::new(ContextManager::new(10));
+ let job_id = cm.create_job("Stuck job", "desc").await.unwrap();
+
+ // Transition to InProgress, then to Stuck.
+ cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
+ .await
+ .unwrap()
+ .unwrap();
+ cm.update_context(job_id, |ctx| {
+ ctx.transition_to(JobState::Stuck, Some("timed out".to_string()))
+ })
+ .await
+ .unwrap()
+ .unwrap();
+
+ // Use a zero threshold -- any stuck duration should be included.
+ let repair = DefaultSelfRepair::new(cm, Duration::from_secs(0), 3);
+ let stuck = repair.detect_stuck_jobs().await;
+ assert_eq!(stuck.len(), 1, "Job should be detected with zero threshold");
+ assert_eq!(stuck[0].job_id, job_id);
+ }
+
+ /// Regression: stuck_duration must be measured from the Stuck transition,
+ /// not from started_at. A job that ran for 2 hours before becoming stuck
+ /// should NOT immediately exceed a 5-minute threshold.
+ #[tokio::test]
+ async fn stuck_duration_measured_from_stuck_transition_not_started_at() {
+ let cm = Arc::new(ContextManager::new(10));
+ let job_id = cm.create_job("Long runner", "desc").await.unwrap();
+
+ // Transition to InProgress (sets started_at to now).
+ cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
+ .await
+ .unwrap()
+ .unwrap();
+
+ // Backdate started_at to 2 hours ago to simulate a long-running job.
+ cm.update_context(job_id, |ctx| {
+ ctx.started_at = Some(Utc::now() - chrono::Duration::hours(2));
+ Ok::<(), crate::error::Error>(())
+ })
+ .await
+ .unwrap()
+ .unwrap();
+
+ // Now transition to Stuck (stuck transition timestamp is ~now).
+ cm.update_context(job_id, |ctx| {
+ ctx.transition_to(JobState::Stuck, Some("wedged".into()))
+ })
+ .await
+ .unwrap()
+ .unwrap();
+
+ // With a 5-minute threshold, the job JUST became stuck — should NOT be detected.
+ let repair = DefaultSelfRepair::new(cm, Duration::from_secs(300), 3);
+ let stuck = repair.detect_stuck_jobs().await;
+ assert!(
+ stuck.is_empty(),
+ "Job stuck for <1s should not exceed 5min threshold, \
+ but stuck_duration was computed from started_at (2h ago)"
+ );
+ }
+
#[tokio::test]
async fn detect_broken_tools_returns_empty_without_store() {
let cm = Arc::new(ContextManager::new(10));
@@ -515,4 +613,148 @@ mod tests {
result
);
}
+
+ /// Mock SoftwareBuilder that returns a successful build result.
+ struct MockBuilder {
+ build_count: std::sync::atomic::AtomicU32,
+ }
+
+ impl MockBuilder {
+ fn new() -> Self {
+ Self {
+ build_count: std::sync::atomic::AtomicU32::new(0),
+ }
+ }
+
+ fn builds(&self) -> u32 {
+ self.build_count.load(std::sync::atomic::Ordering::Relaxed)
+ }
+ }
+
+ #[async_trait]
+ impl crate::tools::SoftwareBuilder for MockBuilder {
+ async fn analyze(
+ &self,
+ _description: &str,
+ ) -> Result {
+ Ok(crate::tools::BuildRequirement {
+ name: "mock-tool".to_string(),
+ description: "mock".to_string(),
+ software_type: crate::tools::SoftwareType::WasmTool,
+ language: crate::tools::Language::Rust,
+ input_spec: None,
+ output_spec: None,
+ dependencies: vec![],
+ capabilities: vec![],
+ })
+ }
+
+ async fn build(
+ &self,
+ requirement: &crate::tools::BuildRequirement,
+ ) -> Result {
+ self.build_count
+ .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+ Ok(crate::tools::BuildResult {
+ build_id: Uuid::new_v4(),
+ requirement: requirement.clone(),
+ artifact_path: std::path::PathBuf::from("/tmp/mock.wasm"),
+ logs: vec![],
+ success: true,
+ error: None,
+ started_at: Utc::now(),
+ completed_at: Utc::now(),
+ iterations: 1,
+ validation_warnings: vec![],
+ tests_passed: 1,
+ tests_failed: 0,
+ registered: true,
+ })
+ }
+
+ async fn repair(
+ &self,
+ _result: &crate::tools::BuildResult,
+ _error: &str,
+ ) -> Result {
+ unimplemented!("not needed for this test")
+ }
+ }
+
+ /// E2E test: stuck job detected -> repaired -> transitions back to InProgress,
+ /// and broken tool detected -> builder invoked -> tool marked repaired.
+ #[cfg(feature = "libsql")]
+ #[tokio::test]
+ async fn e2e_stuck_job_repair_and_tool_rebuild() {
+ // --- Setup ---
+ let cm = Arc::new(ContextManager::new(10));
+ let job_id = cm.create_job("E2E stuck job", "desc").await.unwrap();
+
+ // Transition job: Pending -> InProgress -> Stuck
+ cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
+ .await
+ .unwrap()
+ .unwrap();
+ cm.update_context(job_id, |ctx| {
+ ctx.transition_to(JobState::Stuck, Some("deadlocked".to_string()))
+ })
+ .await
+ .unwrap()
+ .unwrap();
+
+ // Create a mock builder and a real test database (for store)
+ let builder = Arc::new(MockBuilder::new());
+ let tools = Arc::new(ToolRegistry::new());
+ let (db, _tmp_dir) = crate::testing::test_db().await;
+
+ // Create self-repair with zero threshold (detect immediately),
+ // wired with store, builder, and tools.
+ let repair = DefaultSelfRepair::new(Arc::clone(&cm), Duration::from_secs(0), 3)
+ .with_store(Arc::clone(&db))
+ .with_builder(
+ Arc::clone(&builder) as Arc,
+ tools,
+ );
+
+ // --- Phase 1: Detect and repair stuck job ---
+ let stuck_jobs = repair.detect_stuck_jobs().await;
+ assert_eq!(stuck_jobs.len(), 1, "Should detect the stuck job");
+ assert_eq!(stuck_jobs[0].job_id, job_id);
+
+ let result = repair.repair_stuck_job(&stuck_jobs[0]).await.unwrap();
+ assert!(
+ matches!(result, RepairResult::Success { .. }),
+ "Job repair should succeed: {:?}",
+ result
+ );
+
+ // Verify job transitioned back to InProgress
+ let ctx = cm.get_context(job_id).await.unwrap();
+ assert_eq!(
+ ctx.state,
+ JobState::InProgress,
+ "Job should be back to InProgress after repair"
+ );
+
+ // --- Phase 2: Repair a broken tool via builder ---
+ let broken = BrokenTool {
+ name: "broken-wasm-tool".to_string(),
+ failure_count: 10,
+ last_error: Some("panic in tool execution".to_string()),
+ first_failure: Utc::now() - chrono::Duration::hours(1),
+ last_failure: Utc::now(),
+ last_build_result: None,
+ repair_attempts: 0,
+ };
+
+ let tool_result = repair.repair_broken_tool(&broken).await.unwrap();
+ assert!(
+ matches!(tool_result, RepairResult::Success { .. }),
+ "Tool repair should succeed with mock builder: {:?}",
+ tool_result
+ );
+
+ // Verify builder was actually invoked
+ assert_eq!(builder.builds(), 1, "Builder should have been called once");
+ }
}
diff --git a/src/agent/session.rs b/src/agent/session.rs
index 4abbea61..3e84afc0 100644
--- a/src/agent/session.rs
+++ b/src/agent/session.rs
@@ -188,6 +188,15 @@ pub struct PendingApproval {
/// through the approval flow even if the approval message lacks timezone.
#[serde(default)]
pub user_timezone: Option,
+ /// Whether the "always" auto-approve option should be offered to the user.
+ /// `false` when the tool returned `ApprovalRequirement::Always` (e.g.
+ /// destructive shell commands), meaning every invocation must be confirmed.
+ #[serde(default = "default_true")]
+ pub allow_always: bool,
+}
+
+fn default_true() -> bool {
+ true
}
/// A conversation thread within a session.
@@ -1106,6 +1115,7 @@ mod tests {
context_messages: vec![ChatMessage::user("do it")],
deferred_tool_calls: vec![],
user_timezone: None,
+ allow_always: false,
};
thread.await_approval(approval);
@@ -1132,6 +1142,7 @@ mod tests {
context_messages: vec![],
deferred_tool_calls: vec![],
user_timezone: None,
+ allow_always: true,
};
thread.await_approval(approval);
diff --git a/src/agent/submission.rs b/src/agent/submission.rs
index a3ae2524..8594c969 100644
--- a/src/agent/submission.rs
+++ b/src/agent/submission.rs
@@ -382,6 +382,8 @@ pub enum SubmissionResult {
description: String,
/// Parameters being passed.
parameters: serde_json::Value,
+ /// Whether "always" auto-approve should be offered to the user.
+ allow_always: bool,
},
/// Successfully processed (for control commands).
diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs
index 877a4e27..2b489a7a 100644
--- a/src/agent/thread_ops.rs
+++ b/src/agent/thread_ops.rs
@@ -506,7 +506,8 @@ impl Agent {
let tool_name = pending.tool_name.clone();
let description = pending.description.clone();
let parameters = pending.display_parameters.clone();
- thread.await_approval(pending);
+ let allow_always = pending.allow_always;
+ thread.await_approval(*pending);
let _ = self
.channels
.send_status(
@@ -516,6 +517,7 @@ impl Agent {
tool_name: tool_name.clone(),
description: description.clone(),
parameters: parameters.clone(),
+ allow_always,
},
&message.metadata,
)
@@ -525,6 +527,7 @@ impl Agent {
tool_name,
description,
parameters,
+ allow_always,
})
}
Err(e) => {
@@ -1069,28 +1072,31 @@ impl Agent {
usize,
crate::llm::ToolCall,
Arc,
+ bool, // allow_always
)> = None;
for (idx, tc) in deferred_tool_calls.iter().enumerate() {
if let Some(tool) = self.tools().get(&tc.name).await {
// Match dispatcher.rs: when auto_approve_tools is true, skip
// all approval checks (including ApprovalRequirement::Always).
- let needs_approval = if self.config.auto_approve_tools {
- false
+ let (needs_approval, allow_always) = if self.config.auto_approve_tools {
+ (false, true)
} else {
use crate::tools::ApprovalRequirement;
- match tool.requires_approval(&tc.arguments) {
+ let requirement = tool.requires_approval(&tc.arguments);
+ let needs = match requirement {
ApprovalRequirement::Never => false,
ApprovalRequirement::UnlessAutoApproved => {
let sess = session.lock().await;
!sess.is_tool_auto_approved(&tc.name)
}
ApprovalRequirement::Always => true,
- }
+ };
+ (needs, !matches!(requirement, ApprovalRequirement::Always))
};
if needs_approval {
- approval_needed = Some((idx, tc.clone(), tool));
+ approval_needed = Some((idx, tc.clone(), tool, allow_always));
break; // remaining tools stay deferred
}
}
@@ -1298,7 +1304,7 @@ impl Agent {
}
// Handle approval if a tool needed it
- if let Some((approval_idx, tc, tool)) = approval_needed {
+ if let Some((approval_idx, tc, tool, allow_always)) = approval_needed {
let new_pending = PendingApproval {
request_id: Uuid::new_v4(),
tool_name: tc.name.clone(),
@@ -1310,6 +1316,7 @@ impl Agent {
deferred_tool_calls: deferred_tool_calls[approval_idx + 1..].to_vec(),
// Carry forward the resolved timezone from the original pending approval
user_timezone: pending.user_timezone.clone(),
+ allow_always,
};
let request_id = new_pending.request_id;
@@ -1333,6 +1340,7 @@ impl Agent {
tool_name: tool_name.clone(),
description: description.clone(),
parameters: parameters.clone(),
+ allow_always,
},
&message.metadata,
)
@@ -1343,6 +1351,7 @@ impl Agent {
tool_name,
description,
parameters,
+ allow_always,
});
}
@@ -1411,7 +1420,8 @@ impl Agent {
let tool_name = new_pending.tool_name.clone();
let description = new_pending.description.clone();
let parameters = new_pending.display_parameters.clone();
- thread.await_approval(new_pending);
+ let allow_always = new_pending.allow_always;
+ thread.await_approval(*new_pending);
let _ = self
.channels
.send_status(
@@ -1421,6 +1431,7 @@ impl Agent {
tool_name: tool_name.clone(),
description: description.clone(),
parameters: parameters.clone(),
+ allow_always,
},
&message.metadata,
)
@@ -1430,6 +1441,7 @@ impl Agent {
tool_name,
description,
parameters,
+ allow_always,
})
}
Err(e) => {
diff --git a/src/app.rs b/src/app.rs
index d5730645..5cbd6a35 100644
--- a/src/app.rs
+++ b/src/app.rs
@@ -56,6 +56,7 @@ pub struct AppComponents {
pub session: Arc,
pub catalog_entries: Vec,
pub dev_loaded_tool_names: Vec,
+ pub builder: Option>,
}
/// Options that control optional init phases.
@@ -280,6 +281,7 @@ impl AppBuilder {
Arc,
Option>,
Option>,
+ Option>,
),
anyhow::Error,
> {
@@ -367,16 +369,19 @@ impl AppBuilder {
}
// Register builder tool if enabled
- if self.config.builder.enabled
+ let builder = if self.config.builder.enabled
&& (self.config.agent.allow_local_tools || !self.config.sandbox.enabled)
{
- tools
+ let b = tools
.register_builder_tool(llm.clone(), Some(self.config.builder.to_builder_config()))
.await;
- tracing::debug!("Builder mode enabled");
- }
+ tracing::info!("Builder mode enabled");
+ Some(b)
+ } else {
+ None
+ };
- Ok((safety, tools, embeddings, workspace))
+ Ok((safety, tools, embeddings, workspace, builder))
}
/// Phase 5: Load WASM tools, MCP servers, and create extension manager.
@@ -703,7 +708,7 @@ impl AppBuilder {
} else {
self.init_llm().await?
};
- let (safety, tools, embeddings, workspace) = self.init_tools(&llm).await?;
+ let (safety, tools, embeddings, workspace, builder) = self.init_tools(&llm).await?;
// Create hook registry early so runtime extension activation can register hooks.
let hooks = Arc::new(HookRegistry::new());
@@ -823,6 +828,7 @@ impl AppBuilder {
session: self.session,
catalog_entries,
dev_loaded_tool_names,
+ builder,
})
}
}
diff --git a/src/channels/channel.rs b/src/channels/channel.rs
index 43e35688..a85cf8c5 100644
--- a/src/channels/channel.rs
+++ b/src/channels/channel.rs
@@ -305,6 +305,11 @@ pub enum StatusUpdate {
tool_name: String,
description: String,
parameters: serde_json::Value,
+ /// When `true`, the UI should offer an "always" option that auto-approves
+ /// future calls to this tool for the rest of the session. When `false`
+ /// (i.e. `ApprovalRequirement::Always`), the tool must be approved every
+ /// time and the "always" button should be hidden.
+ allow_always: bool,
},
/// Extension needs user authentication (token or OAuth).
AuthRequired {
diff --git a/src/channels/manager.rs b/src/channels/manager.rs
index b026ff85..0c9a3da7 100644
--- a/src/channels/manager.rs
+++ b/src/channels/manager.rs
@@ -239,6 +239,11 @@ impl ChannelManager {
pub async fn get_channel(&self, name: &str) -> Option> {
self.channels.read().await.get(name).cloned()
}
+
+ /// Remove a channel from the manager.
+ pub async fn remove(&self, name: &str) -> Option> {
+ self.channels.write().await.remove(name)
+ }
}
impl Default for ChannelManager {
diff --git a/src/channels/relay/channel.rs b/src/channels/relay/channel.rs
index 52aea478..3b6c3379 100644
--- a/src/channels/relay/channel.rs
+++ b/src/channels/relay/channel.rs
@@ -1,16 +1,16 @@
-//! Channel trait implementation for channel-relay SSE streams.
+//! Channel trait implementation for channel-relay webhook callbacks.
//!
-//! `RelayChannel` connects to a channel-relay service via SSE, converts
-//! incoming events to `IncomingMessage`s, and sends responses via the
-//! relay's provider-specific proxy API (Slack).
+//! `RelayChannel` receives events from channel-relay via HTTP POST callbacks
+//! (pushed through an mpsc channel by the webhook handler), converts them
+//! to `IncomingMessage`s, and sends responses via the relay's provider-specific
+//! proxy API (Slack).
use std::collections::HashMap;
-use std::sync::Arc;
use async_trait::async_trait;
-use tokio::sync::{RwLock, mpsc};
+use tokio::sync::mpsc;
-use crate::channels::relay::client::{RelayClient, RelayError};
+use crate::channels::relay::client::{ChannelEvent, RelayClient};
use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate};
use crate::error::ChannelError;
@@ -39,44 +39,34 @@ impl RelayProvider {
}
}
-/// Channel implementation that connects to a channel-relay SSE stream.
+/// Channel implementation that receives events from channel-relay via webhook callbacks.
pub struct RelayChannel {
client: RelayClient,
provider: RelayProvider,
- stream_token: Arc>,
team_id: String,
instance_id: String,
- user_id: String,
- /// SSE stream long-poll timeout in seconds.
- stream_timeout_secs: u64,
- /// Initial exponential backoff in milliseconds.
- backoff_initial_ms: u64,
- /// Maximum exponential backoff in milliseconds.
- backoff_max_ms: u64,
- /// Handle to the reconnect task for clean shutdown.
- reconnect_handle: RwLock