mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 23:50:17 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
8a320ae9db | ||
|
|
fd41bdf4be | ||
|
|
de5a1c7b0d | ||
|
|
9ce3a9fc53 |
Generated
+16
-5
@@ -3150,7 +3150,7 @@ dependencies = [
|
||||
"libc",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"socket2 0.5.10",
|
||||
"socket2 0.6.3",
|
||||
"system-configuration",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
@@ -3439,6 +3439,7 @@ dependencies = [
|
||||
"pgvector",
|
||||
"postgres-types",
|
||||
"pretty_assertions",
|
||||
"pty-process",
|
||||
"rand 0.8.5",
|
||||
"readabilityrs",
|
||||
"refinery",
|
||||
@@ -3524,7 +3525,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
"windows-sys 0.59.0",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4906,6 +4907,16 @@ dependencies = [
|
||||
"syn 1.0.109",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pty-process"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266"
|
||||
dependencies = [
|
||||
"rustix 1.1.4",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pulley-interpreter"
|
||||
version = "28.0.1"
|
||||
@@ -4930,7 +4941,7 @@ dependencies = [
|
||||
"quinn-udp",
|
||||
"rustc-hash 2.1.1",
|
||||
"rustls 0.23.37",
|
||||
"socket2 0.5.10",
|
||||
"socket2 0.6.3",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tracing",
|
||||
@@ -4967,9 +4978,9 @@ dependencies = [
|
||||
"cfg_aliases",
|
||||
"libc",
|
||||
"once_cell",
|
||||
"socket2 0.5.10",
|
||||
"socket2 0.6.3",
|
||||
"tracing",
|
||||
"windows-sys 0.59.0",
|
||||
"windows-sys 0.60.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
+5
-2
@@ -189,6 +189,10 @@ json5 = { version = "0.4", optional = true }
|
||||
[target.'cfg(target_os = "macos")'.dependencies]
|
||||
security-framework = "3"
|
||||
|
||||
# PTY allocation for Claude CLI stdout buffering fix (Unix only)
|
||||
[target.'cfg(unix)'.dependencies]
|
||||
pty-process = { version = "0.5", features = ["async"] }
|
||||
|
||||
# Linux secret-service (GNOME Keyring, KWallet)
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
secret-service = { version = "4", features = ["rt-tokio-crypto-rust"] }
|
||||
@@ -248,8 +252,7 @@ strip = true # Remove debug symbols from release binaries
|
||||
# The profile that 'cargo dist' will build with
|
||||
[profile.dist]
|
||||
inherits = "release"
|
||||
lto = "fat" # Full cross-crate LTO (slow build, better codegen)
|
||||
codegen-units = 1 # Single codegen unit for maximum optimization
|
||||
lto = "thin"
|
||||
|
||||
# Config for 'dist'
|
||||
[workspace.metadata.dist]
|
||||
|
||||
+118
-23
@@ -28,6 +28,9 @@ use std::{cmp::Ordering, collections::HashMap};
|
||||
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Discord REST API v10 base URL.
|
||||
const DISCORD_API_BASE: &str = "https://discord.com/api/v10";
|
||||
|
||||
use exports::near::agent::channel::{
|
||||
AgentResponse, ChannelConfig, Guest, HttpEndpointConfig, IncomingHttpRequest,
|
||||
OutgoingHttpResponse, PollConfig, StatusUpdate,
|
||||
@@ -427,7 +430,7 @@ impl Guest for DiscordChannel {
|
||||
(
|
||||
"PATCH",
|
||||
format!(
|
||||
"https://discord.com/api/v10/webhooks/{}/{}/messages/@original",
|
||||
"{DISCORD_API_BASE}/webhooks/{}/{}/messages/@original",
|
||||
application_id, token
|
||||
),
|
||||
)
|
||||
@@ -438,20 +441,7 @@ impl Guest for DiscordChannel {
|
||||
payload["allowed_mentions"] = serde_json::json!({
|
||||
"replied_user": true
|
||||
});
|
||||
let mention_payload = serde_json::to_vec(&payload)
|
||||
.map_err(|e| format!("Failed to serialize mention payload: {}", e))?;
|
||||
let mention_url = format!(
|
||||
"https://discord.com/api/v10/channels/{}/messages",
|
||||
metadata.channel_id
|
||||
);
|
||||
let result = channel_host::http_request(
|
||||
"POST",
|
||||
&mention_url,
|
||||
&discord_auth_headers_json(true),
|
||||
Some(&mention_payload),
|
||||
None,
|
||||
);
|
||||
return map_discord_response(result);
|
||||
return send_channel_message(&metadata.channel_id, payload);
|
||||
} else {
|
||||
return Err("Unsupported Discord response metadata".to_string());
|
||||
};
|
||||
@@ -469,8 +459,8 @@ impl Guest for DiscordChannel {
|
||||
|
||||
fn on_status(_update: StatusUpdate) {}
|
||||
|
||||
fn on_broadcast(_user_id: String, _response: AgentResponse) -> Result<(), String> {
|
||||
Err("broadcast not yet implemented for Discord channel".to_string())
|
||||
fn on_broadcast(user_id: String, response: AgentResponse) -> Result<(), String> {
|
||||
broadcast_dm(&user_id, &response.content)
|
||||
}
|
||||
|
||||
fn on_shutdown() {
|
||||
@@ -501,6 +491,21 @@ fn map_discord_response(
|
||||
}
|
||||
}
|
||||
|
||||
/// Post a JSON payload to a Discord channel as a new message.
|
||||
fn send_channel_message(channel_id: &str, payload: serde_json::Value) -> Result<(), String> {
|
||||
let payload_bytes = serde_json::to_vec(&payload)
|
||||
.map_err(|e| format!("Failed to serialize message: {}", e))?;
|
||||
let url = format!("{DISCORD_API_BASE}/channels/{}/messages", channel_id);
|
||||
let result = channel_host::http_request(
|
||||
"POST",
|
||||
&url,
|
||||
&discord_auth_headers_json(true),
|
||||
Some(&payload_bytes),
|
||||
None,
|
||||
);
|
||||
map_discord_response(result)
|
||||
}
|
||||
|
||||
fn load_runtime_config() -> DiscordRuntimeConfig {
|
||||
channel_host::workspace_read("config.json")
|
||||
.and_then(|raw| serde_json::from_str::<DiscordRuntimeConfig>(&raw).ok())
|
||||
@@ -539,7 +544,7 @@ fn get_or_fetch_bot_id() -> Option<String> {
|
||||
|
||||
let response = channel_host::http_request(
|
||||
"GET",
|
||||
"https://discord.com/api/v10/users/@me",
|
||||
&format!("{DISCORD_API_BASE}/users/@me"),
|
||||
&discord_auth_headers_json(false),
|
||||
None,
|
||||
Some(10_000),
|
||||
@@ -659,7 +664,7 @@ fn poll_channel_mentions(channel_id: &str, bot_id: &str) {
|
||||
|
||||
fn fetch_latest_message_id(channel_id: &str) -> Option<String> {
|
||||
let url = format!(
|
||||
"https://discord.com/api/v10/channels/{}/messages?limit=1",
|
||||
"{DISCORD_API_BASE}/channels/{}/messages?limit=1",
|
||||
channel_id
|
||||
);
|
||||
let response = channel_host::http_request(
|
||||
@@ -697,7 +702,7 @@ fn fetch_messages_after_cursor(
|
||||
|
||||
for page in 0..MAX_PAGES {
|
||||
let url = format!(
|
||||
"https://discord.com/api/v10/channels/{}/messages?limit={}&after={}",
|
||||
"{DISCORD_API_BASE}/channels/{}/messages?limit={}&after={}",
|
||||
channel_id, PAGE_LIMIT, after
|
||||
);
|
||||
let response = match channel_host::http_request(
|
||||
@@ -986,7 +991,7 @@ fn handle_slash_command(interaction: &DiscordInteraction) -> bool {
|
||||
);
|
||||
// Attempt to notify user of internal error
|
||||
let url = format!(
|
||||
"https://discord.com/api/v10/webhooks/{}/{}",
|
||||
"{DISCORD_API_BASE}/webhooks/{}/{}",
|
||||
interaction.application_id, interaction.token
|
||||
);
|
||||
let payload = serde_json::json!({
|
||||
@@ -1106,7 +1111,7 @@ fn check_sender_permission(
|
||||
}
|
||||
|
||||
let dm_policy =
|
||||
channel_host::workspace_read(DM_POLICY_PATH).unwrap_or_else(|| default_dm_policy());
|
||||
channel_host::workspace_read(DM_POLICY_PATH).unwrap_or_else(default_dm_policy);
|
||||
if dm_policy == "open" {
|
||||
return true;
|
||||
}
|
||||
@@ -1161,7 +1166,7 @@ fn check_sender_permission(
|
||||
/// Send a pairing code as an ephemeral Discord followup message.
|
||||
fn send_pairing_reply(ctx: &PairingReplyCtx, code: &str) -> Result<(), String> {
|
||||
let url = format!(
|
||||
"https://discord.com/api/v10/webhooks/{}/{}",
|
||||
"{DISCORD_API_BASE}/webhooks/{}/{}",
|
||||
ctx.application_id, ctx.token
|
||||
);
|
||||
let payload = serde_json::json!({
|
||||
@@ -1194,6 +1199,57 @@ fn send_pairing_reply(ctx: &PairingReplyCtx, code: &str) -> Result<(), String> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a broadcast message to a Discord user via DM.
|
||||
///
|
||||
/// Creates a DM channel with the user (Discord caches this, so repeated calls
|
||||
/// for the same user reuse the existing channel) and then posts the message.
|
||||
fn broadcast_dm(user_id: &str, content: &str) -> Result<(), String> {
|
||||
// Validate user_id is a plausible Discord snowflake (numeric, 17-20 digits)
|
||||
// to avoid injecting arbitrary strings into API URLs.
|
||||
if user_id.is_empty()
|
||||
|| !user_id.chars().all(|c| c.is_ascii_digit())
|
||||
|| user_id.len() < 17
|
||||
|| user_id.len() > 20
|
||||
{
|
||||
return Err(format!("Invalid Discord user ID: '{}'", user_id));
|
||||
}
|
||||
|
||||
// Step 1: Open (or reuse) a DM channel with the target user.
|
||||
let create_dm_payload = serde_json::json!({ "recipient_id": user_id });
|
||||
let create_dm_bytes = serde_json::to_vec(&create_dm_payload)
|
||||
.map_err(|e| format!("Failed to serialize DM channel request: {}", e))?;
|
||||
|
||||
let dm_response = channel_host::http_request(
|
||||
"POST",
|
||||
&format!("{DISCORD_API_BASE}/users/@me/channels"),
|
||||
&discord_auth_headers_json(true),
|
||||
Some(&create_dm_bytes),
|
||||
Some(10_000),
|
||||
)
|
||||
.map_err(|e| format!("Failed to create DM channel: {}", e))?;
|
||||
|
||||
if !(200..300).contains(&dm_response.status) {
|
||||
let body = String::from_utf8_lossy(&dm_response.body);
|
||||
return Err(format!(
|
||||
"Discord create-DM failed: {} - {}",
|
||||
dm_response.status, body
|
||||
));
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct DmChannelResponse {
|
||||
id: String,
|
||||
}
|
||||
let dm_channel: DmChannelResponse = serde_json::from_slice(&dm_response.body)
|
||||
.map_err(|e| format!("Failed to parse DM channel response: {}", e))?;
|
||||
let channel_id = &dm_channel.id;
|
||||
|
||||
// Step 2: Send the message to the DM channel.
|
||||
let truncated = truncate_message(content);
|
||||
let payload = serde_json::json!({ "content": truncated });
|
||||
send_channel_message(channel_id, payload)
|
||||
}
|
||||
|
||||
fn json_response(status: u16, value: serde_json::Value) -> OutgoingHttpResponse {
|
||||
let body = serde_json::to_vec(&value).unwrap_or_default();
|
||||
let headers = serde_json::json!({"Content-Type": "application/json"});
|
||||
@@ -1593,4 +1649,43 @@ mod tests {
|
||||
assert_eq!(interaction.interaction_type, 2);
|
||||
assert!(interaction.data.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_broadcast_dm_payload_format() {
|
||||
// Verify the DM channel creation payload is well-formed JSON that
|
||||
// Discord's API expects.
|
||||
let user_id = "123456789012345678";
|
||||
let payload = serde_json::json!({ "recipient_id": user_id });
|
||||
let serialized = serde_json::to_vec(&payload).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_slice(&serialized).unwrap();
|
||||
assert_eq!(
|
||||
parsed.get("recipient_id").and_then(|v| v.as_str()),
|
||||
Some(user_id)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_broadcast_message_truncation() {
|
||||
// Broadcast uses truncate_message, verify it handles content within
|
||||
// Discord's 2000-char limit for DMs.
|
||||
let short = "Hello from broadcast";
|
||||
assert_eq!(truncate_message(short), short);
|
||||
|
||||
let long = "x".repeat(2500);
|
||||
let result = truncate_message(&long);
|
||||
assert!(result.len() <= 2006); // 1990 content + 16 suffix
|
||||
assert!(result.ends_with("\n... (truncated)"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_broadcast_dm_validates_snowflake() {
|
||||
// broadcast_dm rejects invalid Discord snowflake IDs before making
|
||||
// any API calls. We can call it directly since invalid IDs are
|
||||
// rejected before any host function is invoked.
|
||||
assert!(broadcast_dm("", "hi").is_err());
|
||||
assert!(broadcast_dm("abc", "hi").is_err());
|
||||
assert!(broadcast_dm("12345", "hi").is_err()); // too short
|
||||
assert!(broadcast_dm("123456789012345678901", "hi").is_err()); // too long
|
||||
assert!(broadcast_dm("12345678901234567x", "hi").is_err()); // non-digit
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
-- Add source_channel to conversations for cross-channel approval authorization.
|
||||
-- Tracks which channel originally created a conversation so that approval
|
||||
-- messages from other channels can be validated.
|
||||
ALTER TABLE conversations ADD COLUMN source_channel TEXT;
|
||||
+2
-35
@@ -843,10 +843,7 @@ impl Agent {
|
||||
{
|
||||
use crate::agent::session::Thread;
|
||||
let mut sess = session.lock().await;
|
||||
// Bootstrap thread has no incoming message -- use the
|
||||
// "__bootstrap__" sentinel so approvals from any channel are
|
||||
// permitted. None means "deny by default" (fail-closed).
|
||||
let thread = Thread::with_id(id, sess.id, Some("__bootstrap__"));
|
||||
let thread = Thread::with_id(id, sess.id);
|
||||
sess.active_thread = Some(id);
|
||||
sess.threads.entry(id).or_insert(thread);
|
||||
}
|
||||
@@ -1151,37 +1148,7 @@ impl Agent {
|
||||
.get_or_create_session(&message.user_id)
|
||||
.await;
|
||||
let mut sess = session.lock().await;
|
||||
if let Some(thread) = sess.threads.get(&target_thread_id) {
|
||||
// Verify the thread actually has a pending approval before
|
||||
// allowing approval-shaped messages to target it. Without this
|
||||
// check, an attacker could use approval messages to hijack any
|
||||
// thread by UUID.
|
||||
if thread.pending_approval.is_none() {
|
||||
tracing::warn!(
|
||||
%target_thread_id,
|
||||
approval_channel = %message.channel,
|
||||
"Blocked approval for thread with no pending approval"
|
||||
);
|
||||
drop(sess);
|
||||
return Ok(Some("Error: no pending approval on this thread".into()));
|
||||
}
|
||||
|
||||
let authorized = crate::agent::session::is_approval_authorized(
|
||||
thread.source_channel.as_deref(),
|
||||
&message.channel,
|
||||
);
|
||||
if !authorized {
|
||||
tracing::warn!(
|
||||
%target_thread_id,
|
||||
source_channel = ?thread.source_channel,
|
||||
approval_channel = %message.channel,
|
||||
"Blocked cross-channel approval attempt"
|
||||
);
|
||||
drop(sess);
|
||||
return Ok(Some(
|
||||
"Error: approval not authorized for this channel".into(),
|
||||
));
|
||||
}
|
||||
if sess.threads.contains_key(&target_thread_id) {
|
||||
sess.active_thread = Some(target_thread_id);
|
||||
sess.last_active_at = chrono::Utc::now();
|
||||
drop(sess);
|
||||
|
||||
@@ -319,7 +319,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_format_turns() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.start_turn("Hello");
|
||||
thread.complete_turn("Hi there");
|
||||
thread.start_turn("How are you?");
|
||||
@@ -351,7 +351,7 @@ mod tests {
|
||||
/// Helper: build a thread with `n` completed turns.
|
||||
/// Turn `i` has user_input "msg-{i}" and response "resp-{i}".
|
||||
fn make_thread(n: usize) -> Thread {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
for i in 0..n {
|
||||
thread.start_turn(format!("msg-{}", i));
|
||||
thread.complete_turn(format!("resp-{}", i));
|
||||
@@ -457,7 +457,7 @@ mod tests {
|
||||
async fn test_compact_truncate_empty_turns() {
|
||||
let llm = Arc::new(StubLlm::new("unused"));
|
||||
let compactor = make_compactor(llm);
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
assert!(thread.turns.is_empty());
|
||||
|
||||
let result = compactor
|
||||
@@ -698,7 +698,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_format_turns_for_storage_with_tool_calls() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.start_turn("Search for X");
|
||||
// Record a tool call on the current turn
|
||||
if let Some(turn) = thread.turns.last_mut() {
|
||||
@@ -719,7 +719,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_format_turns_for_storage_incomplete_turn() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.start_turn("In progress message");
|
||||
// Don't complete the turn
|
||||
|
||||
|
||||
+22
-2
@@ -1269,6 +1269,14 @@ pub(crate) fn extract_suggestions(text: &str) -> (String, Vec<String>) {
|
||||
(cleaned, suggestions)
|
||||
}
|
||||
|
||||
/// Remove `<suggestions>` tags from a response, returning only the cleaned text.
|
||||
///
|
||||
/// Convenience wrapper around [`extract_suggestions`] for callers that don't
|
||||
/// need the parsed suggestion list (e.g. job worker, plan completion check).
|
||||
pub(crate) fn strip_suggestions(text: &str) -> String {
|
||||
extract_suggestions(text).0
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
@@ -2299,7 +2307,7 @@ mod tests {
|
||||
// Initialize a thread in the session so the loop can record tool calls.
|
||||
let thread_id = {
|
||||
let mut sess = session.lock().await;
|
||||
sess.create_thread(Some("test")).id
|
||||
sess.create_thread().id
|
||||
};
|
||||
|
||||
let message = IncomingMessage::new("test", "test-user", "do something");
|
||||
@@ -2412,7 +2420,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("test-user")));
|
||||
let thread_id = {
|
||||
let mut sess = session.lock().await;
|
||||
sess.create_thread(Some("test")).id
|
||||
sess.create_thread().id
|
||||
};
|
||||
|
||||
let message = IncomingMessage::new("test", "test-user", "keep calling tools");
|
||||
@@ -2539,6 +2547,18 @@ mod tests {
|
||||
assert_eq!(suggestions, vec!["ok"]); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_strip_suggestions_removes_tags() {
|
||||
let input = "The job is complete.\n<suggestions>[\"Check logs\"]</suggestions>";
|
||||
assert_eq!(super::strip_suggestions(input), "The job is complete."); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_strip_suggestions_no_tag_passthrough() {
|
||||
let input = "Plain text without tags.";
|
||||
assert_eq!(super::strip_suggestions(input), input); // safety: test
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_tool_error_format_includes_tool_name() {
|
||||
let tool_name = "http";
|
||||
|
||||
@@ -36,6 +36,7 @@ pub(crate) use agent_loop::truncate_for_preview;
|
||||
pub use agent_loop::{Agent, AgentDeps};
|
||||
pub use compaction::{CompactionResult, ContextCompactor};
|
||||
pub use context_monitor::{CompactionStrategy, ContextBreakdown, ContextMonitor};
|
||||
pub(crate) use dispatcher::strip_suggestions;
|
||||
pub use heartbeat::{
|
||||
HeartbeatConfig, HeartbeatResult, HeartbeatRunner, spawn_heartbeat, spawn_multi_user_heartbeat,
|
||||
};
|
||||
|
||||
@@ -265,8 +265,13 @@ fn default_max_tokens() -> u32 {
|
||||
4096
|
||||
}
|
||||
|
||||
/// Default max agentic loop iterations for full_job routines.
|
||||
///
|
||||
/// Raised from 10 to 25 to accommodate multi-step tool chains that
|
||||
/// stalled at the old cap. Worst-case LLM cost is 2.5x higher per run;
|
||||
/// callers needing tighter budgets should set `max_iterations` explicitly.
|
||||
fn default_max_iterations() -> u32 {
|
||||
10
|
||||
25
|
||||
}
|
||||
|
||||
fn default_max_tool_rounds() -> u32 {
|
||||
|
||||
@@ -1292,11 +1292,23 @@ async fn execute_full_job(
|
||||
}
|
||||
metadata["notify_user"] = serde_json::json!(&routine.notify.user);
|
||||
|
||||
// Prepend execution context so the LLM knows it's already inside a
|
||||
// routine and should execute the task directly — not set up infrastructure.
|
||||
let contextualized_description = format!(
|
||||
"IMPORTANT: You are executing inside routine \"{routine_name}\". \
|
||||
The routine and its schedule are already configured. \
|
||||
Tools and credentials are already set up. \
|
||||
Do NOT create routines, jobs, or try to discover/install/authenticate tools. \
|
||||
Execute the task directly.\n\n{desc}",
|
||||
routine_name = routine.name,
|
||||
desc = execution.description,
|
||||
);
|
||||
|
||||
let job_id = scheduler
|
||||
.dispatch_job(
|
||||
&routine.user_id,
|
||||
execution.title,
|
||||
execution.description,
|
||||
&contextualized_description,
|
||||
Some(metadata),
|
||||
)
|
||||
.await
|
||||
|
||||
+54
-169
@@ -68,8 +68,8 @@ impl Session {
|
||||
}
|
||||
|
||||
/// Create a new thread in this session.
|
||||
pub fn create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
|
||||
let thread = Thread::new(self.id, channel);
|
||||
pub fn create_thread(&mut self) -> &mut Thread {
|
||||
let thread = Thread::new(self.id);
|
||||
let thread_id = thread.id;
|
||||
self.active_thread = Some(thread_id);
|
||||
self.last_active_at = Utc::now();
|
||||
@@ -87,9 +87,9 @@ impl Session {
|
||||
}
|
||||
|
||||
/// Get or create the active thread.
|
||||
pub fn get_or_create_thread(&mut self, channel: Option<&str>) -> &mut Thread {
|
||||
pub fn get_or_create_thread(&mut self) -> &mut Thread {
|
||||
match self.active_thread {
|
||||
None => self.create_thread(channel),
|
||||
None => self.create_thread(),
|
||||
Some(id) => {
|
||||
if self.threads.contains_key(&id) {
|
||||
// Entry existence confirmed by contains_key above.
|
||||
@@ -100,7 +100,7 @@ impl Session {
|
||||
} else {
|
||||
// Stale active_thread ID: create a new thread, which
|
||||
// updates self.active_thread to the new thread's ID.
|
||||
self.create_thread(channel)
|
||||
self.create_thread()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -225,9 +225,6 @@ pub struct Thread {
|
||||
/// Messages queued while the thread was processing a turn.
|
||||
#[serde(default, skip_serializing_if = "VecDeque::is_empty")]
|
||||
pub pending_messages: VecDeque<String>,
|
||||
/// Channel that created this thread (for approval authorization).
|
||||
#[serde(default)]
|
||||
pub source_channel: Option<String>,
|
||||
}
|
||||
|
||||
/// Maximum number of messages that can be queued while a thread is processing.
|
||||
@@ -236,29 +233,9 @@ pub struct Thread {
|
||||
/// rapid follow-ups. The drain loop processes them as one newline-delimited turn.
|
||||
pub const MAX_PENDING_MESSAGES: usize = 10;
|
||||
|
||||
/// Sentinel value for bootstrap threads that accept approvals from any channel.
|
||||
pub const BOOTSTRAP_SOURCE_CHANNEL: &str = "__bootstrap__";
|
||||
|
||||
/// Check whether an approval from `requesting_channel` is authorized for a
|
||||
/// thread whose `source_channel` is `source`.
|
||||
///
|
||||
/// Rules:
|
||||
/// - `None` (unknown origin) -> denied (fail-closed)
|
||||
/// - `Some("__bootstrap__")` -> authorized from any channel
|
||||
/// - `Some(src) == requesting` -> same channel, authorized
|
||||
/// - requesting is "web" or "gateway" -> always authorized (trusted UI)
|
||||
/// - Otherwise -> denied
|
||||
pub fn is_approval_authorized(source: Option<&str>, requesting: &str) -> bool {
|
||||
match source {
|
||||
None => false,
|
||||
Some(src) if src == BOOTSTRAP_SOURCE_CHANNEL => true,
|
||||
Some(src) => src == requesting || requesting == "web" || requesting == "gateway",
|
||||
}
|
||||
}
|
||||
|
||||
impl Thread {
|
||||
/// Create a new thread.
|
||||
pub fn new(session_id: Uuid, source_channel: Option<&str>) -> Self {
|
||||
pub fn new(session_id: Uuid) -> Self {
|
||||
let now = Utc::now();
|
||||
Self {
|
||||
id: Uuid::new_v4(),
|
||||
@@ -271,12 +248,11 @@ impl Thread {
|
||||
pending_approval: None,
|
||||
pending_auth: None,
|
||||
pending_messages: VecDeque::new(),
|
||||
source_channel: source_channel.map(String::from),
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a thread with a specific ID (for DB hydration).
|
||||
pub fn with_id(id: Uuid, session_id: Uuid, source_channel: Option<&str>) -> Self {
|
||||
pub fn with_id(id: Uuid, session_id: Uuid) -> Self {
|
||||
let now = Utc::now();
|
||||
Self {
|
||||
id,
|
||||
@@ -289,7 +265,6 @@ impl Thread {
|
||||
pending_approval: None,
|
||||
pending_auth: None,
|
||||
pending_messages: VecDeque::new(),
|
||||
source_channel: source_channel.map(String::from),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -812,13 +787,13 @@ mod tests {
|
||||
let mut session = Session::new("user-123");
|
||||
assert!(session.active_thread.is_none());
|
||||
|
||||
session.create_thread(None);
|
||||
session.create_thread();
|
||||
assert!(session.active_thread.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_turns() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("Hello");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
@@ -831,7 +806,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_messages() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("First message");
|
||||
thread.complete_turn("First response");
|
||||
@@ -854,7 +829,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// First add some turns
|
||||
thread.start_turn("Original message");
|
||||
@@ -880,7 +855,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_incomplete_turn() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Messages with incomplete last turn (no assistant response)
|
||||
let messages = vec![
|
||||
@@ -899,7 +874,7 @@ mod tests {
|
||||
#[test]
|
||||
fn test_enter_auth_mode() {
|
||||
let before = Utc::now();
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
assert!(thread.pending_auth.is_none());
|
||||
|
||||
thread.enter_auth_mode("telegram".to_string());
|
||||
@@ -912,7 +887,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_take_pending_auth() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.enter_auth_mode("notion".to_string());
|
||||
|
||||
let pending = thread.take_pending_auth();
|
||||
@@ -927,7 +902,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_pending_auth_serialization() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.enter_auth_mode("openai".to_string());
|
||||
|
||||
let json = serde_json::to_string(&thread).expect("should serialize");
|
||||
@@ -957,7 +932,7 @@ mod tests {
|
||||
#[test]
|
||||
fn test_pending_auth_default_none() {
|
||||
// Deserialization of old data without pending_auth should default to None
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.pending_auth = None;
|
||||
let json = serde_json::to_string(&thread).expect("serialize");
|
||||
|
||||
@@ -971,7 +946,7 @@ mod tests {
|
||||
fn test_thread_with_id() {
|
||||
let specific_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let thread = Thread::with_id(specific_id, session_id, None);
|
||||
let thread = Thread::with_id(specific_id, session_id);
|
||||
|
||||
assert_eq!(thread.id, specific_id);
|
||||
assert_eq!(thread.session_id, session_id);
|
||||
@@ -983,7 +958,7 @@ mod tests {
|
||||
fn test_thread_with_id_restore_messages() {
|
||||
let thread_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||
let mut thread = Thread::with_id(thread_id, session_id);
|
||||
|
||||
let messages = vec![
|
||||
ChatMessage::user("Hello from DB"),
|
||||
@@ -1002,7 +977,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_empty() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Add a turn first, then restore with empty vec
|
||||
thread.start_turn("hello");
|
||||
@@ -1018,7 +993,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_only_assistant_messages() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Only assistant messages (no user messages to anchor turns)
|
||||
let messages = vec![
|
||||
@@ -1035,7 +1010,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_multiple_user_messages_in_a_row() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Two user messages with no assistant response between them
|
||||
let messages = vec![
|
||||
@@ -1062,8 +1037,8 @@ mod tests {
|
||||
fn test_thread_switch() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
let t1_id = session.create_thread(None).id;
|
||||
let t2_id = session.create_thread(None).id;
|
||||
let t1_id = session.create_thread().id;
|
||||
let t2_id = session.create_thread().id;
|
||||
|
||||
// After creating two threads, active should be the last one
|
||||
assert_eq!(session.active_thread, Some(t2_id));
|
||||
@@ -1083,8 +1058,8 @@ mod tests {
|
||||
fn test_get_or_create_thread_idempotent() {
|
||||
let mut session = Session::new("user-1");
|
||||
|
||||
let tid1 = session.get_or_create_thread(None).id;
|
||||
let tid2 = session.get_or_create_thread(None).id;
|
||||
let tid1 = session.get_or_create_thread().id;
|
||||
let tid2 = session.get_or_create_thread().id;
|
||||
|
||||
// Should return the same thread (not create a new one each time)
|
||||
assert_eq!(tid1, tid2);
|
||||
@@ -1093,7 +1068,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_truncate_turns() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
for i in 0..5 {
|
||||
thread.start_turn(format!("msg-{}", i));
|
||||
@@ -1117,7 +1092,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_truncate_turns_noop_when_fewer() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("only one");
|
||||
thread.complete_turn("response");
|
||||
@@ -1129,7 +1104,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_interrupt_and_resume() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("do something");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
@@ -1147,7 +1122,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_resume_only_from_interrupted() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Idle thread: resume should be a no-op
|
||||
assert_eq!(thread.state, ThreadState::Idle);
|
||||
@@ -1163,7 +1138,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_turn_fail() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("risky operation");
|
||||
thread.fail_turn("connection timed out");
|
||||
@@ -1179,7 +1154,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_messages_with_incomplete_last_turn() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("first");
|
||||
thread.complete_turn("first reply");
|
||||
@@ -1195,7 +1170,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_serialization_round_trip() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("hello");
|
||||
thread.complete_turn("world");
|
||||
@@ -1213,7 +1188,7 @@ mod tests {
|
||||
#[test]
|
||||
fn test_session_serialization_round_trip() {
|
||||
let mut session = Session::new("user-ser");
|
||||
session.create_thread(None);
|
||||
session.create_thread();
|
||||
session.auto_approve_tool("echo");
|
||||
|
||||
let json = serde_json::to_string(&session).unwrap();
|
||||
@@ -1251,7 +1226,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_turn_number_increments() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Before any turns, turn_number() is 1 (1-indexed for display)
|
||||
assert_eq!(thread.turn_number(), 1);
|
||||
@@ -1266,7 +1241,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_complete_turn_on_empty_thread() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Completing a turn when there are no turns should be a safe no-op
|
||||
thread.complete_turn("phantom response");
|
||||
@@ -1276,7 +1251,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_fail_turn_on_empty_thread() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Failing a turn when there are no turns should be a safe no-op
|
||||
thread.fail_turn("phantom error");
|
||||
@@ -1286,7 +1261,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_pending_approval_flow() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let approval = PendingApproval {
|
||||
request_id: Uuid::new_v4(),
|
||||
@@ -1313,7 +1288,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_clear_pending_approval() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let approval = PendingApproval {
|
||||
request_id: Uuid::new_v4(),
|
||||
@@ -1342,7 +1317,7 @@ mod tests {
|
||||
assert!(session.active_thread().is_none());
|
||||
assert!(session.active_thread_mut().is_none());
|
||||
|
||||
let tid = session.create_thread(None).id;
|
||||
let tid = session.create_thread().id;
|
||||
|
||||
assert!(session.active_thread().is_some());
|
||||
assert_eq!(session.active_thread().unwrap().id, tid);
|
||||
@@ -1359,7 +1334,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_messages_includes_tool_calls() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("Search for X");
|
||||
{
|
||||
@@ -1391,7 +1366,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_messages_multiple_tool_calls_per_turn() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("Do two things");
|
||||
{
|
||||
@@ -1418,7 +1393,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_with_tool_calls() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Build a message sequence with tool calls
|
||||
let tc = ToolCall {
|
||||
@@ -1450,7 +1425,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_from_messages_with_tool_error() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let tc = ToolCall {
|
||||
id: "call_0".to_string(),
|
||||
@@ -1481,7 +1456,7 @@ mod tests {
|
||||
fn test_messages_round_trip_with_tools() {
|
||||
// Build a thread with tool calls, get messages(), restore, get messages() again
|
||||
// The two message sequences should be equivalent.
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("Do search");
|
||||
{
|
||||
@@ -1494,7 +1469,7 @@ mod tests {
|
||||
let messages_original = thread.messages();
|
||||
|
||||
// Restore into a new thread
|
||||
let mut thread2 = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread2 = Thread::new(Uuid::new_v4());
|
||||
thread2.restore_from_messages(messages_original.clone());
|
||||
|
||||
let messages_restored = thread2.messages();
|
||||
@@ -1516,7 +1491,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_restore_multi_stage_tool_calls() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
let tc1 = ToolCall {
|
||||
id: "call_a".to_string(),
|
||||
@@ -1559,7 +1534,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_messages_truncates_large_tool_results() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
thread.start_turn("Read big file");
|
||||
{
|
||||
@@ -1582,7 +1557,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_message_queue() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Queue is initially empty
|
||||
assert!(thread.pending_messages.is_empty());
|
||||
@@ -1618,7 +1593,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_message_queue_serialization() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Empty queue should not appear in serialization (skip_serializing_if)
|
||||
let json = serde_json::to_string(&thread).unwrap();
|
||||
@@ -1638,7 +1613,7 @@ mod tests {
|
||||
#[test]
|
||||
fn test_thread_message_queue_default_on_old_data() {
|
||||
// Deserialization of old data without pending_messages should default to empty
|
||||
let thread = Thread::new(Uuid::new_v4(), None);
|
||||
let thread = Thread::new(Uuid::new_v4());
|
||||
let json = serde_json::to_string(&thread).unwrap();
|
||||
|
||||
// The field is absent (skip_serializing_if), simulating old data
|
||||
@@ -1649,7 +1624,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_interrupt_clears_pending_messages() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Start a turn so there's something to interrupt
|
||||
thread.start_turn("initial input");
|
||||
@@ -1668,7 +1643,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_thread_state_idle_after_full_drain() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Simulate a full drain cycle: start turn, queue messages, complete turn,
|
||||
// then drain all queued messages as a single merged turn (#259).
|
||||
@@ -1696,7 +1671,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_drain_pending_messages_merges_with_newlines() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Empty queue returns None
|
||||
assert!(thread.drain_pending_messages().is_none());
|
||||
@@ -1725,7 +1700,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_requeue_drained_preserves_content_at_front() {
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
|
||||
// Re-queue into empty queue
|
||||
thread.requeue_drained("failed batch".to_string());
|
||||
@@ -1836,94 +1811,4 @@ mod tests {
|
||||
&serde_json::json!("done")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_new_stores_source_channel() {
|
||||
let thread = Thread::new(Uuid::new_v4(), Some("telegram"));
|
||||
assert_eq!(thread.source_channel.as_deref(), Some("telegram"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_thread_new_none_channel() {
|
||||
let thread = Thread::new(Uuid::new_v4(), None);
|
||||
assert!(thread.source_channel.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_source_channel_serde_backcompat() {
|
||||
// Simulate deserializing a Thread from older DB records that lack source_channel.
|
||||
let thread = Thread::new(Uuid::new_v4(), Some("cli"));
|
||||
let json = serde_json::to_string(&thread).unwrap();
|
||||
|
||||
// Remove the source_channel field to simulate an old record.
|
||||
let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
|
||||
value.as_object_mut().unwrap().remove("source_channel");
|
||||
let old_json = serde_json::to_string(&value).unwrap();
|
||||
|
||||
let deserialized: Thread = serde_json::from_str(&old_json).unwrap();
|
||||
assert!(
|
||||
deserialized.source_channel.is_none(),
|
||||
"missing source_channel should deserialize as None"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_same_channel() {
|
||||
assert!(
|
||||
is_approval_authorized(Some("telegram"), "telegram"),
|
||||
"same channel should be authorized"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_different_channel_blocked() {
|
||||
assert!(
|
||||
!is_approval_authorized(Some("telegram"), "http"),
|
||||
"different channel should be blocked"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_web_always_allowed() {
|
||||
assert!(
|
||||
is_approval_authorized(Some("telegram"), "web"),
|
||||
"web channel should always be authorized"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_gateway_always_allowed() {
|
||||
assert!(
|
||||
is_approval_authorized(Some("telegram"), "gateway"),
|
||||
"gateway channel should always be authorized"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_none_denied() {
|
||||
assert!(
|
||||
!is_approval_authorized(None, "telegram"),
|
||||
"None source_channel should be denied (fail-closed)"
|
||||
);
|
||||
assert!(
|
||||
!is_approval_authorized(None, "web"),
|
||||
"None source_channel should be denied even for web"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_approval_authorized_bootstrap_any_channel() {
|
||||
assert!(
|
||||
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "telegram"),
|
||||
"__bootstrap__ should be authorized from any channel"
|
||||
);
|
||||
assert!(
|
||||
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "http"),
|
||||
"__bootstrap__ should be authorized from any channel"
|
||||
);
|
||||
assert!(
|
||||
is_approval_authorized(Some(BOOTSTRAP_SOURCE_CHANNEL), "cli"),
|
||||
"__bootstrap__ should be authorized from any channel"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -200,7 +200,7 @@ impl SessionManager {
|
||||
// Create new thread (always create a new one for a new key)
|
||||
let thread_id = {
|
||||
let mut sess = session.lock().await;
|
||||
let thread = sess.create_thread(Some(channel));
|
||||
let thread = sess.create_thread();
|
||||
thread.id
|
||||
};
|
||||
|
||||
@@ -476,7 +476,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-hydrate")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(thread_id, sess.id, None);
|
||||
let thread = Thread::with_id(thread_id, sess.id);
|
||||
sess.threads.insert(thread_id, thread);
|
||||
sess.active_thread = Some(thread_id);
|
||||
}
|
||||
@@ -600,7 +600,7 @@ mod tests {
|
||||
// Simulate hydration: create thread with a known UUID
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(known_uuid, session_id, None);
|
||||
let thread = Thread::with_id(known_uuid, session_id);
|
||||
sess.threads.insert(known_uuid, thread);
|
||||
}
|
||||
|
||||
@@ -627,7 +627,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-idem")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
@@ -656,7 +656,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-undo")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
@@ -680,7 +680,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-new")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
@@ -788,7 +788,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
@@ -815,7 +815,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-cross")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
|
||||
@@ -966,7 +966,7 @@ mod tests {
|
||||
let adopted_id = Uuid::new_v4();
|
||||
{
|
||||
let mut sess = session1.lock().await;
|
||||
let thread = Thread::with_id(adopted_id, sess.id, None);
|
||||
let thread = Thread::with_id(adopted_id, sess.id);
|
||||
sess.threads.insert(adopted_id, thread);
|
||||
}
|
||||
// Resolve with the UUID as external_thread_id -- should adopt it
|
||||
@@ -992,7 +992,7 @@ mod tests {
|
||||
let session = Arc::new(Mutex::new(Session::new("user-direct")));
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(tid, sess.id, None);
|
||||
let thread = Thread::with_id(tid, sess.id);
|
||||
sess.threads.insert(tid, thread);
|
||||
}
|
||||
{
|
||||
@@ -1030,7 +1030,7 @@ mod tests {
|
||||
let known_id = Uuid::new_v4();
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(known_id, sess.id, None);
|
||||
let thread = Thread::with_id(known_id, sess.id);
|
||||
sess.threads.insert(known_id, thread);
|
||||
}
|
||||
|
||||
@@ -1057,7 +1057,7 @@ mod tests {
|
||||
let known_id = Uuid::new_v4();
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(known_id, sess.id, None);
|
||||
let thread = Thread::with_id(known_id, sess.id);
|
||||
sess.threads.insert(known_id, thread);
|
||||
}
|
||||
|
||||
@@ -1081,7 +1081,7 @@ mod tests {
|
||||
let known_id = Uuid::new_v4();
|
||||
{
|
||||
let mut sess = session.lock().await;
|
||||
let thread = Thread::with_id(known_id, sess.id, None);
|
||||
let thread = Thread::with_id(known_id, sess.id);
|
||||
sess.threads.insert(known_id, thread);
|
||||
}
|
||||
|
||||
@@ -1102,19 +1102,4 @@ mod tests {
|
||||
"should NOT adopt UUID when external_thread_id is None"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_thread_stores_source_channel() {
|
||||
let manager = SessionManager::new();
|
||||
|
||||
let (session, thread_id) = manager.resolve_thread("user-1", "telegram", None).await;
|
||||
|
||||
let sess = session.lock().await;
|
||||
let thread = sess.threads.get(&thread_id).unwrap();
|
||||
assert_eq!(
|
||||
thread.source_channel.as_deref(),
|
||||
Some("telegram"),
|
||||
"resolve_thread should store source_channel from the channel parameter"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+9
-25
@@ -135,29 +135,13 @@ impl Agent {
|
||||
msg_count = 0;
|
||||
}
|
||||
|
||||
// Create thread with the historical ID and restore messages.
|
||||
// Read source_channel from DB so the authorization check uses the
|
||||
// original creator's channel, not the requesting message's channel.
|
||||
let db_source_channel = if let Some(store) = self.store() {
|
||||
store
|
||||
.get_conversation_source_channel(thread_uuid)
|
||||
.await
|
||||
.unwrap_or(None)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let effective_source_channel = db_source_channel.as_deref();
|
||||
|
||||
// Create thread with the historical ID and restore messages
|
||||
let session_id = {
|
||||
let sess = session.lock().await;
|
||||
sess.id
|
||||
};
|
||||
|
||||
let mut thread = crate::agent::session::Thread::with_id(
|
||||
thread_uuid,
|
||||
session_id,
|
||||
effective_source_channel,
|
||||
);
|
||||
let mut thread = crate::agent::session::Thread::with_id(thread_uuid, session_id);
|
||||
if !chat_messages.is_empty() {
|
||||
thread.restore_from_messages(chat_messages);
|
||||
}
|
||||
@@ -652,7 +636,7 @@ impl Agent {
|
||||
user_id: &str,
|
||||
) -> bool {
|
||||
match store
|
||||
.ensure_conversation(thread_id, channel, user_id, None, Some(channel))
|
||||
.ensure_conversation(thread_id, channel, user_id, None)
|
||||
.await
|
||||
{
|
||||
Ok(true) => true,
|
||||
@@ -1797,7 +1781,7 @@ impl Agent {
|
||||
.get_or_create_session(&message.user_id)
|
||||
.await;
|
||||
let mut sess = session.lock().await;
|
||||
let thread = sess.create_thread(Some(&message.channel));
|
||||
let thread = sess.create_thread();
|
||||
let thread_id = thread.id;
|
||||
Ok(SubmissionResult::ok_with_message(format!(
|
||||
"New thread: {}",
|
||||
@@ -2133,7 +2117,7 @@ mod tests {
|
||||
|
||||
let session_id = Uuid::new_v4();
|
||||
let thread_id = Uuid::new_v4();
|
||||
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||
let mut thread = Thread::with_id(thread_id, session_id);
|
||||
|
||||
// Set thread to AwaitingApproval with a pending tool approval
|
||||
let pending = PendingApproval {
|
||||
@@ -2201,7 +2185,7 @@ mod tests {
|
||||
use crate::agent::session::{MAX_PENDING_MESSAGES, Thread, ThreadState};
|
||||
use uuid::Uuid;
|
||||
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.start_turn("processing something");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
|
||||
@@ -2227,7 +2211,7 @@ mod tests {
|
||||
use crate::agent::session::{Thread, ThreadState};
|
||||
use uuid::Uuid;
|
||||
|
||||
let mut thread = Thread::new(Uuid::new_v4(), None);
|
||||
let mut thread = Thread::new(Uuid::new_v4());
|
||||
thread.start_turn("processing");
|
||||
|
||||
thread.queue_message("pending-1".to_string());
|
||||
@@ -2257,7 +2241,7 @@ mod tests {
|
||||
|
||||
let thread_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||
let mut thread = Thread::with_id(thread_id, session_id);
|
||||
thread.start_turn("working");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
|
||||
@@ -2284,7 +2268,7 @@ mod tests {
|
||||
|
||||
let thread_id = Uuid::new_v4();
|
||||
let session_id = Uuid::new_v4();
|
||||
let mut thread = Thread::with_id(thread_id, session_id, None);
|
||||
let mut thread = Thread::with_id(thread_id, session_id);
|
||||
thread.start_turn("working");
|
||||
assert_eq!(thread.state, ThreadState::Processing);
|
||||
|
||||
|
||||
+13
-24
@@ -492,10 +492,12 @@ impl Channel for ReplChannel {
|
||||
|
||||
async fn start(&self) -> Result<MessageStream, ChannelError> {
|
||||
let (tx, rx) = mpsc::channel(32);
|
||||
// Approval prompts inject responses back through this sender.
|
||||
// In single-message mode we keep it until the turn finishes, then
|
||||
// drop it after enqueuing /quit so the receiver stream can close.
|
||||
if let Ok(mut guard) = self.msg_tx.lock() {
|
||||
// Store tx so send_status can inject approval responses directly.
|
||||
// Skip for single-message mode — no interactive approval is needed
|
||||
// and the extra sender would keep the stream open after /quit.
|
||||
if self.single_message.is_none()
|
||||
&& let Ok(mut guard) = self.msg_tx.lock()
|
||||
{
|
||||
*guard = Some(tx.clone());
|
||||
}
|
||||
let single_message = self.single_message.clone();
|
||||
@@ -914,8 +916,10 @@ mod tests {
|
||||
|
||||
use super::*;
|
||||
|
||||
/// Regression: single-message mode must close the stream after the one
|
||||
/// message so callers (and tests) don't hang forever.
|
||||
#[tokio::test]
|
||||
async fn single_message_mode_sends_message_then_quit() {
|
||||
async fn single_message_mode_sends_message_and_closes_stream() {
|
||||
let repl = ReplChannel::with_message("hi".to_string());
|
||||
let mut stream = repl.start().await.expect("repl start should succeed");
|
||||
|
||||
@@ -926,30 +930,15 @@ mod tests {
|
||||
assert_eq!(first.channel, "repl");
|
||||
assert_eq!(first.content, "hi");
|
||||
|
||||
assert!(
|
||||
timeout(Duration::from_millis(100), stream.next())
|
||||
.await
|
||||
.is_err(),
|
||||
"single-message mode should wait for the turn to finish before quitting"
|
||||
);
|
||||
|
||||
repl.respond(&first, OutgoingResponse::text("done"))
|
||||
.await
|
||||
.expect("respond should succeed");
|
||||
|
||||
let second = timeout(Duration::from_secs(1), stream.next())
|
||||
.await
|
||||
.expect("timed out waiting for quit message")
|
||||
.expect("quit message missing");
|
||||
assert_eq!(second.channel, "repl");
|
||||
assert_eq!(second.content, "/quit");
|
||||
|
||||
// The spawned thread sent the message and returned, dropping its
|
||||
// sender. Because we skip storing a clone in msg_tx for single-
|
||||
// message mode, the stream should close immediately.
|
||||
assert!(
|
||||
timeout(Duration::from_secs(1), stream.next())
|
||||
.await
|
||||
.expect("timed out waiting for stream to close")
|
||||
.is_none(),
|
||||
"stream should end after /quit"
|
||||
"stream should end after the single message"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -34,7 +34,6 @@ pub async fn setup_wasm_channels(
|
||||
secrets_store: &Option<Arc<dyn SecretsStore + Send + Sync>>,
|
||||
extension_manager: Option<&Arc<ExtensionManager>>,
|
||||
database: Option<&Arc<dyn Database>>,
|
||||
registered_channel_names: &[String],
|
||||
) -> Option<WasmChannelSetup> {
|
||||
let runtime = match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) {
|
||||
Ok(r) => Arc::new(r),
|
||||
@@ -72,46 +71,7 @@ pub async fn setup_wasm_channels(
|
||||
let mut channels: Vec<(String, Box<dyn crate::channels::Channel>)> = Vec::new();
|
||||
let mut channel_names: Vec<String> = Vec::new();
|
||||
|
||||
// Reserved channel names that WASM modules must not claim.
|
||||
// A malicious module could otherwise register as a trusted built-in
|
||||
// channel and bypass cross-channel authorization checks.
|
||||
// This list must cover every built-in channel name to prevent a WASM
|
||||
// module from impersonating a built-in and satisfying same-channel
|
||||
// approval checks.
|
||||
const RESERVED_CHANNEL_NAMES: &[&str] = &[
|
||||
"web",
|
||||
"gateway",
|
||||
"cli",
|
||||
"repl",
|
||||
"http",
|
||||
"signal",
|
||||
"slack-relay",
|
||||
"secret_save",
|
||||
];
|
||||
|
||||
for loaded in results.loaded {
|
||||
let name_lower = loaded.name().to_ascii_lowercase();
|
||||
if RESERVED_CHANNEL_NAMES.contains(&name_lower.as_str()) {
|
||||
tracing::warn!(
|
||||
channel = %loaded.name(),
|
||||
"Rejected WASM channel with reserved name"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
// Also reject any name that collides with an already-registered
|
||||
// channel to prevent a WASM module from shadowing a channel that
|
||||
// was registered earlier in the startup sequence.
|
||||
if registered_channel_names
|
||||
.iter()
|
||||
.any(|n| n.to_ascii_lowercase() == name_lower)
|
||||
{
|
||||
tracing::warn!(
|
||||
channel = %loaded.name(),
|
||||
"Rejected WASM channel that collides with already-registered channel"
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
let (name, channel) = register_channel(
|
||||
loaded,
|
||||
config,
|
||||
|
||||
@@ -574,7 +574,7 @@ pub async fn chat_new_thread_handler(
|
||||
.await;
|
||||
let (thread_id, info) = {
|
||||
let mut sess = session.lock().await;
|
||||
let thread = sess.create_thread(Some("web"));
|
||||
let thread = sess.create_thread();
|
||||
let id = thread.id;
|
||||
let info = ThreadInfo {
|
||||
id: thread.id,
|
||||
@@ -593,13 +593,7 @@ pub async fn chat_new_thread_handler(
|
||||
// so that the subsequent loadThreads() call from the frontend sees it.
|
||||
if let Some(ref store) = state.store {
|
||||
match store
|
||||
.ensure_conversation(
|
||||
thread_id,
|
||||
"gateway",
|
||||
&identity.user_id,
|
||||
None,
|
||||
Some("gateway"),
|
||||
)
|
||||
.ensure_conversation(thread_id, "gateway", &identity.user_id, None)
|
||||
.await
|
||||
{
|
||||
Ok(true) => {}
|
||||
|
||||
@@ -236,6 +236,18 @@ pub async fn jobs_detail_handler(
|
||||
(end - start).num_seconds().max(0) as u64
|
||||
});
|
||||
|
||||
// Build transitions from the job's state transition history.
|
||||
let transitions: Vec<TransitionInfo> = ctx
|
||||
.transitions
|
||||
.iter()
|
||||
.map(|t| TransitionInfo {
|
||||
from: t.from.to_string(),
|
||||
to: t.to.to_string(),
|
||||
timestamp: t.timestamp.to_rfc3339(),
|
||||
reason: t.reason.clone(),
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Only show prompt bar for jobs that have a running worker (Pending/InProgress).
|
||||
// Stuck jobs have no active worker loop, so messages would be silently dropped.
|
||||
let is_promptable = matches!(
|
||||
@@ -255,7 +267,7 @@ pub async fn jobs_detail_handler(
|
||||
project_dir: None,
|
||||
browse_url: None,
|
||||
job_mode: None,
|
||||
transitions: Vec::new(),
|
||||
transitions,
|
||||
can_restart: state.scheduler.is_some(),
|
||||
can_prompt: is_promptable && state.scheduler.is_some(),
|
||||
job_kind: Some("agent".to_string()),
|
||||
@@ -643,25 +655,28 @@ pub async fn jobs_events_handler(
|
||||
.parse()
|
||||
.map_err(|_| (StatusCode::BAD_REQUEST, "Invalid job ID".to_string()))?;
|
||||
|
||||
// Verify ownership before returning events.
|
||||
match store.get_sandbox_job(job_id).await {
|
||||
Ok(Some(job)) => {
|
||||
if job.user_id != user.user_id {
|
||||
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
|
||||
// Verify ownership before returning events (check both sandbox and agent jobs).
|
||||
let is_owner = match store.get_sandbox_job(job_id).await {
|
||||
Ok(Some(job)) => job.user_id == user.user_id,
|
||||
Ok(None) => {
|
||||
// Fall back to agent job ownership check.
|
||||
match store.get_job(job_id).await {
|
||||
Ok(Some(ctx)) => ctx.user_id == user.user_id,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(db_error("jobs_handler", e));
|
||||
return Err(db_error("jobs_events_handler", e));
|
||||
}
|
||||
};
|
||||
if !is_owner {
|
||||
return Err((StatusCode::NOT_FOUND, "Job not found".to_string()));
|
||||
}
|
||||
|
||||
let events = store
|
||||
.list_job_events(job_id, None)
|
||||
.await
|
||||
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
|
||||
.map_err(|e| db_error("jobs_events_handler", e))?;
|
||||
|
||||
let events_json: Vec<serde_json::Value> = events
|
||||
.into_iter()
|
||||
|
||||
@@ -122,6 +122,16 @@ pub async fn routines_detail_handler(
|
||||
.collect();
|
||||
let routine_info = RoutineInfo::from_routine(&routine);
|
||||
|
||||
// Read-only lookup — do not create a conversation on a GET request.
|
||||
// The conversation is created lazily when the routine first executes.
|
||||
let conversation_id = store
|
||||
.find_routine_conversation(routine.id, &routine.user_id)
|
||||
.await
|
||||
.unwrap_or_else(|e| {
|
||||
tracing::warn!(routine_id = %routine.id, error = %e, "Failed to look up routine conversation");
|
||||
None
|
||||
});
|
||||
|
||||
Ok(Json(RoutineDetailResponse {
|
||||
id: routine.id,
|
||||
name: routine.name.clone(),
|
||||
@@ -139,6 +149,7 @@ pub async fn routines_detail_handler(
|
||||
run_count: routine.run_count,
|
||||
consecutive_failures: routine.consecutive_failures,
|
||||
created_at: routine.created_at.to_rfc3339(),
|
||||
conversation_id,
|
||||
recent_runs,
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -347,10 +347,10 @@ impl Channel for GatewayChannel {
|
||||
let thread_id = match &msg.thread_id {
|
||||
Some(tid) => tid.clone(),
|
||||
None => {
|
||||
tracing::warn!(
|
||||
"Gateway respond with no thread_id — skipping (clients would drop it)"
|
||||
);
|
||||
return Ok(());
|
||||
return Err(ChannelError::MissingRoutingTarget {
|
||||
name: "gateway".to_string(),
|
||||
reason: "respond() requires a thread_id on the incoming message".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
@@ -507,10 +507,10 @@ impl Channel for GatewayChannel {
|
||||
let thread_id = match response.thread_id {
|
||||
Some(tid) => tid,
|
||||
None => {
|
||||
tracing::warn!(
|
||||
"Gateway broadcast with no thread_id — skipping (clients would drop it)"
|
||||
);
|
||||
return Ok(());
|
||||
return Err(ChannelError::MissingRoutingTarget {
|
||||
name: "gateway".to_string(),
|
||||
reason: "broadcast() requires a thread_id on the response".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
self.state.sse.broadcast_for_user(
|
||||
|
||||
@@ -2014,7 +2014,7 @@ async fn chat_new_thread_handler(
|
||||
let session = session_manager.get_or_create_session(&user.user_id).await;
|
||||
let (thread_id, info) = {
|
||||
let mut sess = session.lock().await;
|
||||
let thread = sess.create_thread(Some("gateway"));
|
||||
let thread = sess.create_thread();
|
||||
let id = thread.id;
|
||||
let info = ThreadInfo {
|
||||
id: thread.id,
|
||||
@@ -2033,7 +2033,7 @@ async fn chat_new_thread_handler(
|
||||
// so that the subsequent loadThreads() call from the frontend sees it.
|
||||
if let Some(ref store) = state.store {
|
||||
match store
|
||||
.ensure_conversation(thread_id, "gateway", &user.user_id, None, Some("gateway"))
|
||||
.ensure_conversation(thread_id, "gateway", &user.user_id, None)
|
||||
.await
|
||||
{
|
||||
Ok(true) => {}
|
||||
|
||||
@@ -4265,6 +4265,13 @@ function renderRoutineDetail(routine) {
|
||||
html += '<div class="job-description"><h3>Action</h3>'
|
||||
+ '<pre class="action-json">' + escapeHtml(JSON.stringify(routine.action, null, 2)) + '</pre></div>';
|
||||
|
||||
// Conversation thread link
|
||||
if (routine.conversation_id) {
|
||||
html += '<div class="job-description">'
|
||||
+ '<a href="#" data-action="view-routine-thread" data-id="' + escapeHtml(routine.conversation_id) + '" class="btn-primary" style="display:inline-block;margin:0.5rem 0">'
|
||||
+ 'View Execution Thread</a></div>';
|
||||
}
|
||||
|
||||
// Recent runs
|
||||
if (routine.recent_runs && routine.recent_runs.length > 0) {
|
||||
html += '<div class="job-timeline-section"><h3>Recent Runs</h3>'
|
||||
@@ -6190,6 +6197,11 @@ document.addEventListener('click', function(e) {
|
||||
switchTab('jobs');
|
||||
openJobDetail(el.dataset.id);
|
||||
break;
|
||||
case 'view-routine-thread':
|
||||
e.preventDefault();
|
||||
switchTab('chat');
|
||||
switchThread(el.dataset.id);
|
||||
break;
|
||||
case 'copy-tee-report':
|
||||
copyTeeReport();
|
||||
break;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
//! Integration tests for the web gateway module.
|
||||
|
||||
mod multi_tenant;
|
||||
mod no_silent_drop;
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
//! Regression tests: the gateway channel must never silently drop messages.
|
||||
//!
|
||||
//! Previously, `respond()` and `broadcast()` returned `Ok(())` when thread_id
|
||||
//! was missing, making callers believe the message was delivered when it wasn't.
|
||||
//! These tests ensure that missing routing info produces an explicit error.
|
||||
|
||||
use crate::channels::channel::{Channel, IncomingMessage, OutgoingResponse};
|
||||
use crate::channels::web::GatewayChannel;
|
||||
use crate::config::GatewayConfig;
|
||||
use crate::error::ChannelError;
|
||||
|
||||
fn test_gateway() -> GatewayChannel {
|
||||
GatewayChannel::new(
|
||||
GatewayConfig {
|
||||
host: "127.0.0.1".to_string(),
|
||||
port: 0,
|
||||
auth_token: Some("test-token".to_string()),
|
||||
workspace_read_scopes: vec![],
|
||||
memory_layers: vec![],
|
||||
},
|
||||
"test-user".to_string(),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_respond_without_thread_id_returns_error() {
|
||||
let gw = test_gateway();
|
||||
let msg = IncomingMessage::new("gateway", "test-user", "hello");
|
||||
// msg has no thread_id by default
|
||||
assert!(msg.thread_id.is_none());
|
||||
|
||||
let response = OutgoingResponse::text("reply");
|
||||
let result = gw.respond(&msg, response).await;
|
||||
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"respond() must not silently succeed without thread_id"
|
||||
);
|
||||
assert!(
|
||||
matches!(result, Err(ChannelError::MissingRoutingTarget { .. })),
|
||||
"Expected MissingRoutingTarget, got: {:?}",
|
||||
result
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_respond_with_thread_id_succeeds() {
|
||||
let gw = test_gateway();
|
||||
let mut msg = IncomingMessage::new("gateway", "test-user", "hello");
|
||||
msg.thread_id = Some("thread-123".to_string());
|
||||
|
||||
let response = OutgoingResponse::text("reply");
|
||||
let result = gw.respond(&msg, response).await;
|
||||
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"respond() should succeed with thread_id: {:?}",
|
||||
result
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_broadcast_without_thread_id_returns_error() {
|
||||
let gw = test_gateway();
|
||||
let response = OutgoingResponse::text("notification");
|
||||
// response has no thread_id by default
|
||||
|
||||
let result = gw.broadcast("test-user", response).await;
|
||||
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"broadcast() must not silently succeed without thread_id"
|
||||
);
|
||||
assert!(
|
||||
matches!(result, Err(ChannelError::MissingRoutingTarget { .. })),
|
||||
"Expected MissingRoutingTarget, got: {:?}",
|
||||
result
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn gateway_broadcast_with_thread_id_succeeds() {
|
||||
let gw = test_gateway();
|
||||
let response = OutgoingResponse::text("notification").in_thread("thread-456".to_string());
|
||||
|
||||
let result = gw.broadcast("test-user", response).await;
|
||||
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"broadcast() should succeed with thread_id: {:?}",
|
||||
result
|
||||
);
|
||||
}
|
||||
@@ -768,6 +768,7 @@ pub struct RoutineDetailResponse {
|
||||
pub run_count: u64,
|
||||
pub consecutive_failures: u32,
|
||||
pub created_at: String,
|
||||
pub conversation_id: Option<Uuid>,
|
||||
pub recent_runs: Vec<RoutineRunInfo>,
|
||||
}
|
||||
|
||||
|
||||
@@ -67,20 +67,19 @@ impl ConversationStore for LibSqlBackend {
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
source_channel: Option<&str>,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
let affected = conn
|
||||
.execute(
|
||||
r#"
|
||||
INSERT INTO conversations (id, channel, user_id, thread_id, source_channel, started_at, last_activity)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6)
|
||||
INSERT INTO conversations (id, channel, user_id, thread_id, started_at, last_activity)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?5)
|
||||
ON CONFLICT (id) DO UPDATE SET last_activity = excluded.last_activity
|
||||
WHERE conversations.user_id = excluded.user_id
|
||||
AND conversations.channel = excluded.channel
|
||||
"#,
|
||||
params![id.to_string(), channel, user_id, opt_text(thread_id), opt_text(source_channel), now],
|
||||
params![id.to_string(), channel, user_id, opt_text(thread_id), now],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
@@ -291,6 +290,41 @@ impl ConversationStore for LibSqlBackend {
|
||||
result
|
||||
}
|
||||
|
||||
async fn find_routine_conversation(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<Option<Uuid>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let rid = routine_id.to_string();
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id FROM conversations
|
||||
WHERE user_id = ?1 AND json_extract(metadata, '$.routine_id') = ?2
|
||||
LIMIT 1
|
||||
"#,
|
||||
params![user_id, rid],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
if let Some(row) = rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
let id_str: String = row.get(0).map_err(|e| {
|
||||
DatabaseError::Query(format!("Failed to read conversation id: {e}"))
|
||||
})?;
|
||||
let id = id_str
|
||||
.parse()
|
||||
.map_err(|_| DatabaseError::Serialization("Invalid UUID".to_string()))?;
|
||||
return Ok(Some(id));
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Uses BEGIN IMMEDIATE to serialize concurrent writers and prevent
|
||||
/// duplicate heartbeat conversations (TOCTOU race).
|
||||
async fn get_or_create_heartbeat_conversation(
|
||||
@@ -566,28 +600,6 @@ impl ConversationStore for LibSqlBackend {
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
|
||||
async fn get_conversation_source_channel(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Option<String>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT source_channel FROM conversations WHERE id = ?1",
|
||||
params![conversation_id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(get_opt_text(&row, 0)),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -981,7 +981,7 @@ mod tests {
|
||||
assert!(alice_stats.last_active_at.is_some());
|
||||
|
||||
// Bob has no LLM calls so doesn't appear in summary stats
|
||||
assert!(!stats.iter().any(|s| s.user_id == "bob"));
|
||||
assert!(stats.iter().find(|s| s.user_id == "bob").is_none());
|
||||
|
||||
// Filter to single user
|
||||
let alice_only = db.user_summary_stats(Some("alice")).await.unwrap();
|
||||
|
||||
@@ -785,14 +785,6 @@ CREATE TABLE IF NOT EXISTS api_tokens (
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_api_tokens_user ON api_tokens(user_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash);
|
||||
"#,
|
||||
),
|
||||
(
|
||||
15,
|
||||
"conversation_source_channel",
|
||||
// Add source_channel to conversations for cross-channel approval authorization.
|
||||
r#"
|
||||
ALTER TABLE conversations ADD COLUMN source_channel TEXT;
|
||||
"#,
|
||||
),
|
||||
];
|
||||
|
||||
+7
-6
@@ -373,7 +373,6 @@ pub trait ConversationStore: Send + Sync {
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
source_channel: Option<&str>,
|
||||
) -> Result<bool, DatabaseError>;
|
||||
async fn list_conversations_with_preview(
|
||||
&self,
|
||||
@@ -392,6 +391,13 @@ pub trait ConversationStore: Send + Sync {
|
||||
routine_name: &str,
|
||||
user_id: &str,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
/// Read-only lookup for an existing routine conversation. Returns `None`
|
||||
/// if the routine has never executed (no conversation created yet).
|
||||
async fn find_routine_conversation(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<Option<Uuid>, DatabaseError>;
|
||||
async fn get_or_create_heartbeat_conversation(
|
||||
&self,
|
||||
user_id: &str,
|
||||
@@ -432,11 +438,6 @@ pub trait ConversationStore: Send + Sync {
|
||||
conversation_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<bool, DatabaseError>;
|
||||
/// Get the source_channel for a conversation (the channel that created it).
|
||||
async fn get_conversation_source_channel(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Option<String>, DatabaseError>;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
+11
-11
@@ -99,10 +99,9 @@ impl ConversationStore for PgBackend {
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
source_channel: Option<&str>,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
self.store
|
||||
.ensure_conversation(id, channel, user_id, thread_id, source_channel)
|
||||
.ensure_conversation(id, channel, user_id, thread_id)
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -138,6 +137,16 @@ impl ConversationStore for PgBackend {
|
||||
.await
|
||||
}
|
||||
|
||||
async fn find_routine_conversation(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<Option<Uuid>, DatabaseError> {
|
||||
self.store
|
||||
.find_routine_conversation(routine_id, user_id)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_or_create_heartbeat_conversation(
|
||||
&self,
|
||||
user_id: &str,
|
||||
@@ -213,15 +222,6 @@ impl ConversationStore for PgBackend {
|
||||
.conversation_belongs_to_user(conversation_id, user_id)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_conversation_source_channel(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Option<String>, DatabaseError> {
|
||||
self.store
|
||||
.get_conversation_source_channel(conversation_id)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== JobStore ====================
|
||||
|
||||
+24
-19
@@ -1581,20 +1581,19 @@ impl Store {
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
source_channel: Option<&str>,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
let conn = self.conn().await?;
|
||||
let affected = conn
|
||||
.execute(
|
||||
r#"
|
||||
INSERT INTO conversations (id, channel, user_id, thread_id, source_channel)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
INSERT INTO conversations (id, channel, user_id, thread_id)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (id) DO UPDATE
|
||||
SET last_activity = NOW()
|
||||
WHERE conversations.user_id = EXCLUDED.user_id
|
||||
AND conversations.channel = EXCLUDED.channel
|
||||
"#,
|
||||
&[&id, &channel, &user_id, &thread_id, &source_channel],
|
||||
&[&id, &channel, &user_id, &thread_id],
|
||||
)
|
||||
.await?;
|
||||
Ok(affected > 0)
|
||||
@@ -1772,6 +1771,27 @@ impl Store {
|
||||
Ok(row.get("id"))
|
||||
}
|
||||
|
||||
/// Read-only lookup for an existing routine conversation.
|
||||
pub async fn find_routine_conversation(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<Option<Uuid>, DatabaseError> {
|
||||
let conn = self.conn().await?;
|
||||
let rid = routine_id.to_string();
|
||||
let row = conn
|
||||
.query_opt(
|
||||
r#"
|
||||
SELECT id FROM conversations
|
||||
WHERE user_id = $1 AND metadata->>'routine_id' = $2
|
||||
LIMIT 1
|
||||
"#,
|
||||
&[&user_id, &rid],
|
||||
)
|
||||
.await?;
|
||||
Ok(row.map(|r| r.get("id")))
|
||||
}
|
||||
|
||||
/// Get or create the singleton heartbeat conversation for a user.
|
||||
///
|
||||
/// Looks for a conversation where `metadata->>'thread_type' = 'heartbeat'`.
|
||||
@@ -1893,21 +1913,6 @@ impl Store {
|
||||
Ok(row.is_some())
|
||||
}
|
||||
|
||||
/// Get the source_channel for a conversation.
|
||||
pub async fn get_conversation_source_channel(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Option<String>, DatabaseError> {
|
||||
let conn = self.conn().await?;
|
||||
let row = conn
|
||||
.query_opt(
|
||||
"SELECT source_channel FROM conversations WHERE id = $1",
|
||||
&[&conversation_id],
|
||||
)
|
||||
.await?;
|
||||
Ok(row.and_then(|r| r.get::<_, Option<String>>(0)))
|
||||
}
|
||||
|
||||
/// Load messages for a conversation with cursor-based pagination.
|
||||
///
|
||||
/// Returns `(messages_oldest_first, has_more)`.
|
||||
|
||||
@@ -234,6 +234,7 @@ fn is_transient(err: &LlmError) -> bool {
|
||||
LlmError::RequestFailed { .. }
|
||||
| LlmError::RateLimited { .. }
|
||||
| LlmError::InvalidResponse { .. }
|
||||
| LlmError::EmptyResponse { .. }
|
||||
| LlmError::SessionExpired { .. }
|
||||
| LlmError::SessionRenewalFailed { .. }
|
||||
| LlmError::Http(_)
|
||||
|
||||
@@ -17,6 +17,9 @@ pub enum LlmError {
|
||||
#[error("Invalid response from {provider}: {reason}")]
|
||||
InvalidResponse { provider: String, reason: String },
|
||||
|
||||
#[error("Empty response from {provider}: no content returned")]
|
||||
EmptyResponse { provider: String },
|
||||
|
||||
#[error("Context length exceeded: {used} tokens used, {limit} allowed")]
|
||||
ContextLengthExceeded { used: usize, limit: usize },
|
||||
|
||||
|
||||
@@ -231,9 +231,8 @@ impl LlmProvider for GithubCopilotProvider {
|
||||
.choices
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| LlmError::InvalidResponse {
|
||||
.ok_or_else(|| LlmError::EmptyResponse {
|
||||
provider: "github_copilot".to_string(),
|
||||
reason: "No choices in response".to_string(),
|
||||
})?;
|
||||
|
||||
let (content, _tool_calls) = extract_choice_content(&choice);
|
||||
@@ -309,9 +308,8 @@ impl LlmProvider for GithubCopilotProvider {
|
||||
.choices
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| LlmError::InvalidResponse {
|
||||
.ok_or_else(|| LlmError::EmptyResponse {
|
||||
provider: "github_copilot".to_string(),
|
||||
reason: "No choices in response".to_string(),
|
||||
})?;
|
||||
|
||||
let (content, tool_calls) = extract_choice_content(&choice);
|
||||
|
||||
@@ -490,9 +490,8 @@ impl LlmProvider for NearAiChatProvider {
|
||||
.choices
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| LlmError::InvalidResponse {
|
||||
.ok_or_else(|| LlmError::EmptyResponse {
|
||||
provider: "nearai_chat".to_string(),
|
||||
reason: "No choices in response".to_string(),
|
||||
})?;
|
||||
|
||||
// Fall back to reasoning_content when content is null (same as
|
||||
@@ -570,9 +569,8 @@ impl LlmProvider for NearAiChatProvider {
|
||||
.choices
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| LlmError::InvalidResponse {
|
||||
.ok_or_else(|| LlmError::EmptyResponse {
|
||||
provider: "nearai_chat".to_string(),
|
||||
reason: "No choices in response".to_string(),
|
||||
})?;
|
||||
|
||||
let tool_calls: Vec<ToolCall> = choice
|
||||
|
||||
@@ -276,8 +276,33 @@ impl LlmProvider for OpenAiCodexProvider {
|
||||
&self,
|
||||
request: ToolCompletionRequest,
|
||||
) -> Result<ToolCompletionResponse, LlmError> {
|
||||
// Build a reverse map so we can translate sanitized names back to originals.
|
||||
// Only needed when sanitization actually changes a name (e.g. MCP tools with dots).
|
||||
let name_map: std::collections::HashMap<String, String> = request
|
||||
.tools
|
||||
.iter()
|
||||
.filter_map(|t| {
|
||||
let sanitized = sanitize_tool_name(&t.name);
|
||||
if sanitized != t.name {
|
||||
Some((sanitized, t.name.clone()))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
let body = self.build_request_body(&request.messages, Some(&request.tools));
|
||||
let parsed = self.send_request(body).await?;
|
||||
let mut parsed = self.send_request(body).await?;
|
||||
|
||||
// Reverse-map sanitized tool names back to originals so the caller
|
||||
// can look them up in the tool registry.
|
||||
if !name_map.is_empty() {
|
||||
for tc in &mut parsed.tool_calls {
|
||||
if let Some(original) = name_map.get(&tc.name) {
|
||||
tc.name = original.clone();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let finish_reason = if !parsed.tool_calls.is_empty() {
|
||||
FinishReason::ToolUse
|
||||
@@ -421,7 +446,7 @@ fn convert_message(msg: &ChatMessage, index: usize) -> Vec<serde_json::Value> {
|
||||
serde_json::json!({
|
||||
"type": "function_call",
|
||||
"call_id": tc.id,
|
||||
"name": tc.name,
|
||||
"name": sanitize_tool_name(&tc.name),
|
||||
"arguments": args_str,
|
||||
})
|
||||
})
|
||||
@@ -452,6 +477,20 @@ fn convert_message(msg: &ChatMessage, index: usize) -> Vec<serde_json::Value> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Sanitize a tool name to match the OpenAI Responses API pattern `^[a-zA-Z0-9_-]+$`.
|
||||
/// Replaces any invalid character (e.g. dots in MCP tool names) with underscores.
|
||||
fn sanitize_tool_name(name: &str) -> String {
|
||||
name.chars()
|
||||
.map(|c| {
|
||||
if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
|
||||
c
|
||||
} else {
|
||||
'_'
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Convert a `ToolDefinition` to Responses API tool format.
|
||||
///
|
||||
/// Applies strict-mode schema normalization (same as OpenAI Chat Completions):
|
||||
@@ -461,7 +500,7 @@ fn convert_tool_definition(tool: &ToolDefinition) -> serde_json::Value {
|
||||
|
||||
serde_json::json!({
|
||||
"type": "function",
|
||||
"name": tool.name,
|
||||
"name": sanitize_tool_name(&tool.name),
|
||||
"description": tool.description,
|
||||
"parameters": normalize_schema_strict(&tool.parameters),
|
||||
})
|
||||
@@ -1093,4 +1132,95 @@ data: {"type":"response.completed","response":{"status":"completed","usage":{"in
|
||||
assert_eq!(parsed.tool_calls[1].name, "read_file");
|
||||
assert_eq!(parsed.finish_reason, FinishReason::ToolUse);
|
||||
}
|
||||
|
||||
/// Regression test: tool names with dots (e.g. MCP tools) must be sanitized
|
||||
/// to match OpenAI's `^[a-zA-Z0-9_-]+$` pattern.
|
||||
#[test]
|
||||
fn test_sanitize_tool_name_replaces_dots() {
|
||||
assert_eq!(super::sanitize_tool_name("memory_search"), "memory_search");
|
||||
assert_eq!(
|
||||
super::sanitize_tool_name("mcp.server.tool"),
|
||||
"mcp_server_tool"
|
||||
);
|
||||
assert_eq!(super::sanitize_tool_name("tool@v2"), "tool_v2");
|
||||
assert_eq!(super::sanitize_tool_name("my-tool"), "my-tool");
|
||||
}
|
||||
|
||||
/// Regression test: convert_tool_definition sanitizes the name.
|
||||
#[test]
|
||||
fn test_convert_tool_definition_sanitizes_name() {
|
||||
let tool = ToolDefinition {
|
||||
name: "mcp.server.search".to_string(),
|
||||
description: "Search".to_string(),
|
||||
parameters: serde_json::json!({"type": "object", "properties": {}}),
|
||||
};
|
||||
let json = super::convert_tool_definition(&tool);
|
||||
assert_eq!(json["name"], "mcp_server_search");
|
||||
}
|
||||
|
||||
/// Regression test: function_call items sanitize tool names.
|
||||
#[test]
|
||||
fn test_convert_message_sanitizes_tool_call_name() {
|
||||
let tool_calls = vec![ToolCall {
|
||||
id: "call_1".to_string(),
|
||||
name: "mcp.server.search".to_string(),
|
||||
arguments: serde_json::json!({"q": "test"}),
|
||||
reasoning: None,
|
||||
}];
|
||||
let msg = ChatMessage::assistant_with_tool_calls(None, tool_calls);
|
||||
let items = super::convert_message(&msg, 0);
|
||||
assert_eq!(items[0]["name"], "mcp_server_search");
|
||||
}
|
||||
|
||||
/// Regression: sanitized tool names in API responses must be reverse-mapped
|
||||
/// back to original names so the tool registry can look them up.
|
||||
#[test]
|
||||
fn test_sanitized_name_reverse_mapping() {
|
||||
use std::collections::HashMap;
|
||||
|
||||
let tools = [
|
||||
ToolDefinition {
|
||||
name: "mcp.server.search".to_string(),
|
||||
description: "Search".to_string(),
|
||||
parameters: serde_json::json!({"type": "object", "properties": {}}),
|
||||
},
|
||||
ToolDefinition {
|
||||
name: "memory_search".to_string(),
|
||||
description: "Memory".to_string(),
|
||||
parameters: serde_json::json!({"type": "object", "properties": {}}),
|
||||
},
|
||||
];
|
||||
|
||||
// Build name map (same logic as complete_with_tools)
|
||||
let name_map: HashMap<String, String> = tools
|
||||
.iter()
|
||||
.filter_map(|t| {
|
||||
let sanitized = super::sanitize_tool_name(&t.name);
|
||||
if sanitized != t.name {
|
||||
Some((sanitized, t.name.clone()))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Only the MCP tool should appear (its name changed)
|
||||
assert_eq!(name_map.len(), 1);
|
||||
assert_eq!(
|
||||
name_map.get("mcp_server_search"),
|
||||
Some(&"mcp.server.search".to_string())
|
||||
);
|
||||
|
||||
// Simulate a tool call coming back with the sanitized name
|
||||
let mut tc = ToolCall {
|
||||
id: "call_1".to_string(),
|
||||
name: "mcp_server_search".to_string(),
|
||||
arguments: serde_json::json!({}),
|
||||
reasoning: None,
|
||||
};
|
||||
if let Some(original) = name_map.get(&tc.name) {
|
||||
tc.name = original.clone();
|
||||
}
|
||||
assert_eq!(tc.name, "mcp.server.search");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,6 +48,7 @@ pub(crate) fn is_retryable(err: &LlmError) -> bool {
|
||||
LlmError::RequestFailed { .. }
|
||||
| LlmError::RateLimited { .. }
|
||||
| LlmError::InvalidResponse { .. }
|
||||
| LlmError::EmptyResponse { .. }
|
||||
| LlmError::SessionRenewalFailed { .. }
|
||||
| LlmError::Http(_)
|
||||
| LlmError::Io(_)
|
||||
|
||||
@@ -449,7 +449,6 @@ async fn async_main() -> anyhow::Result<()> {
|
||||
&components.secrets_store,
|
||||
components.extension_manager.as_ref(),
|
||||
components.db.as_ref(),
|
||||
&channel_names,
|
||||
)
|
||||
.await;
|
||||
|
||||
|
||||
+10
-2
@@ -182,8 +182,14 @@ impl SkillCatalog {
|
||||
/// Create a catalog with a custom registry URL (for testing).
|
||||
#[cfg(test)]
|
||||
pub fn with_url(url: &str) -> Self {
|
||||
Self::with_url_and_timeout(url, REQUEST_TIMEOUT)
|
||||
}
|
||||
|
||||
/// Create a catalog with a custom registry URL and timeout (for testing).
|
||||
#[cfg(test)]
|
||||
pub fn with_url_and_timeout(url: &str, timeout: Duration) -> Self {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(REQUEST_TIMEOUT)
|
||||
.timeout(timeout)
|
||||
.user_agent(concat!("ironclaw/", env!("CARGO_PKG_VERSION")))
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
@@ -458,7 +464,9 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn test_search_returns_error_on_network_failure() {
|
||||
// Use RFC 5737 TEST-NET-1 (192.0.2.0/24) for reliable failure even behind proxies.
|
||||
let catalog = SkillCatalog::with_url("http://192.0.2.1:9999");
|
||||
// Short timeout so the test doesn't block for the full 10s REQUEST_TIMEOUT.
|
||||
let catalog =
|
||||
SkillCatalog::with_url_and_timeout("http://192.0.2.1:9999", Duration::from_secs(1));
|
||||
let outcome = catalog.search("test").await;
|
||||
assert!(outcome.results.is_empty());
|
||||
assert!(outcome.error.is_some());
|
||||
|
||||
+1
-1
@@ -278,7 +278,7 @@ impl TenantScope {
|
||||
thread_id: Option<&str>,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
self.inner
|
||||
.ensure_conversation(id, channel, &self.user_id, thread_id, None)
|
||||
.ensure_conversation(id, channel, &self.user_id, thread_id)
|
||||
.await
|
||||
}
|
||||
|
||||
|
||||
+3
-3
@@ -753,7 +753,7 @@ mod tests {
|
||||
|
||||
// ensure_conversation should create the row.
|
||||
assert!(
|
||||
db.ensure_conversation(conv_id, "web", "carol", None, Some("web"))
|
||||
db.ensure_conversation(conv_id, "web", "carol", None)
|
||||
.await
|
||||
.expect("ensure first"),
|
||||
"first ensure_conversation should create the row"
|
||||
@@ -761,7 +761,7 @@ mod tests {
|
||||
|
||||
// Calling again with the same ID should not error.
|
||||
assert!(
|
||||
db.ensure_conversation(conv_id, "web", "carol", None, Some("web"))
|
||||
db.ensure_conversation(conv_id, "web", "carol", None)
|
||||
.await
|
||||
.expect("ensure second (idempotent)"),
|
||||
"second ensure_conversation should touch owned row"
|
||||
@@ -806,7 +806,7 @@ mod tests {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
||||
|
||||
assert!(
|
||||
!db.ensure_conversation(conv_id, "web", "mallory", None, None)
|
||||
!db.ensure_conversation(conv_id, "web", "mallory", None)
|
||||
.await
|
||||
.expect("foreign ensure should not error"),
|
||||
"foreign ensure_conversation should report not ensured"
|
||||
|
||||
@@ -224,7 +224,13 @@ impl Tool for MessageTool {
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
let content = require_str(¶ms, "content")?;
|
||||
// Accept "message" as an alias for "content" — LLMs frequently use
|
||||
// the wrong parameter name in autonomous job execution.
|
||||
let content = require_str(¶ms, "content").or_else(|_| {
|
||||
require_str(¶ms, "message").map_err(|_| {
|
||||
ToolError::InvalidParameters("missing 'content' parameter".to_string())
|
||||
})
|
||||
})?;
|
||||
|
||||
let explicit_channel = params
|
||||
.get("channel")
|
||||
@@ -323,8 +329,11 @@ impl Tool for MessageTool {
|
||||
if !attachments.is_empty() {
|
||||
response = response.with_attachments(attachments);
|
||||
}
|
||||
if channel.as_deref() == Some("gateway")
|
||||
&& response.thread_id.is_none()
|
||||
// Attach thread_id so the gateway can route the message into the
|
||||
// correct conversation. Previously this only fired when channel was
|
||||
// explicitly "gateway", which meant broadcast_all (channel=null) sent
|
||||
// a response without a thread_id and the gateway silently dropped it.
|
||||
if response.thread_id.is_none()
|
||||
&& let Some(thread_id) = metadata_string(&ctx.metadata, "notify_thread_id")
|
||||
{
|
||||
response = response.in_thread(thread_id);
|
||||
@@ -480,6 +489,31 @@ mod tests {
|
||||
assert!(params.get("attachments").is_some());
|
||||
}
|
||||
|
||||
/// Regression: LLMs frequently pass {"message": "..."} instead of
|
||||
/// {"content": "..."}. The tool should accept both.
|
||||
#[tokio::test]
|
||||
async fn message_param_alias_accepted() {
|
||||
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
|
||||
tool.set_context(Some("gateway".to_string()), Some("user".to_string()))
|
||||
.await;
|
||||
|
||||
let ctx = crate::context::JobContext::new("test", "test");
|
||||
|
||||
// "message" alias should not produce InvalidParameters
|
||||
let result = tool
|
||||
.execute(serde_json::json!({"message": "hello from alias"}), &ctx)
|
||||
.await;
|
||||
// Execution may fail for other reasons (no real channel), but
|
||||
// the error must NOT be about a missing 'content' parameter.
|
||||
if let Err(ref e) = result {
|
||||
let msg = e.to_string();
|
||||
assert!(
|
||||
!msg.contains("missing 'content'"),
|
||||
"Should accept 'message' as alias for 'content', got: {msg}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn message_tool_set_context_updates_defaults() {
|
||||
let tool = MessageTool::new(Arc::new(ChannelManager::new()));
|
||||
|
||||
@@ -65,6 +65,7 @@ struct NormalizedExecutionRequest {
|
||||
context_paths: Vec<String>,
|
||||
use_tools: bool,
|
||||
max_tool_rounds: u32,
|
||||
max_iterations: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
@@ -328,6 +329,13 @@ fn full_job_execution_variant() -> Value {
|
||||
"type": "string",
|
||||
"enum": ["full_job"],
|
||||
"description": "Full-job execution mode."
|
||||
},
|
||||
"max_iterations": {
|
||||
"type": "integer",
|
||||
"description": "Maximum LLM iterations for the job (default: 25). Increase for complex multi-step tasks.",
|
||||
"default": 25,
|
||||
"minimum": 1,
|
||||
"maximum": 200
|
||||
}
|
||||
},
|
||||
"required": ["mode"]
|
||||
@@ -644,6 +652,12 @@ pub(crate) fn routine_update_parameters_schema() -> Value {
|
||||
"description": {
|
||||
"type": "string",
|
||||
"description": "New description"
|
||||
},
|
||||
"max_iterations": {
|
||||
"type": "integer",
|
||||
"description": "Maximum LLM iterations for full_job routines (1-200).",
|
||||
"minimum": 1,
|
||||
"maximum": 200
|
||||
}
|
||||
},
|
||||
"required": ["name"]
|
||||
@@ -887,11 +901,16 @@ fn parse_routine_execution(
|
||||
.clamp(1, crate::agent::routine::MAX_TOOL_ROUNDS_LIMIT as u64)
|
||||
as u32;
|
||||
|
||||
let max_iterations = u64_field(params, "execution", "max_iterations", &["max_iterations"])
|
||||
.unwrap_or(25)
|
||||
.clamp(1, 200) as u32;
|
||||
|
||||
Ok(NormalizedExecutionRequest {
|
||||
mode,
|
||||
context_paths,
|
||||
use_tools,
|
||||
max_tool_rounds,
|
||||
max_iterations,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -972,7 +991,7 @@ fn build_routine_action(
|
||||
NormalizedExecutionMode::FullJob => RoutineAction::FullJob {
|
||||
title: name.to_string(),
|
||||
description: prompt.to_string(),
|
||||
max_iterations: 10,
|
||||
max_iterations: execution.max_iterations,
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1317,6 +1336,12 @@ impl Tool for RoutineUpdateTool {
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(iters) = params.get("max_iterations").and_then(|v| v.as_u64())
|
||||
&& let RoutineAction::FullJob { max_iterations, .. } = &mut routine.action
|
||||
{
|
||||
*max_iterations = (iters.clamp(1, 200)) as u32;
|
||||
}
|
||||
|
||||
// Validate timezone param if provided
|
||||
let new_timezone = params
|
||||
.get("timezone")
|
||||
@@ -1544,6 +1569,7 @@ impl Tool for RoutineFireTool {
|
||||
"name": name,
|
||||
"run_id": run_id.to_string(),
|
||||
"status": "fired",
|
||||
"note": "Routine is executing asynchronously. Use routine_history to check the result.",
|
||||
});
|
||||
|
||||
Ok(ToolOutput::success(result, start.elapsed()))
|
||||
@@ -1642,10 +1668,47 @@ impl Tool for RoutineHistoryTool {
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Look up the routine's conversation thread and fetch recent messages
|
||||
// so the user can see the full output of routine runs.
|
||||
let (conversation_id, recent_output) = match self
|
||||
.store
|
||||
.get_or_create_routine_conversation(routine.id, name, &ctx.user_id)
|
||||
.await
|
||||
{
|
||||
Ok(conv_id) => {
|
||||
let messages = self
|
||||
.store
|
||||
.list_conversation_messages_paginated(conv_id, None, limit)
|
||||
.await
|
||||
.map(|(msgs, _)| msgs)
|
||||
.unwrap_or_default();
|
||||
let msg_list: Vec<serde_json::Value> = messages
|
||||
.iter()
|
||||
.map(|m| {
|
||||
serde_json::json!({
|
||||
"role": m.role,
|
||||
"content": m.content,
|
||||
"timestamp": m.created_at.to_rfc3339(),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
(Some(conv_id.to_string()), msg_list)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
routine = %name,
|
||||
"Failed to fetch routine conversation thread: {e}"
|
||||
);
|
||||
(None, Vec::new())
|
||||
}
|
||||
};
|
||||
|
||||
let result = serde_json::json!({
|
||||
"routine": name,
|
||||
"total_runs": routine.run_count,
|
||||
"conversation_id": conversation_id,
|
||||
"runs": run_list,
|
||||
"recent_output": recent_output,
|
||||
});
|
||||
|
||||
Ok(ToolOutput::success(result, start.elapsed()))
|
||||
@@ -2282,8 +2345,8 @@ mod tests {
|
||||
.and_then(Value::as_object)
|
||||
.expect("full_job properties");
|
||||
assert!(
|
||||
full_job_props.len() == 1 && full_job_props.contains_key("mode"),
|
||||
"full_job variant should only expose the execution mode",
|
||||
full_job_props.contains_key("mode") && full_job_props.contains_key("max_iterations"),
|
||||
"full_job variant should expose mode and max_iterations",
|
||||
);
|
||||
}
|
||||
|
||||
@@ -2491,6 +2554,7 @@ mod tests {
|
||||
context_paths: Vec::new(),
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
max_iterations: 25,
|
||||
};
|
||||
|
||||
let action = build_routine_action("issue-1316", "Run it", &execution);
|
||||
@@ -2503,7 +2567,7 @@ mod tests {
|
||||
max_iterations,
|
||||
} if title == "issue-1316"
|
||||
&& description == "Run it"
|
||||
&& max_iterations == 10
|
||||
&& max_iterations == 25
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ use chrono::{DateTime, LocalResult, NaiveDate, NaiveDateTime, TimeZone, Utc};
|
||||
use chrono_tz::Tz;
|
||||
|
||||
use crate::context::JobContext;
|
||||
use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str};
|
||||
use crate::tools::tool::{Tool, ToolError, ToolOutput};
|
||||
|
||||
/// Tool for getting current time and date operations.
|
||||
pub struct TimeTool;
|
||||
@@ -62,7 +62,7 @@ impl Tool for TimeTool {
|
||||
"description": "Second timestamp for diff."
|
||||
}
|
||||
},
|
||||
"required": ["operation"]
|
||||
"required": []
|
||||
})
|
||||
}
|
||||
|
||||
@@ -73,7 +73,10 @@ impl Tool for TimeTool {
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
let operation = require_str(¶ms, "operation")?;
|
||||
let operation = params
|
||||
.get("operation")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("now");
|
||||
|
||||
let result = match operation {
|
||||
"now" => execute_now(¶ms, ctx)?,
|
||||
|
||||
+28
-7
@@ -954,16 +954,22 @@ pub async fn store_tokens(
|
||||
server_config: &McpServerConfig,
|
||||
token: &AccessToken,
|
||||
) -> Result<(), AuthError> {
|
||||
// Store access token
|
||||
let params = CreateSecretParams::new(server_config.token_secret_name(), &token.access_token)
|
||||
.with_provider(format!("mcp:{}", server_config.name));
|
||||
// Store access token (with expiry if provided)
|
||||
let mut params =
|
||||
CreateSecretParams::new(server_config.token_secret_name(), &token.access_token)
|
||||
.with_provider(format!("mcp:{}", server_config.name));
|
||||
|
||||
if let Some(secs) = token.expires_in {
|
||||
let expires_at = chrono::Utc::now() + chrono::Duration::seconds(secs as i64);
|
||||
params = params.with_expiry(expires_at);
|
||||
}
|
||||
|
||||
secrets
|
||||
.create(user_id, params)
|
||||
.await
|
||||
.map_err(|e| AuthError::Secrets(e.to_string()))?;
|
||||
|
||||
// Store refresh token if present
|
||||
// Store refresh token if present (no expiry — long-lived)
|
||||
if let Some(ref refresh_token) = token.refresh_token {
|
||||
let params =
|
||||
CreateSecretParams::new(server_config.refresh_token_secret_name(), refresh_token)
|
||||
@@ -1064,11 +1070,26 @@ pub async fn refresh_access_token(
|
||||
// Get client_id (from config or stored DCR)
|
||||
let client_id = get_client_id(server_config, secrets, user_id).await?;
|
||||
|
||||
// Get the refresh token
|
||||
let refresh_token = secrets
|
||||
// Get the refresh token (try current name, fall back to legacy name for
|
||||
// users who authenticated before the naming convention was fixed).
|
||||
// Only fall back on NotFound/Expired — propagate real errors (DB, decryption).
|
||||
let refresh_token = match secrets
|
||||
.get_decrypted(user_id, &server_config.refresh_token_secret_name())
|
||||
.await
|
||||
.map_err(|e| AuthError::RefreshFailed(format!("No refresh token: {}", e)))?;
|
||||
{
|
||||
Ok(token) => token,
|
||||
Err(crate::secrets::SecretError::NotFound(_) | crate::secrets::SecretError::Expired) => {
|
||||
secrets
|
||||
.get_decrypted(user_id, &server_config.legacy_refresh_token_secret_name())
|
||||
.await
|
||||
.map_err(|e| AuthError::RefreshFailed(format!("No refresh token: {}", e)))?
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(AuthError::RefreshFailed(format!(
|
||||
"Failed to read refresh token: {e}"
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
// Discover the token endpoint
|
||||
let token_url = if let Some(ref oauth) = server_config.oauth {
|
||||
|
||||
@@ -259,6 +259,9 @@ impl McpClient {
|
||||
}
|
||||
|
||||
/// Get the access token for this server (if authenticated).
|
||||
///
|
||||
/// If the stored token has expired, automatically attempts a refresh using
|
||||
/// the stored refresh token before failing.
|
||||
async fn get_access_token(&self) -> Result<Option<String>, ToolError> {
|
||||
let Some(ref secrets) = self.secrets else {
|
||||
return Ok(None);
|
||||
@@ -272,6 +275,33 @@ impl McpClient {
|
||||
{
|
||||
Ok(token) => Ok(Some(token.expose().to_string())),
|
||||
Err(crate::secrets::SecretError::NotFound(_)) => Ok(None),
|
||||
Err(crate::secrets::SecretError::Expired) => {
|
||||
// Token expired — attempt refresh before failing.
|
||||
tracing::info!(
|
||||
server = %self.server_name,
|
||||
"Access token expired, attempting refresh"
|
||||
);
|
||||
match refresh_access_token(config, secrets, &self.user_id).await {
|
||||
Ok(new_token) => {
|
||||
tracing::info!(
|
||||
server = %self.server_name,
|
||||
"Access token refreshed successfully"
|
||||
);
|
||||
Ok(Some(new_token.access_token))
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
server = %self.server_name,
|
||||
"Token refresh failed: {}", e
|
||||
);
|
||||
Err(ToolError::ExternalService(format!(
|
||||
"Failed to get access token: Secret has expired \
|
||||
and refresh failed: {}",
|
||||
e
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => Err(ToolError::ExternalService(format!(
|
||||
"Failed to get access token: {}",
|
||||
e
|
||||
|
||||
@@ -250,7 +250,19 @@ impl McpServerConfig {
|
||||
}
|
||||
|
||||
/// Get the secret name used to store the refresh token.
|
||||
///
|
||||
/// Matches the convention used by the hosted OAuth flow in
|
||||
/// `store_oauth_tokens`: `{token_secret_name}_refresh_token`.
|
||||
pub fn refresh_token_secret_name(&self) -> String {
|
||||
format!("{}_refresh_token", self.token_secret_name())
|
||||
}
|
||||
|
||||
/// Legacy secret name for refresh tokens (pre-v0.22).
|
||||
///
|
||||
/// Earlier versions stored refresh tokens as `mcp_{name}_refresh_token`
|
||||
/// instead of `{token_secret_name}_refresh_token`. Used as a fallback
|
||||
/// during lookup to avoid forcing re-auth on existing users.
|
||||
pub fn legacy_refresh_token_secret_name(&self) -> String {
|
||||
format!("mcp_{}_refresh_token", self.name)
|
||||
}
|
||||
|
||||
@@ -750,8 +762,15 @@ mod tests {
|
||||
fn test_token_secret_names() {
|
||||
let config = McpServerConfig::new("notion", "https://mcp.notion.com");
|
||||
assert_eq!(config.token_secret_name(), "mcp_notion_access_token");
|
||||
// Refresh token name follows the hosted OAuth convention:
|
||||
// {token_secret_name}_refresh_token
|
||||
assert_eq!(
|
||||
config.refresh_token_secret_name(),
|
||||
"mcp_notion_access_token_refresh_token"
|
||||
);
|
||||
// Legacy name used before v0.22 — fallback lookup prevents forced re-auth
|
||||
assert_eq!(
|
||||
config.legacy_refresh_token_secret_name(),
|
||||
"mcp_notion_refresh_token"
|
||||
);
|
||||
}
|
||||
|
||||
+22
@@ -225,4 +225,26 @@ mod tests {
|
||||
"The tool returned: TASK_COMPLETE signal"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signals_completion_after_suggestions_stripped() {
|
||||
// Regression: after stripping <suggestions> tags, the completion
|
||||
// signal should still be detected in the cleaned text.
|
||||
assert!(llm_signals_completion(
|
||||
"The job is complete. All requested work has been finished."
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signals_completion_self_dialogue_pattern() {
|
||||
// Regression: the "not complete" pattern that caused the self-dialogue
|
||||
// loop when left in job context after plan completion.
|
||||
assert!(!llm_signals_completion(
|
||||
"No — the job is **not complete**.\n\n\
|
||||
What still needs to be done:\n\
|
||||
1. Fetch actual meeting note contents\n\
|
||||
2. Create the Notion page\n\
|
||||
3. Send the completion message"
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
+144
-36
@@ -31,6 +31,7 @@ use std::time::Duration;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
#[cfg(not(unix))]
|
||||
use tokio::process::Command;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -340,6 +341,11 @@ impl ClaudeBridgeRuntime {
|
||||
|
||||
/// Spawn a `claude` CLI process and stream its output.
|
||||
///
|
||||
/// Uses a PTY on Unix so Node.js line-buffers stdout instead of
|
||||
/// full-buffering (which causes the bridge to hang on non-TTY pipes).
|
||||
/// Arguments are passed via `execve` (no shell) — injection-safe by
|
||||
/// construction.
|
||||
///
|
||||
/// Returns the session_id if captured from the `system` init message.
|
||||
async fn run_claude_session(
|
||||
&self,
|
||||
@@ -347,47 +353,102 @@ impl ClaudeBridgeRuntime {
|
||||
resume_session_id: Option<&str>,
|
||||
extra_env: &std::collections::HashMap<String, String>,
|
||||
) -> Result<Option<String>, WorkerError> {
|
||||
let mut cmd = Command::new("claude");
|
||||
cmd.arg("-p")
|
||||
.arg(prompt)
|
||||
.arg("--output-format")
|
||||
.arg("stream-json")
|
||||
.arg("--verbose")
|
||||
.arg("--max-turns")
|
||||
.arg(self.config.max_turns.to_string())
|
||||
.arg("--model")
|
||||
.arg(&self.config.model);
|
||||
let max_turns_str = self.config.max_turns.to_string();
|
||||
|
||||
if let Some(sid) = resume_session_id {
|
||||
cmd.arg("--resume").arg(sid);
|
||||
}
|
||||
|
||||
// Inject credentials into the child process environment without
|
||||
// mutating the global process env (which is unsafe in multi-threaded programs).
|
||||
cmd.envs(extra_env);
|
||||
|
||||
cmd.current_dir("/workspace")
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped());
|
||||
|
||||
let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
|
||||
reason: format!("failed to spawn claude: {}", e),
|
||||
})?;
|
||||
|
||||
let stdout = child
|
||||
.stdout
|
||||
.take()
|
||||
.ok_or_else(|| WorkerError::ExecutionFailed {
|
||||
reason: "failed to capture claude stdout".to_string(),
|
||||
// Spawn with PTY on Unix to fix Node.js stdout buffering.
|
||||
// All arguments are passed individually via execve — never through
|
||||
// a shell interpreter. This eliminates shell injection by construction.
|
||||
#[cfg(unix)]
|
||||
let (mut child, stdout, stderr) = {
|
||||
let (pty, pts) = pty_process::open().map_err(|e| WorkerError::ExecutionFailed {
|
||||
reason: format!("failed to allocate PTY: {}", e),
|
||||
})?;
|
||||
|
||||
let stderr = child
|
||||
.stderr
|
||||
.take()
|
||||
.ok_or_else(|| WorkerError::ExecutionFailed {
|
||||
reason: "failed to capture claude stderr".to_string(),
|
||||
let mut cmd = pty_process::Command::new("claude");
|
||||
cmd = cmd
|
||||
.arg("-p")
|
||||
.arg(prompt)
|
||||
.arg("--output-format")
|
||||
.arg("stream-json")
|
||||
.arg("--verbose")
|
||||
.arg("--max-turns")
|
||||
.arg(&max_turns_str)
|
||||
.arg("--model")
|
||||
.arg(&self.config.model);
|
||||
|
||||
if let Some(sid) = resume_session_id {
|
||||
cmd = cmd.arg("--resume").arg(sid);
|
||||
}
|
||||
|
||||
cmd = cmd.envs(extra_env.iter());
|
||||
cmd = cmd.current_dir("/workspace");
|
||||
// Keep stderr on a separate pipe — pty-process attaches the PTY
|
||||
// to all fds by default, which would merge stderr into the PTY
|
||||
// stream and break NDJSON parsing.
|
||||
cmd = cmd.stderr(std::process::Stdio::piped());
|
||||
|
||||
let mut child = cmd.spawn(pts).map_err(|e| WorkerError::ExecutionFailed {
|
||||
reason: format!("failed to spawn claude with PTY: {}", e),
|
||||
})?;
|
||||
|
||||
let stderr = child
|
||||
.stderr
|
||||
.take()
|
||||
.ok_or_else(|| WorkerError::ExecutionFailed {
|
||||
reason: "failed to capture claude stderr".to_string(),
|
||||
})?;
|
||||
|
||||
// stdout comes from the PTY master, which implements AsyncRead
|
||||
let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(pty);
|
||||
(child, stdout, stderr)
|
||||
};
|
||||
|
||||
// Non-Unix fallback (Windows CI) — no PTY, direct spawn.
|
||||
// Claude bridge only runs in Linux Docker containers, so this path
|
||||
// exists solely for compilation on Windows targets.
|
||||
#[cfg(not(unix))]
|
||||
let (mut child, stdout, stderr) = {
|
||||
let mut cmd = Command::new("claude");
|
||||
cmd.arg("-p")
|
||||
.arg(prompt)
|
||||
.arg("--output-format")
|
||||
.arg("stream-json")
|
||||
.arg("--verbose")
|
||||
.arg("--max-turns")
|
||||
.arg(&max_turns_str)
|
||||
.arg("--model")
|
||||
.arg(&self.config.model);
|
||||
|
||||
if let Some(sid) = resume_session_id {
|
||||
cmd.arg("--resume").arg(sid);
|
||||
}
|
||||
|
||||
cmd.envs(extra_env);
|
||||
cmd.current_dir("/workspace")
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped());
|
||||
|
||||
let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
|
||||
reason: format!("failed to spawn claude: {}", e),
|
||||
})?;
|
||||
|
||||
let stdout_pipe = child
|
||||
.stdout
|
||||
.take()
|
||||
.ok_or_else(|| WorkerError::ExecutionFailed {
|
||||
reason: "failed to capture claude stdout".to_string(),
|
||||
})?;
|
||||
let stderr = child
|
||||
.stderr
|
||||
.take()
|
||||
.ok_or_else(|| WorkerError::ExecutionFailed {
|
||||
reason: "failed to capture claude stderr".to_string(),
|
||||
})?;
|
||||
|
||||
let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(stdout_pipe);
|
||||
(child, stdout, stderr)
|
||||
};
|
||||
|
||||
// Spawn stderr reader that forwards lines as log events
|
||||
let client_for_stderr = Arc::clone(&self.client);
|
||||
let job_id = self.config.job_id;
|
||||
@@ -1027,4 +1088,51 @@ mod tests {
|
||||
let copied = copy_dir_recursive(nonexistent, dst.path()).unwrap();
|
||||
assert_eq!(copied, 0);
|
||||
}
|
||||
|
||||
/// Regression test: arguments are passed individually (not via shell string),
|
||||
/// so shell metacharacters in prompt/model/session_id are harmless.
|
||||
#[test]
|
||||
fn command_args_no_shell_interpretation() {
|
||||
// Prompt, model, and session_id may contain shell metacharacters from
|
||||
// user-supplied task descriptions or LLM output. Since we use
|
||||
// Command::arg() (execve), these are passed as literal strings.
|
||||
let prompt = "Fix the user's bug; echo $HOME && rm -rf /";
|
||||
let model = "claude-3-opus-20240229";
|
||||
let session_id = "'; DROP TABLE jobs; --";
|
||||
|
||||
let max_turns = 10u32;
|
||||
let max_turns_str = max_turns.to_string();
|
||||
let args: Vec<&str> = vec![
|
||||
"-p",
|
||||
prompt,
|
||||
"--output-format",
|
||||
"stream-json",
|
||||
"--verbose",
|
||||
"--max-turns",
|
||||
&max_turns_str,
|
||||
"--model",
|
||||
model,
|
||||
"--resume",
|
||||
session_id,
|
||||
];
|
||||
|
||||
// All values present as literal strings — no shell interpretation
|
||||
// ["-p", prompt, "--output-format", "stream-json", "--verbose",
|
||||
// "--max-turns", "10", "--model", model, "--resume", session_id]
|
||||
assert_eq!(args[1], prompt);
|
||||
assert_eq!(args[8], model);
|
||||
assert_eq!(args[10], session_id);
|
||||
// Shell metacharacters preserved, not expanded
|
||||
assert!(args[1].contains("$HOME"));
|
||||
assert!(args[1].contains("&&"));
|
||||
assert!(args[10].contains("'; DROP TABLE"));
|
||||
}
|
||||
|
||||
/// Verify PTY is available on Unix platforms.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn pty_opens_successfully() {
|
||||
let result = pty_process::open();
|
||||
assert!(result.is_ok(), "PTY allocation should succeed on Unix");
|
||||
}
|
||||
}
|
||||
|
||||
+271
-27
@@ -391,6 +391,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
worker: self,
|
||||
rx: tokio::sync::Mutex::new(rx),
|
||||
consecutive_rate_limits: std::sync::atomic::AtomicUsize::new(0),
|
||||
has_text_response: std::sync::atomic::AtomicBool::new(false),
|
||||
};
|
||||
|
||||
let config = AgenticLoopConfig {
|
||||
@@ -827,14 +828,11 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
}),
|
||||
);
|
||||
|
||||
if matches!(
|
||||
&e,
|
||||
Error::Tool(crate::error::ToolError::AutonomousUnavailable { .. })
|
||||
) {
|
||||
Err(e)
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
// All tool errors (including AutonomousUnavailable) are
|
||||
// recoverable — the error message is already recorded in
|
||||
// reason_ctx so the LLM can see it and try a different
|
||||
// approach. Returning Err here would kill the entire job.
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -929,17 +927,31 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
|
||||
// Plan completed, check with LLM if job is done
|
||||
// Plan completed — ask the LLM whether the job is done.
|
||||
let msg_count_before = reason_ctx.messages.len();
|
||||
reason_ctx.messages.push(ChatMessage::user(
|
||||
"All planned actions have been executed. Is the job complete? If not, what else needs to be done?",
|
||||
"All planned actions have been executed. Assess the results: \
|
||||
if the job is fully complete, state that the job is complete. \
|
||||
Otherwise, briefly list what remains.",
|
||||
));
|
||||
|
||||
let response = reasoning.respond(reason_ctx).await?;
|
||||
reason_ctx.messages.push(ChatMessage::assistant(&response));
|
||||
let response = crate::agent::strip_suggestions(&response);
|
||||
|
||||
if crate::util::llm_signals_completion(&response) {
|
||||
reason_ctx.messages.push(ChatMessage::assistant(&response));
|
||||
self.mark_completed().await?;
|
||||
} else {
|
||||
// Replace the completion-check exchange with an action-oriented
|
||||
// continuation prompt. Leaving the "Is the job complete?" / "No"
|
||||
// dialogue in context causes the agentic loop to repeat the same
|
||||
// analysis instead of calling tools (self-dialogue loop).
|
||||
reason_ctx.messages.truncate(msg_count_before);
|
||||
reason_ctx.messages.push(ChatMessage::user(format!(
|
||||
"The planned actions are done but the job is not yet complete. \
|
||||
Remaining work:\n\n{response}\n\n\
|
||||
Continue executing now — use tools to finish the job."
|
||||
)));
|
||||
tracing::info!(
|
||||
"Job {} plan completed but work remains, falling back to direct selection",
|
||||
self.job_id
|
||||
@@ -1101,6 +1113,15 @@ fn store_fallback_in_metadata(
|
||||
}
|
||||
|
||||
/// Job delegate: implements `LoopDelegate` for the background job context.
|
||||
/// Whether an LLM error represents a completion-eligible empty response.
|
||||
///
|
||||
/// Only `EmptyResponse` (provider returned no choices/content) qualifies.
|
||||
/// Infrastructure errors (`AuthFailed`, `Http`, `Io`, etc.) never qualify —
|
||||
/// they must propagate even if prior text output was produced.
|
||||
fn is_completion_eligible_error(error: &crate::error::LlmError) -> bool {
|
||||
matches!(error, crate::error::LlmError::EmptyResponse { .. })
|
||||
}
|
||||
|
||||
///
|
||||
/// Handles: signal channel (stop/ping/user messages), cancellation checks,
|
||||
/// rate-limit retry, parallel tool execution, DB persistence, SSE broadcasting.
|
||||
@@ -1109,6 +1130,10 @@ struct JobDelegate<'a> {
|
||||
rx: tokio::sync::Mutex<&'a mut mpsc::Receiver<WorkerMessage>>,
|
||||
/// Tracks consecutive rate-limit errors to fail fast instead of burning iterations.
|
||||
consecutive_rate_limits: std::sync::atomic::AtomicUsize,
|
||||
/// Whether a substantive (non-empty) text response has been produced.
|
||||
/// When true, an empty follow-up response is treated as job completion
|
||||
/// rather than a retry signal (prevents spurious failures in routines).
|
||||
has_text_response: std::sync::atomic::AtomicBool,
|
||||
}
|
||||
|
||||
impl<'a> JobDelegate<'a> {
|
||||
@@ -1161,6 +1186,53 @@ impl<'a> JobDelegate<'a> {
|
||||
finish_reason: crate::llm::FinishReason::Stop,
|
||||
})
|
||||
}
|
||||
|
||||
/// Mark the job as completed, logging a warning on failure.
|
||||
async fn mark_completed_or_warn(&self, context: &str) {
|
||||
if let Err(e) = self.worker.mark_completed().await {
|
||||
tracing::warn!(
|
||||
job_id = %self.worker.job_id,
|
||||
error = %e,
|
||||
"Failed to mark job completed ({context})"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// If a substantive text response was already produced and the error
|
||||
/// indicates the LLM simply returned nothing, treat it as successful
|
||||
/// completion rather than a fatal failure.
|
||||
///
|
||||
/// Only swallows `EmptyResponse` — infrastructure errors (`AuthFailed`,
|
||||
/// `ContextLengthExceeded`, `Http`, `Io`, etc.) always propagate.
|
||||
///
|
||||
/// Returns `Some(empty RespondOutput)` when the error should be swallowed,
|
||||
/// `None` when it should propagate normally.
|
||||
async fn try_complete_on_error(
|
||||
&self,
|
||||
context: &str,
|
||||
error: &crate::error::LlmError,
|
||||
) -> Option<crate::llm::RespondOutput> {
|
||||
if !is_completion_eligible_error(error) {
|
||||
return None;
|
||||
}
|
||||
if !self
|
||||
.has_text_response
|
||||
.load(std::sync::atomic::Ordering::Relaxed)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
tracing::info!(
|
||||
job_id = %self.worker.job_id,
|
||||
error = %error,
|
||||
"{context} empty response after text output — treating as completion"
|
||||
);
|
||||
self.mark_completed_or_warn(context).await;
|
||||
Some(crate::llm::RespondOutput {
|
||||
result: RespondResult::Text(String::new()),
|
||||
usage: crate::llm::TokenUsage::default(),
|
||||
finish_reason: crate::llm::FinishReason::Stop,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -1291,7 +1363,12 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
|
||||
Err(crate::error::LlmError::RateLimited { retry_after, .. }) => {
|
||||
return self.handle_rate_limit(retry_after, "tool selection").await;
|
||||
}
|
||||
Err(e) => return Err(e.into()),
|
||||
Err(e) => {
|
||||
if let Some(output) = self.try_complete_on_error("select_tools", &e).await {
|
||||
return Ok(output);
|
||||
}
|
||||
return Err(e.into());
|
||||
}
|
||||
};
|
||||
|
||||
// Fall back to respond_with_tools
|
||||
@@ -1321,7 +1398,12 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
|
||||
self.handle_rate_limit(retry_after, "respond_with_tools")
|
||||
.await
|
||||
}
|
||||
Err(e) => Err(e.into()),
|
||||
Err(e) => {
|
||||
if let Some(output) = self.try_complete_on_error("respond_with_tools", &e).await {
|
||||
return Ok(output);
|
||||
}
|
||||
Err(e.into())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1330,26 +1412,47 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
|
||||
text: &str,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> TextAction {
|
||||
// Empty text from rate-limit backoff retry — skip processing and let the
|
||||
// loop proceed to the next iteration which will re-call the LLM.
|
||||
// Empty text after a substantive response means the LLM has finished.
|
||||
// Treat as successful completion rather than continuing the loop (which
|
||||
// would produce "Response contained no message or tool call (empty)").
|
||||
if text.is_empty() {
|
||||
if self
|
||||
.has_text_response
|
||||
.load(std::sync::atomic::Ordering::Relaxed)
|
||||
{
|
||||
tracing::debug!(
|
||||
job_id = %self.worker.job_id,
|
||||
"Empty response after text output — treating as completion"
|
||||
);
|
||||
self.mark_completed_or_warn("empty text response").await;
|
||||
return TextAction::Return(LoopOutcome::Response(String::new()));
|
||||
}
|
||||
// No prior text response — this is likely a rate-limit backoff retry.
|
||||
return TextAction::Continue;
|
||||
}
|
||||
|
||||
// Check for explicit completion
|
||||
if crate::util::llm_signals_completion(text) {
|
||||
if let Err(e) = self.worker.mark_completed().await {
|
||||
tracing::warn!(
|
||||
"Failed to mark job {} as completed: {}",
|
||||
self.worker.job_id,
|
||||
e
|
||||
);
|
||||
}
|
||||
return TextAction::Return(LoopOutcome::Response(text.to_string()));
|
||||
// Jobs run autonomously — strip <suggestions> tags that are only
|
||||
// meaningful for interactive chat sessions.
|
||||
let text = crate::agent::strip_suggestions(text);
|
||||
|
||||
// A non-empty text response with no tool intent (already filtered
|
||||
// by the agentic loop's nudge mechanism) is the LLM's final answer.
|
||||
// Mark the job complete and stop the loop. Without this, the LLM
|
||||
// restates its summary every iteration until the cap is hit.
|
||||
if let Err(e) = self.worker.mark_completed().await {
|
||||
tracing::warn!(
|
||||
"Failed to mark job {} as completed: {}",
|
||||
self.worker.job_id,
|
||||
e
|
||||
);
|
||||
}
|
||||
|
||||
// Track that a substantive response has been produced.
|
||||
self.has_text_response
|
||||
.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
// Add assistant response to context
|
||||
reason_ctx.messages.push(ChatMessage::assistant(text));
|
||||
reason_ctx.messages.push(ChatMessage::assistant(&text));
|
||||
|
||||
self.worker.log_event(
|
||||
"message",
|
||||
@@ -1359,7 +1462,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
|
||||
}),
|
||||
);
|
||||
|
||||
TextAction::Continue
|
||||
TextAction::Return(LoopOutcome::Response(text))
|
||||
}
|
||||
|
||||
async fn execute_tool_calls(
|
||||
@@ -1368,6 +1471,9 @@ impl<'a> LoopDelegate for JobDelegate<'a> {
|
||||
content: Option<String>,
|
||||
reason_ctx: &mut ReasoningContext,
|
||||
) -> Result<Option<LoopOutcome>, crate::error::Error> {
|
||||
// Strip suggestions from accompanying text (not useful in job context).
|
||||
let content = content.map(|c| crate::agent::strip_suggestions(&c));
|
||||
|
||||
if let Some(ref text) = content {
|
||||
self.worker.log_event(
|
||||
"message",
|
||||
@@ -2068,6 +2174,53 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Regression: a text response without rigid completion phrases (e.g.
|
||||
/// "Weekly review completed and saved to Notion") must still terminate the
|
||||
/// agentic loop and mark the job complete, rather than continuing until
|
||||
/// max_iterations.
|
||||
#[tokio::test]
|
||||
async fn test_text_response_terminates_loop_without_explicit_completion_phrase() {
|
||||
let worker = make_worker(vec![]).await;
|
||||
worker
|
||||
.context_manager()
|
||||
.update_context(worker.job_id, |ctx| {
|
||||
ctx.transition_to(JobState::InProgress, None)
|
||||
})
|
||||
.await
|
||||
.unwrap() // safety: test
|
||||
.unwrap(); // safety: test
|
||||
|
||||
let (_, mut rx) = tokio::sync::mpsc::channel(1);
|
||||
let delegate = JobDelegate {
|
||||
worker: &worker,
|
||||
rx: tokio::sync::Mutex::new(&mut rx),
|
||||
consecutive_rate_limits: std::sync::atomic::AtomicUsize::new(0),
|
||||
has_text_response: std::sync::atomic::AtomicBool::new(false),
|
||||
};
|
||||
|
||||
let mut reason_ctx = ReasoningContext::new();
|
||||
|
||||
// Text that a real LLM would produce but doesn't match llm_signals_completion
|
||||
let action = delegate
|
||||
.handle_text_response(
|
||||
"Weekly review created in Notion and notification sent.",
|
||||
&mut reason_ctx,
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
matches!(action, TextAction::Return(_)),
|
||||
"Text response should terminate the loop, got Continue"
|
||||
); // safety: test
|
||||
|
||||
let ctx = worker
|
||||
.context_manager()
|
||||
.get_context(worker.job_id)
|
||||
.await
|
||||
.unwrap(); // safety: test
|
||||
assert_eq!(ctx.state, JobState::Completed); // safety: test
|
||||
}
|
||||
|
||||
/// Regression test: selections_to_tool_calls must preserve tool_call_id
|
||||
/// so that tool_result messages match the assistant_with_tool_calls message
|
||||
/// and are not treated as orphaned by sanitize_tool_messages.
|
||||
@@ -2285,4 +2438,95 @@ mod tests {
|
||||
assert_eq!(telegram[0].0, "owner-scope");
|
||||
assert_eq!(telegram[0].1.content, "hello from routine");
|
||||
}
|
||||
|
||||
/// Regression test: only `EmptyResponse` errors are eligible for
|
||||
/// completion-swallowing. Infrastructure errors must always propagate.
|
||||
#[test]
|
||||
fn is_completion_eligible_only_matches_empty_response() {
|
||||
use crate::error::LlmError;
|
||||
|
||||
// EmptyResponse is eligible
|
||||
assert!(super::is_completion_eligible_error(
|
||||
&LlmError::EmptyResponse {
|
||||
provider: "test".to_string(),
|
||||
}
|
||||
));
|
||||
|
||||
// All other variants are NOT eligible
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::InvalidResponse {
|
||||
provider: "test".to_string(),
|
||||
reason: "parse error".to_string(),
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::AuthFailed {
|
||||
provider: "test".to_string(),
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::ContextLengthExceeded {
|
||||
used: 100_000,
|
||||
limit: 50_000,
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::ModelNotAvailable {
|
||||
provider: "test".to_string(),
|
||||
model: "gpt-4".to_string(),
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::RequestFailed {
|
||||
provider: "test".to_string(),
|
||||
reason: "timeout".to_string(),
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::SessionExpired {
|
||||
provider: "test".to_string(),
|
||||
}
|
||||
));
|
||||
assert!(!super::is_completion_eligible_error(
|
||||
&LlmError::SessionRenewalFailed {
|
||||
provider: "test".to_string(),
|
||||
reason: "timeout".to_string(),
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
/// Regression test: AutonomousUnavailable errors must be recoverable.
|
||||
/// Previously the job worker treated them as fatal, killing the entire
|
||||
/// job instead of feeding the error back to the LLM.
|
||||
#[tokio::test]
|
||||
async fn test_autonomous_unavailable_is_recoverable() {
|
||||
let worker = make_worker(vec![]).await;
|
||||
let mut reason_ctx = ReasoningContext::new();
|
||||
let selection = ToolSelection {
|
||||
tool_name: "secret_list".to_string(),
|
||||
parameters: serde_json::json!({}),
|
||||
reasoning: "list secrets".to_string(),
|
||||
alternatives: vec![],
|
||||
tool_call_id: "call_123".to_string(),
|
||||
};
|
||||
let err = Error::Tool(crate::error::ToolError::AutonomousUnavailable {
|
||||
name: "secret_list".to_string(),
|
||||
reason: "not available in autonomous jobs".to_string(),
|
||||
});
|
||||
|
||||
let result = worker
|
||||
.process_tool_result_job(&mut reason_ctx, &selection, Err(err))
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"AutonomousUnavailable must be recoverable, not fatal: {:?}",
|
||||
result
|
||||
);
|
||||
// The error should be fed back to the LLM as a message.
|
||||
assert!(
|
||||
!reason_ctx.messages.is_empty(),
|
||||
"Error message should be added to reason_ctx for the LLM"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,9 +14,12 @@ scenarios/test_extensions.py
|
||||
scenarios/test_html_injection.py
|
||||
scenarios/test_mcp_auth_flow.py
|
||||
scenarios/test_oauth_credential_fallback.py
|
||||
scenarios/test_oauth_refresh.py
|
||||
scenarios/test_oauth_url_parameters.py
|
||||
scenarios/test_owner_scope.py
|
||||
scenarios/test_pairing.py
|
||||
scenarios/test_routine_event_batch.py
|
||||
scenarios/test_routine_full_job.py
|
||||
scenarios/test_routine_oauth_credential_injection.py
|
||||
scenarios/test_skills.py
|
||||
scenarios/test_sse_reconnect.py
|
||||
|
||||
@@ -121,6 +121,75 @@ def _last_user_content(messages: list[dict]) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def _is_job_mode(messages: list[dict]) -> bool:
|
||||
"""Detect if this conversation is a background job (not chat)."""
|
||||
for msg in messages:
|
||||
if msg.get("role") == "system":
|
||||
content = msg.get("content", "")
|
||||
if "autonomous agent working on a job" in content:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _count_tool_results(messages: list[dict]) -> int:
|
||||
"""Count how many tool result messages are in the conversation."""
|
||||
return sum(1 for m in messages if m.get("role") == "tool")
|
||||
|
||||
|
||||
def match_job_response(messages: list[dict], has_tools: bool) -> dict | None:
|
||||
"""Handle background job conversations.
|
||||
|
||||
Returns a dict with either {"text": ...} or {"tool_call": ...},
|
||||
or None if this isn't a job conversation.
|
||||
"""
|
||||
if not _is_job_mode(messages):
|
||||
return None
|
||||
|
||||
last_user = _last_user_content(messages)
|
||||
tool_result_count = _count_tool_results(messages)
|
||||
|
||||
# Planning call (no tools available = complete() not complete_with_tools())
|
||||
if "create a plan" in last_user.lower():
|
||||
return {"text": json.dumps({
|
||||
"goal": "Complete the requested routine job",
|
||||
"actions": [
|
||||
{
|
||||
"tool_name": "echo",
|
||||
"parameters": {"message": "job-step-1"},
|
||||
"reasoning": "First step: echo a test message",
|
||||
"expected_outcome": "Echo returns the message",
|
||||
},
|
||||
{
|
||||
"tool_name": "time",
|
||||
"parameters": {"operation": "now"},
|
||||
"reasoning": "Second step: get the current time",
|
||||
"expected_outcome": "Returns current timestamp",
|
||||
},
|
||||
],
|
||||
"estimated_cost": 0.001,
|
||||
"estimated_time_secs": 5,
|
||||
"confidence": 0.95,
|
||||
})}
|
||||
|
||||
# Post-plan completion check: after tool results, say complete
|
||||
if "planned actions" in last_user.lower() and tool_result_count >= 2:
|
||||
return {"text": "The job is complete. All tasks are done."}
|
||||
|
||||
# Continuation prompt (from our fix): the plan didn't fully complete,
|
||||
# now the agentic loop should call tools
|
||||
if "continue executing now" in last_user.lower() and has_tools:
|
||||
return {"tool_call": {
|
||||
"tool_name": "echo",
|
||||
"arguments": {"message": "continuation-step"},
|
||||
}}
|
||||
|
||||
# After a tool result in the agentic loop, signal completion
|
||||
if tool_result_count > 0 and has_tools:
|
||||
return {"text": "The job is complete. All requested work has been finished."}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def match_response(messages: list[dict]) -> str:
|
||||
content = _last_user_content(messages)
|
||||
for pattern, response in CANNED_RESPONSES:
|
||||
@@ -193,6 +262,19 @@ async def chat_completions(request: web.Request) -> web.StreamResponse:
|
||||
has_tools = bool(body.get("tools"))
|
||||
cid = f"mock-{uuid.uuid4().hex[:8]}"
|
||||
|
||||
# Job-mode conversations (background routine/job execution)
|
||||
job_resp = match_job_response(messages, has_tools)
|
||||
if job_resp:
|
||||
if "tool_call" in job_resp:
|
||||
tc = job_resp["tool_call"]
|
||||
if not stream:
|
||||
return _tool_call_response(cid, tc)
|
||||
return await _stream_tool_call(request, cid, tc)
|
||||
text = job_resp["text"]
|
||||
if not stream:
|
||||
return _text_response(cid, text)
|
||||
return await _stream_text(request, cid, text)
|
||||
|
||||
# Tool result in messages -> text summary
|
||||
tr = _find_tool_result(messages)
|
||||
if tr:
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
"""E2E tests for full_job routine execution.
|
||||
|
||||
Exercises the complete lifecycle: create a full_job routine via the
|
||||
web UI, trigger it via the API, and verify the job runs tools and
|
||||
completes without hitting the iteration cap.
|
||||
|
||||
Requires Playwright (browser-based tests).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import uuid
|
||||
|
||||
from helpers import SEL, api_get, api_post
|
||||
|
||||
|
||||
# -- Helpers ------------------------------------------------------------------
|
||||
|
||||
async def _send_chat_message(page, message: str) -> None:
|
||||
"""Send a chat message and wait for the assistant turn to appear."""
|
||||
chat_input = page.locator(SEL["chat_input"])
|
||||
await chat_input.wait_for(state="visible", timeout=5000)
|
||||
assistant_messages = page.locator(SEL["message_assistant"])
|
||||
before_count = await assistant_messages.count()
|
||||
|
||||
await chat_input.fill(message)
|
||||
await chat_input.press("Enter")
|
||||
|
||||
await page.wait_for_function(
|
||||
"""({ selector, expectedCount }) => {
|
||||
return document.querySelectorAll(selector).length >= expectedCount;
|
||||
}""",
|
||||
arg={
|
||||
"selector": SEL["message_assistant"],
|
||||
"expectedCount": before_count + 1,
|
||||
},
|
||||
timeout=30000,
|
||||
)
|
||||
|
||||
|
||||
async def _wait_for_routine(base_url: str, name: str, timeout: float = 20.0) -> dict:
|
||||
"""Poll until the named routine exists."""
|
||||
for _ in range(int(timeout * 2)):
|
||||
resp = await api_get(base_url, "/api/routines")
|
||||
resp.raise_for_status()
|
||||
for routine in resp.json()["routines"]:
|
||||
if routine["name"] == name:
|
||||
return routine
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(f"Routine '{name}' not created within {timeout}s")
|
||||
|
||||
|
||||
async def _get_routine_runs(base_url: str, routine_id: str) -> list[dict]:
|
||||
"""Fetch routine runs."""
|
||||
resp = await api_get(base_url, f"/api/routines/{routine_id}/runs")
|
||||
resp.raise_for_status()
|
||||
return resp.json()["runs"]
|
||||
|
||||
|
||||
async def _wait_for_completed_run(
|
||||
base_url: str,
|
||||
routine_id: str,
|
||||
*,
|
||||
timeout: float = 60.0,
|
||||
) -> dict:
|
||||
"""Poll until the newest run reaches a terminal state."""
|
||||
for _ in range(int(timeout * 2)):
|
||||
runs = await _get_routine_runs(base_url, routine_id)
|
||||
if runs and runs[0]["status"].lower() not in ("running", "pending"):
|
||||
return runs[0]
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(
|
||||
f"Routine '{routine_id}' did not complete within {timeout}s"
|
||||
)
|
||||
|
||||
|
||||
async def _wait_for_job_terminal(
|
||||
base_url: str,
|
||||
job_id: str,
|
||||
*,
|
||||
timeout: float = 60.0,
|
||||
) -> dict:
|
||||
"""Poll until a job reaches a terminal state."""
|
||||
terminal = {"completed", "failed", "cancelled", "submitted", "accepted"}
|
||||
for _ in range(int(timeout * 2)):
|
||||
resp = await api_get(base_url, f"/api/jobs/{job_id}")
|
||||
resp.raise_for_status()
|
||||
detail = resp.json()
|
||||
if detail.get("state", "").lower() in terminal:
|
||||
return detail
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(f"Job '{job_id}' did not reach terminal state within {timeout}s")
|
||||
|
||||
|
||||
# -- Tests --------------------------------------------------------------------
|
||||
|
||||
async def test_full_job_routine_completes_with_tools(page, ironclaw_server):
|
||||
"""A full_job routine should plan, execute tools, and complete."""
|
||||
name = f"fjob-{uuid.uuid4().hex[:8]}"
|
||||
|
||||
# Step 1: Create full_job routine via chat
|
||||
await _send_chat_message(page, f"create full-job owner routine {name}")
|
||||
routine = await _wait_for_routine(ironclaw_server, name)
|
||||
|
||||
assert routine["id"]
|
||||
assert routine["action_type"] == "full_job"
|
||||
|
||||
# Step 2: Trigger the routine
|
||||
resp = await api_post(ironclaw_server, f"/api/routines/{routine['id']}/trigger")
|
||||
resp.raise_for_status()
|
||||
trigger_data = resp.json()
|
||||
assert trigger_data["status"] == "triggered"
|
||||
|
||||
# Step 3: Wait for the run to complete
|
||||
completed_run = await _wait_for_completed_run(
|
||||
ironclaw_server, routine["id"], timeout=60
|
||||
)
|
||||
|
||||
# The run should have succeeded (not failed)
|
||||
assert completed_run["status"].lower() != "failed", (
|
||||
f"Full job routine run failed: {completed_run}"
|
||||
)
|
||||
|
||||
# Step 4: Verify the job reached a success state.
|
||||
# Jobs may advance past "completed" to "submitted" or "accepted",
|
||||
# so treat all post-completion states as success.
|
||||
success_states = {"completed", "submitted", "accepted"}
|
||||
if completed_run.get("job_id"):
|
||||
job = await _wait_for_job_terminal(
|
||||
ironclaw_server, completed_run["job_id"], timeout=30
|
||||
)
|
||||
assert job["state"].lower() in success_states, (
|
||||
f"Expected job state in {success_states}, got '{job['state']}'"
|
||||
)
|
||||
@@ -21,9 +21,7 @@ mod tests {
|
||||
NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger,
|
||||
};
|
||||
use ironclaw::agent::routine_engine::RoutineEngine;
|
||||
use ironclaw::agent::{
|
||||
HeartbeatConfig, HeartbeatRunner, SandboxReadiness, Scheduler, SchedulerDeps,
|
||||
};
|
||||
use ironclaw::agent::{HeartbeatConfig, HeartbeatRunner, Scheduler, SchedulerDeps};
|
||||
use ironclaw::channels::IncomingMessage;
|
||||
use ironclaw::config::{AgentConfig, RoutineConfig, SafetyConfig};
|
||||
use ironclaw::context::{ContextManager, JobContext};
|
||||
@@ -352,7 +350,7 @@ mod tests {
|
||||
extension_manager,
|
||||
registry,
|
||||
safety,
|
||||
SandboxReadiness::Available,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -456,7 +454,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
// Insert a cron routine with next_fire_at in the past.
|
||||
@@ -535,7 +533,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
// Insert an event routine matching "deploy.*production".
|
||||
@@ -622,7 +620,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
let routine = make_routine(
|
||||
@@ -731,7 +729,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
let mut filters = std::collections::HashMap::new();
|
||||
@@ -874,7 +872,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
// Insert an event routine with 1-hour cooldown.
|
||||
@@ -1057,7 +1055,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
(engine, db, dir)
|
||||
@@ -1179,7 +1177,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
// Create a full_job routine with max_concurrent = 1
|
||||
@@ -1287,7 +1285,7 @@ mod tests {
|
||||
None,
|
||||
tools,
|
||||
safety,
|
||||
SandboxReadiness::DisabledByConfig,
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
|
||||
// Insert a due cron routine
|
||||
|
||||
@@ -198,7 +198,7 @@ mod tests {
|
||||
http_interceptor: None,
|
||||
transcription: None,
|
||||
document_extraction: None,
|
||||
sandbox_readiness: ironclaw::agent::SandboxReadiness::DisabledByConfig,
|
||||
sandbox_readiness: ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
builder: None,
|
||||
llm_backend: "nearai".to_string(),
|
||||
tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)),
|
||||
|
||||
@@ -48,13 +48,7 @@ mod tests {
|
||||
let store = rig.database();
|
||||
assert!(
|
||||
store
|
||||
.ensure_conversation(
|
||||
foreign_thread_id,
|
||||
"gateway",
|
||||
"victim-user",
|
||||
None,
|
||||
Some("gateway")
|
||||
)
|
||||
.ensure_conversation(foreign_thread_id, "gateway", "victim-user", None)
|
||||
.await
|
||||
.expect("failed to create victim conversation"),
|
||||
"test setup failed: victim conversation was not created"
|
||||
|
||||
@@ -264,7 +264,8 @@ impl GatewayWorkflowHarness {
|
||||
http_interceptor: None,
|
||||
transcription: None,
|
||||
document_extraction: None,
|
||||
sandbox_readiness: ironclaw::agent::SandboxReadiness::DisabledByConfig,
|
||||
sandbox_readiness:
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
builder: None,
|
||||
llm_backend: "nearai".to_string(),
|
||||
tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)),
|
||||
|
||||
@@ -650,7 +650,7 @@ impl TestRigBuilder {
|
||||
None,
|
||||
components.tools.clone(),
|
||||
components.safety.clone(),
|
||||
ironclaw::agent::SandboxReadiness::Available, // tests don't use real Docker
|
||||
ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
));
|
||||
components
|
||||
.tools
|
||||
@@ -759,7 +759,7 @@ impl TestRigBuilder {
|
||||
http_interceptor,
|
||||
transcription: None,
|
||||
document_extraction: None,
|
||||
sandbox_readiness: ironclaw::agent::SandboxReadiness::Available, // tests don't use real Docker
|
||||
sandbox_readiness: ironclaw::agent::routine_engine::SandboxReadiness::DisabledByConfig,
|
||||
builder: None,
|
||||
llm_backend: "nearai".to_string(),
|
||||
tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)),
|
||||
|
||||
Reference in New Issue
Block a user