diff --git a/Cargo.toml b/Cargo.toml index 0382a2a7..fbd3d6ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -252,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] diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index 9e639171..4420a450 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -1269,6 +1269,14 @@ pub(crate) fn extract_suggestions(text: &str) -> (String, Vec) { (cleaned, suggestions) } +/// Remove `` 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; @@ -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[\"Check logs\"]"; + 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"; diff --git a/src/agent/mod.rs b/src/agent/mod.rs index e7242845..79616aae 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -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, }; diff --git a/src/agent/routine.rs b/src/agent/routine.rs index 26e769da..5a57a8a6 100644 --- a/src/agent/routine.rs +++ b/src/agent/routine.rs @@ -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 { diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 64c3b94c..3687ebd4 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -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 diff --git a/src/channels/repl.rs b/src/channels/repl.rs index 41d73a8c..27b6ea40 100644 --- a/src/channels/repl.rs +++ b/src/channels/repl.rs @@ -492,10 +492,12 @@ impl Channel for ReplChannel { async fn start(&self) -> Result { 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" ); } } diff --git a/src/channels/web/handlers/jobs.rs b/src/channels/web/handlers/jobs.rs index b171561c..aca3e97c 100644 --- a/src/channels/web/handlers/jobs.rs +++ b/src/channels/web/handlers/jobs.rs @@ -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 = 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 = events .into_iter() diff --git a/src/channels/web/handlers/routines.rs b/src/channels/web/handlers/routines.rs index fc56b187..5597a47c 100644 --- a/src/channels/web/handlers/routines.rs +++ b/src/channels/web/handlers/routines.rs @@ -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, })) } diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 2cadece6..77968223 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -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( diff --git a/src/channels/web/static/app.js b/src/channels/web/static/app.js index 9cfd35df..c0c15acf 100644 --- a/src/channels/web/static/app.js +++ b/src/channels/web/static/app.js @@ -4265,6 +4265,13 @@ function renderRoutineDetail(routine) { html += '

Action

' + '
' + escapeHtml(JSON.stringify(routine.action, null, 2)) + '
'; + // Conversation thread link + if (routine.conversation_id) { + html += '
' + + '' + + 'View Execution Thread
'; + } + // Recent runs if (routine.recent_runs && routine.recent_runs.length > 0) { html += '

Recent Runs

' @@ -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; diff --git a/src/channels/web/tests/mod.rs b/src/channels/web/tests/mod.rs index fa6db197..daeee1bf 100644 --- a/src/channels/web/tests/mod.rs +++ b/src/channels/web/tests/mod.rs @@ -1,3 +1,4 @@ //! Integration tests for the web gateway module. mod multi_tenant; +mod no_silent_drop; diff --git a/src/channels/web/tests/no_silent_drop.rs b/src/channels/web/tests/no_silent_drop.rs new file mode 100644 index 00000000..5ffd9f04 --- /dev/null +++ b/src/channels/web/tests/no_silent_drop.rs @@ -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 + ); +} diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index 8698c030..9ecece57 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -768,6 +768,7 @@ pub struct RoutineDetailResponse { pub run_count: u64, pub consecutive_failures: u32, pub created_at: String, + pub conversation_id: Option, pub recent_runs: Vec, } diff --git a/src/db/libsql/conversations.rs b/src/db/libsql/conversations.rs index 911ee863..4f9f1079 100644 --- a/src/db/libsql/conversations.rs +++ b/src/db/libsql/conversations.rs @@ -290,6 +290,41 @@ impl ConversationStore for LibSqlBackend { result } + async fn find_routine_conversation( + &self, + routine_id: Uuid, + user_id: &str, + ) -> Result, 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( diff --git a/src/db/mod.rs b/src/db/mod.rs index 14cad543..e2a81412 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -391,6 +391,13 @@ pub trait ConversationStore: Send + Sync { routine_name: &str, user_id: &str, ) -> Result; + /// 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, DatabaseError>; async fn get_or_create_heartbeat_conversation( &self, user_id: &str, diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 2fba0b53..462c3c46 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -137,6 +137,16 @@ impl ConversationStore for PgBackend { .await } + async fn find_routine_conversation( + &self, + routine_id: Uuid, + user_id: &str, + ) -> Result, DatabaseError> { + self.store + .find_routine_conversation(routine_id, user_id) + .await + } + async fn get_or_create_heartbeat_conversation( &self, user_id: &str, diff --git a/src/history/store.rs b/src/history/store.rs index e6e869b6..c2b75a88 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1771,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, 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'`. diff --git a/src/llm/openai_codex_provider.rs b/src/llm/openai_codex_provider.rs index 3449a08a..f1f688ac 100644 --- a/src/llm/openai_codex_provider.rs +++ b/src/llm/openai_codex_provider.rs @@ -276,8 +276,33 @@ impl LlmProvider for OpenAiCodexProvider { &self, request: ToolCompletionRequest, ) -> Result { + // 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 = 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::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 { } } +/// 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 = 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"); + } } diff --git a/src/skills/catalog.rs b/src/skills/catalog.rs index 93584f5f..30759546 100644 --- a/src/skills/catalog.rs +++ b/src/skills/catalog.rs @@ -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()); diff --git a/src/tools/builtin/message.rs b/src/tools/builtin/message.rs index 08029d6f..d42de4ce 100644 --- a/src/tools/builtin/message.rs +++ b/src/tools/builtin/message.rs @@ -224,7 +224,13 @@ impl Tool for MessageTool { ) -> Result { 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())); diff --git a/src/tools/builtin/routine.rs b/src/tools/builtin/routine.rs index 76f6e38b..20257f16 100644 --- a/src/tools/builtin/routine.rs +++ b/src/tools/builtin/routine.rs @@ -65,6 +65,7 @@ struct NormalizedExecutionRequest { context_paths: Vec, 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 = 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 )); } } diff --git a/src/tools/builtin/time.rs b/src/tools/builtin/time.rs index 5f037964..f5c944a6 100644 --- a/src/tools/builtin/time.rs +++ b/src/tools/builtin/time.rs @@ -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 { 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)?, diff --git a/src/tools/mcp/auth.rs b/src/tools/mcp/auth.rs index 1926e78d..9e8792ad 100644 --- a/src/tools/mcp/auth.rs +++ b/src/tools/mcp/auth.rs @@ -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 { diff --git a/src/tools/mcp/client.rs b/src/tools/mcp/client.rs index 32c5767d..4125e1f4 100644 --- a/src/tools/mcp/client.rs +++ b/src/tools/mcp/client.rs @@ -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, 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 diff --git a/src/tools/mcp/config.rs b/src/tools/mcp/config.rs index 06adbd3d..c7eb62f6 100644 --- a/src/tools/mcp/config.rs +++ b/src/tools/mcp/config.rs @@ -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" ); } diff --git a/src/util.rs b/src/util.rs index a76f3b27..568e943d 100644 --- a/src/util.rs +++ b/src/util.rs @@ -225,4 +225,26 @@ mod tests { "The tool returned: TASK_COMPLETE signal" )); } + + #[test] + fn signals_completion_after_suggestions_stripped() { + // Regression: after stripping 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" + )); + } } diff --git a/src/worker/job.rs b/src/worker/job.rs index 94a04290..edf87bf8 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -828,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(()) } } } @@ -930,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 @@ -1420,16 +1431,20 @@ impl<'a> LoopDelegate for JobDelegate<'a> { 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 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. @@ -1437,7 +1452,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> { .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", @@ -1447,7 +1462,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> { }), ); - TextAction::Continue + TextAction::Return(LoopOutcome::Response(text)) } async fn execute_tool_calls( @@ -1456,6 +1471,9 @@ impl<'a> LoopDelegate for JobDelegate<'a> { content: Option, reason_ctx: &mut ReasoningContext, ) -> Result, 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", @@ -2156,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. @@ -2429,4 +2494,39 @@ mod tests { } )); } + + /// 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" + ); + } } diff --git a/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt b/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt index c2784f64..aa4ea46c 100644 --- a/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt +++ b/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt @@ -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 diff --git a/tests/e2e/mock_llm.py b/tests/e2e/mock_llm.py index 1147662c..f049e504 100644 --- a/tests/e2e/mock_llm.py +++ b/tests/e2e/mock_llm.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: diff --git a/tests/e2e/scenarios/test_routine_full_job.py b/tests/e2e/scenarios/test_routine_full_job.py new file mode 100644 index 00000000..64472dc9 --- /dev/null +++ b/tests/e2e/scenarios/test_routine_full_job.py @@ -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']}'" + ) diff --git a/tests/e2e_routine_heartbeat.rs b/tests/e2e_routine_heartbeat.rs index 6849ee05..36d87a07 100644 --- a/tests/e2e_routine_heartbeat.rs +++ b/tests/e2e_routine_heartbeat.rs @@ -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 diff --git a/tests/e2e_telegram_message_routing.rs b/tests/e2e_telegram_message_routing.rs index 810fc218..e91aae8e 100644 --- a/tests/e2e_telegram_message_routing.rs +++ b/tests/e2e_telegram_message_routing.rs @@ -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)), diff --git a/tests/support/gateway_workflow_harness.rs b/tests/support/gateway_workflow_harness.rs index d6341704..efb5cf96 100644 --- a/tests/support/gateway_workflow_harness.rs +++ b/tests/support/gateway_workflow_harness.rs @@ -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)), diff --git a/tests/support/test_rig.rs b/tests/support/test_rig.rs index 5775b86d..0e288390 100644 --- a/tests/support/test_rig.rs +++ b/tests/support/test_rig.rs @@ -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)),