From ca8d5c6b5e0249e4536782d290c099f785cbe71c Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Sat, 14 Feb 2026 21:39:52 -0800 Subject: [PATCH] refactor: deduplicate tool code and remove dead stubs (#98) * refactor: deduplicate tool parameter extraction and remove dead stub tools Delete 4 never-registered stub tools (marketplace, restaurant, ecommerce, taskrabbit) removing ~625 lines of dead code. Add require_str/require_param helpers to tool.rs and refactor ~30 call sites across 10 tool files from 4-6 line inline extractions to single-line calls. Consolidate worker HTTP client with get_json/post_json helpers, reducing boilerplate in 4 methods. Co-Authored-By: Claude Opus 4.6 * fix: return JSON from orchestrator /complete endpoint The report_complete handler returned bare StatusCode::OK (no body), which broke the post_json helper that expects a JSON response. Return {"status": "ok"} for consistency with other worker endpoints. Addresses review feedback on PR #98. Co-Authored-By: Claude Opus 4.6 --------- Co-authored-by: Claude Opus 4.6 --- src/estimation/cost.rs | 6 +- src/estimation/time.rs | 4 - src/orchestrator/api.rs | 4 +- src/tools/builtin/echo.rs | 9 +- src/tools/builtin/ecommerce.rs | 136 --------------------- src/tools/builtin/extension_tools.rs | 22 +--- src/tools/builtin/file.rs | 32 ++--- src/tools/builtin/http.rs | 14 +-- src/tools/builtin/job.rs | 24 +--- src/tools/builtin/json.rs | 13 +- src/tools/builtin/marketplace.rs | 160 ------------------------- src/tools/builtin/memory.rs | 19 +-- src/tools/builtin/mod.rs | 8 -- src/tools/builtin/restaurant.rs | 172 --------------------------- src/tools/builtin/routine.rs | 32 ++--- src/tools/builtin/shell.rs | 7 +- src/tools/builtin/taskrabbit.rs | 157 ------------------------ src/tools/builtin/time.rs | 30 +---- src/tools/tool.rs | 65 +++++++++- src/worker/api.rs | 124 +++++++++---------- 20 files changed, 159 insertions(+), 879 deletions(-) delete mode 100644 src/tools/builtin/ecommerce.rs delete mode 100644 src/tools/builtin/marketplace.rs delete mode 100644 src/tools/builtin/restaurant.rs delete mode 100644 src/tools/builtin/taskrabbit.rs diff --git a/src/estimation/cost.rs b/src/estimation/cost.rs index 6ebe2b17..4e8fc62b 100644 --- a/src/estimation/cost.rs +++ b/src/estimation/cost.rs @@ -20,10 +20,6 @@ impl CostEstimator { // Default tool costs (in USD or equivalent) tool_costs.insert("http".to_string(), dec!(0.0001)); // API call - tool_costs.insert("marketplace".to_string(), dec!(0.01)); // Gas costs - tool_costs.insert("ecommerce".to_string(), dec!(0.001)); // API call - tool_costs.insert("taskrabbit".to_string(), dec!(0.0)); // Cost comes from task itself - tool_costs.insert("restaurant".to_string(), dec!(0.001)); // API call tool_costs.insert("echo".to_string(), dec!(0.0)); // Free tool_costs.insert("time".to_string(), dec!(0.0)); // Free tool_costs.insert("json".to_string(), dec!(0.0)); // Free @@ -74,7 +70,7 @@ mod tests { let estimator = CostEstimator::new(); assert_eq!(estimator.estimate_tool("echo"), dec!(0.0)); - assert_eq!(estimator.estimate_tool("marketplace"), dec!(0.01)); + assert_eq!(estimator.estimate_tool("http"), dec!(0.0001)); assert!(estimator.estimate_tool("unknown") > dec!(0.0)); } diff --git a/src/estimation/time.rs b/src/estimation/time.rs index 4f2c4e72..76012bdb 100644 --- a/src/estimation/time.rs +++ b/src/estimation/time.rs @@ -16,10 +16,6 @@ impl TimeEstimator { // Default tool durations tool_durations.insert("http".to_string(), Duration::from_secs(5)); - tool_durations.insert("marketplace".to_string(), Duration::from_secs(10)); - tool_durations.insert("ecommerce".to_string(), Duration::from_secs(8)); - tool_durations.insert("taskrabbit".to_string(), Duration::from_secs(30)); // Just API, not task itself - tool_durations.insert("restaurant".to_string(), Duration::from_secs(5)); tool_durations.insert("echo".to_string(), Duration::from_millis(10)); tool_durations.insert("time".to_string(), Duration::from_millis(1)); tool_durations.insert("json".to_string(), Duration::from_millis(5)); diff --git a/src/orchestrator/api.rs b/src/orchestrator/api.rs index 012ec270..afa5ce2a 100644 --- a/src/orchestrator/api.rs +++ b/src/orchestrator/api.rs @@ -202,7 +202,7 @@ async fn report_complete( State(state): State, Path(job_id): Path, Json(report): Json, -) -> Result { +) -> Result, StatusCode> { if report.success { tracing::info!( job_id = %job_id, @@ -223,7 +223,7 @@ async fn report_complete( }; let _ = state.job_manager.complete_job(job_id, result).await; - Ok(StatusCode::OK) + Ok(Json(serde_json::json!({"status": "ok"}))) } // -- Sandbox job event handlers -- diff --git a/src/tools/builtin/echo.rs b/src/tools/builtin/echo.rs index 6119eeb3..faf31982 100644 --- a/src/tools/builtin/echo.rs +++ b/src/tools/builtin/echo.rs @@ -3,7 +3,7 @@ use async_trait::async_trait; use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; /// Simple echo tool for testing. pub struct EchoTool; @@ -38,12 +38,7 @@ impl Tool for EchoTool { ) -> Result { let start = std::time::Instant::now(); - let message = params - .get("message") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'message' parameter".to_string()) - })?; + let message = require_str(¶ms, "message")?; Ok(ToolOutput::text(message, start.elapsed())) } diff --git a/src/tools/builtin/ecommerce.rs b/src/tools/builtin/ecommerce.rs deleted file mode 100644 index 86ba1acb..00000000 --- a/src/tools/builtin/ecommerce.rs +++ /dev/null @@ -1,136 +0,0 @@ -//! E-commerce tool for shopping and price comparison. - -use async_trait::async_trait; - -use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; - -/// Tool for e-commerce operations (Amazon, price comparison, etc.). -pub struct EcommerceTool { - // TODO: Add API clients -} - -impl EcommerceTool { - /// Create a new e-commerce tool. - pub fn new() -> Self { - Self {} - } -} - -impl Default for EcommerceTool { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl Tool for EcommerceTool { - fn name(&self) -> &str { - "ecommerce" - } - - fn description(&self) -> &str { - "Search products, compare prices, and find deals across e-commerce platforms." - } - - fn parameters_schema(&self) -> serde_json::Value { - serde_json::json!({ - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["search", "get_product", "compare_prices", "track_price"], - "description": "The e-commerce action to perform" - }, - "query": { - "type": "string", - "description": "Search query (for search action)" - }, - "product_id": { - "type": "string", - "description": "Product ID or ASIN (for get_product, compare_prices)" - }, - "platform": { - "type": "string", - "enum": ["amazon", "ebay", "walmart", "all"], - "description": "E-commerce platform to search" - }, - "max_price": { - "type": "number", - "description": "Maximum price filter" - }, - "category": { - "type": "string", - "description": "Product category filter" - } - }, - "required": ["action"] - }) - } - - async fn execute( - &self, - params: serde_json::Value, - _ctx: &JobContext, - ) -> Result { - let start = std::time::Instant::now(); - - let action = params - .get("action") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'action' parameter".to_string()) - })?; - - // TODO: Implement actual e-commerce API integrations - let result = match action { - "search" => { - let query = params.get("query").and_then(|v| v.as_str()).unwrap_or(""); - - serde_json::json!({ - "query": query, - "results": [], - "message": "E-commerce integration not yet implemented" - }) - } - "get_product" => { - let product_id = params - .get("product_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'product_id' parameter".to_string()) - })?; - - serde_json::json!({ - "product_id": product_id, - "found": false, - "message": "E-commerce integration not yet implemented" - }) - } - "compare_prices" => { - serde_json::json!({ - "prices": [], - "message": "E-commerce integration not yet implemented" - }) - } - "track_price" => { - serde_json::json!({ - "tracking": false, - "message": "E-commerce integration not yet implemented" - }) - } - _ => { - return Err(ToolError::InvalidParameters(format!( - "unknown action: {}", - action - ))); - } - }; - - Ok(ToolOutput::success(result, start.elapsed())) - } - - fn requires_sanitization(&self) -> bool { - true // External e-commerce data - } -} diff --git a/src/tools/builtin/extension_tools.rs b/src/tools/builtin/extension_tools.rs index c707fd69..a7909195 100644 --- a/src/tools/builtin/extension_tools.rs +++ b/src/tools/builtin/extension_tools.rs @@ -9,7 +9,7 @@ use async_trait::async_trait; use crate::context::JobContext; use crate::extensions::{ExtensionKind, ExtensionManager}; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; // ── tool_search ────────────────────────────────────────────────────────── @@ -133,10 +133,7 @@ impl Tool for ToolInstallTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("name is required".to_string()))?; + let name = require_str(¶ms, "name")?; let url = params.get("url").and_then(|v| v.as_str()); @@ -210,10 +207,7 @@ impl Tool for ToolAuthTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("name is required".to_string()))?; + let name = require_str(¶ms, "name")?; let result = self .manager @@ -306,10 +300,7 @@ impl Tool for ToolActivateTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("name is required".to_string()))?; + let name = require_str(¶ms, "name")?; match self.manager.activate(name).await { Ok(result) => { @@ -471,10 +462,7 @@ impl Tool for ToolRemoveTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("name is required".to_string()))?; + let name = require_str(¶ms, "name")?; let message = self .manager diff --git a/src/tools/builtin/file.rs b/src/tools/builtin/file.rs index 5a1c0aea..9e7ffbcb 100644 --- a/src/tools/builtin/file.rs +++ b/src/tools/builtin/file.rs @@ -11,7 +11,7 @@ use async_trait::async_trait; use tokio::fs; use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolDomain, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolDomain, ToolError, ToolOutput, require_str}; use crate::workspace::paths as ws_paths; /// Well-known workspace filenames that must go through memory_write, not write_file. @@ -203,10 +203,7 @@ impl Tool for ReadFileTool { params: serde_json::Value, _ctx: &JobContext, ) -> Result { - let path_str = params - .get("path") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'path' parameter".into()))?; + let path_str = require_str(¶ms, "path")?; let offset = params.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as usize; let limit = params.get("limit").and_then(|v| v.as_u64()); @@ -328,10 +325,7 @@ impl Tool for WriteFileTool { params: serde_json::Value, _ctx: &JobContext, ) -> Result { - let path_str = params - .get("path") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'path' parameter".into()))?; + let path_str = require_str(¶ms, "path")?; // Reject workspace paths: these live in the database, not on disk. if is_workspace_path(path_str) { @@ -342,10 +336,7 @@ impl Tool for WriteFileTool { ))); } - let content = params - .get("content") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'content' parameter".into()))?; + let content = require_str(¶ms, "content")?; let start = std::time::Instant::now(); @@ -650,20 +641,11 @@ impl Tool for ApplyPatchTool { params: serde_json::Value, _ctx: &JobContext, ) -> Result { - let path_str = params - .get("path") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'path' parameter".into()))?; + let path_str = require_str(¶ms, "path")?; - let old_string = params - .get("old_string") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'old_string' parameter".into()))?; + let old_string = require_str(¶ms, "old_string")?; - let new_string = params - .get("new_string") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'new_string' parameter".into()))?; + let new_string = require_str(¶ms, "new_string")?; let replace_all = params .get("replace_all") diff --git a/src/tools/builtin/http.rs b/src/tools/builtin/http.rs index 3312eda5..79e5caa6 100644 --- a/src/tools/builtin/http.rs +++ b/src/tools/builtin/http.rs @@ -9,7 +9,7 @@ use reqwest::Client; use crate::context::JobContext; use crate::safety::LeakDetector; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; /// Maximum response body size (5 MB). Prevents OOM from unbounded responses. const MAX_RESPONSE_SIZE: usize = 5 * 1024 * 1024; @@ -154,17 +154,9 @@ impl Tool for HttpTool { ) -> Result { let start = std::time::Instant::now(); - let method = params - .get("method") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'method' parameter".to_string()) - })?; + let method = require_str(¶ms, "method")?; - let url = params - .get("url") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'url' parameter".to_string()))?; + let url = require_str(¶ms, "url")?; let parsed_url = validate_url(url)?; // Parse headers diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 015c3b13..b0adb226 100644 --- a/src/tools/builtin/job.rs +++ b/src/tools/builtin/job.rs @@ -18,7 +18,7 @@ use crate::context::{ContextManager, JobContext, JobState}; use crate::db::Database; use crate::history::SandboxJobRecord; use crate::orchestrator::job_manager::{ContainerJobManager, JobMode}; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; /// Tool for creating a new job. /// @@ -467,17 +467,9 @@ impl Tool for CreateJobTool { params: serde_json::Value, ctx: &JobContext, ) -> Result { - let title = params - .get("title") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'title' parameter".into()))?; + let title = require_str(¶ms, "title")?; - let description = params - .get("description") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'description' parameter".into()) - })?; + let description = require_str(¶ms, "description")?; if self.sandbox_enabled() { let wait = params.get("wait").and_then(|v| v.as_bool()).unwrap_or(true); @@ -635,10 +627,7 @@ impl Tool for JobStatusTool { let start = std::time::Instant::now(); let requester_id = ctx.user_id.clone(); - let job_id_str = params - .get("job_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'job_id' parameter".into()))?; + let job_id_str = require_str(¶ms, "job_id")?; let job_id = Uuid::parse_str(job_id_str).map_err(|_| { ToolError::InvalidParameters(format!("invalid job ID format: {}", job_id_str)) @@ -720,10 +709,7 @@ impl Tool for CancelJobTool { let start = std::time::Instant::now(); let requester_id = ctx.user_id.clone(); - let job_id_str = params - .get("job_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'job_id' parameter".into()))?; + let job_id_str = require_str(¶ms, "job_id")?; let job_id = Uuid::parse_str(job_id_str).map_err(|_| { ToolError::InvalidParameters(format!("invalid job ID format: {}", job_id_str)) diff --git a/src/tools/builtin/json.rs b/src/tools/builtin/json.rs index 3ff6e1a9..5350dbb7 100644 --- a/src/tools/builtin/json.rs +++ b/src/tools/builtin/json.rs @@ -3,7 +3,7 @@ use async_trait::async_trait; use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_param, require_str}; /// Tool for JSON manipulation (parse, query, transform). pub struct JsonTool; @@ -46,16 +46,9 @@ impl Tool for JsonTool { ) -> Result { let start = std::time::Instant::now(); - let operation = params - .get("operation") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'operation' parameter".to_string()) - })?; + let operation = require_str(¶ms, "operation")?; - let data = params - .get("data") - .ok_or_else(|| ToolError::InvalidParameters("missing 'data' parameter".to_string()))?; + let data = require_param(¶ms, "data")?; let result = match operation { "parse" => { diff --git a/src/tools/builtin/marketplace.rs b/src/tools/builtin/marketplace.rs deleted file mode 100644 index 73c4c4a6..00000000 --- a/src/tools/builtin/marketplace.rs +++ /dev/null @@ -1,160 +0,0 @@ -//! NEAR AI Marketplace tool. - -use async_trait::async_trait; -use rust_decimal::Decimal; - -use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; - -/// Tool for interacting with the NEAR AI marketplace. -pub struct MarketplaceTool { - // TODO: Add marketplace client -} - -impl MarketplaceTool { - /// Create a new marketplace tool. - pub fn new() -> Self { - Self {} - } -} - -impl Default for MarketplaceTool { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl Tool for MarketplaceTool { - fn name(&self) -> &str { - "marketplace" - } - - fn description(&self) -> &str { - "Interact with the NEAR AI marketplace: search jobs, submit bids, deliver work." - } - - fn parameters_schema(&self) -> serde_json::Value { - serde_json::json!({ - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["search_jobs", "get_job", "submit_bid", "accept_job", "submit_work", "get_status"], - "description": "The marketplace action to perform" - }, - "job_id": { - "type": "string", - "description": "Job ID (for get_job, submit_bid, accept_job, submit_work)" - }, - "query": { - "type": "string", - "description": "Search query (for search_jobs)" - }, - "category": { - "type": "string", - "description": "Job category filter (for search_jobs)" - }, - "bid_amount": { - "type": "number", - "description": "Bid amount in NEAR (for submit_bid)" - }, - "work_url": { - "type": "string", - "description": "URL to submitted work (for submit_work)" - }, - "work_description": { - "type": "string", - "description": "Description of completed work (for submit_work)" - } - }, - "required": ["action"] - }) - } - - async fn execute( - &self, - params: serde_json::Value, - _ctx: &JobContext, - ) -> Result { - let start = std::time::Instant::now(); - - let action = params - .get("action") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'action' parameter".to_string()) - })?; - - // TODO: Implement actual marketplace integration - let result = match action { - "search_jobs" => { - // Placeholder response - serde_json::json!({ - "jobs": [], - "total": 0, - "message": "Marketplace integration not yet implemented" - }) - } - "get_job" => { - let job_id = params - .get("job_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'job_id' parameter".to_string()) - })?; - - serde_json::json!({ - "job_id": job_id, - "status": "not_found", - "message": "Marketplace integration not yet implemented" - }) - } - "submit_bid" => { - serde_json::json!({ - "success": false, - "message": "Marketplace integration not yet implemented" - }) - } - "accept_job" => { - serde_json::json!({ - "success": false, - "message": "Marketplace integration not yet implemented" - }) - } - "submit_work" => { - serde_json::json!({ - "success": false, - "message": "Marketplace integration not yet implemented" - }) - } - "get_status" => { - serde_json::json!({ - "connected": false, - "message": "Marketplace integration not yet implemented" - }) - } - _ => { - return Err(ToolError::InvalidParameters(format!( - "unknown action: {}", - action - ))); - } - }; - - Ok(ToolOutput::success(result, start.elapsed())) - } - - fn estimated_cost(&self, params: &serde_json::Value) -> Option { - // Bidding has a cost - if params.get("action").and_then(|v| v.as_str()) == Some("submit_bid") { - Some(Decimal::new(1, 2)) // 0.01 NEAR gas cost - } else { - None - } - } - - fn requires_sanitization(&self) -> bool { - true // External marketplace data - } -} diff --git a/src/tools/builtin/memory.rs b/src/tools/builtin/memory.rs index 8a2bbbdb..cdcf504d 100644 --- a/src/tools/builtin/memory.rs +++ b/src/tools/builtin/memory.rs @@ -17,7 +17,7 @@ use std::sync::Arc; use async_trait::async_trait; use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; use crate::workspace::{Workspace, paths}; /// Identity files that the LLM must not overwrite via tool calls. @@ -81,10 +81,7 @@ impl Tool for MemorySearchTool { ) -> Result { let start = std::time::Instant::now(); - let query = params - .get("query") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'query' parameter".to_string()))?; + let query = require_str(¶ms, "query")?; let limit = params .get("limit") @@ -176,12 +173,7 @@ impl Tool for MemoryWriteTool { ) -> Result { let start = std::time::Instant::now(); - let content = params - .get("content") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'content' parameter".to_string()) - })?; + let content = require_str(¶ms, "content")?; if content.trim().is_empty() { return Err(ToolError::InvalidParameters( @@ -337,10 +329,7 @@ impl Tool for MemoryReadTool { ) -> Result { let start = std::time::Instant::now(); - let path = params - .get("path") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'path' parameter".to_string()))?; + let path = require_str(¶ms, "path")?; let doc = self .workspace diff --git a/src/tools/builtin/mod.rs b/src/tools/builtin/mod.rs index 8ee99b16..0b98a1ed 100644 --- a/src/tools/builtin/mod.rs +++ b/src/tools/builtin/mod.rs @@ -1,22 +1,17 @@ //! Built-in tools that come with the agent. mod echo; -mod ecommerce; pub mod extension_tools; mod file; mod http; mod job; mod json; -mod marketplace; mod memory; -mod restaurant; pub mod routine; pub(crate) mod shell; -mod taskrabbit; mod time; pub use echo::EchoTool; -pub use ecommerce::EcommerceTool; pub use extension_tools::{ ToolActivateTool, ToolAuthTool, ToolInstallTool, ToolListTool, ToolRemoveTool, ToolSearchTool, }; @@ -24,12 +19,9 @@ pub use file::{ApplyPatchTool, ListDirTool, ReadFileTool, WriteFileTool}; pub use http::HttpTool; pub use job::{CancelJobTool, CreateJobTool, JobStatusTool, ListJobsTool}; pub use json::JsonTool; -pub use marketplace::MarketplaceTool; pub use memory::{MemoryReadTool, MemorySearchTool, MemoryTreeTool, MemoryWriteTool}; -pub use restaurant::RestaurantTool; pub use routine::{ RoutineCreateTool, RoutineDeleteTool, RoutineHistoryTool, RoutineListTool, RoutineUpdateTool, }; pub use shell::ShellTool; -pub use taskrabbit::TaskRabbitTool; pub use time::TimeTool; diff --git a/src/tools/builtin/restaurant.rs b/src/tools/builtin/restaurant.rs deleted file mode 100644 index f29c2aab..00000000 --- a/src/tools/builtin/restaurant.rs +++ /dev/null @@ -1,172 +0,0 @@ -//! Restaurant reservation tool. - -use async_trait::async_trait; - -use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; - -/// Tool for restaurant reservations (OpenTable, Resy, etc.). -pub struct RestaurantTool { - // TODO: Add reservation API clients -} - -impl RestaurantTool { - /// Create a new restaurant tool. - pub fn new() -> Self { - Self {} - } -} - -impl Default for RestaurantTool { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl Tool for RestaurantTool { - fn name(&self) -> &str { - "restaurant" - } - - fn description(&self) -> &str { - "Search restaurants, check availability, and make reservations via OpenTable, Resy, etc." - } - - fn parameters_schema(&self) -> serde_json::Value { - serde_json::json!({ - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["search", "check_availability", "make_reservation", "cancel_reservation", "get_reservation"], - "description": "The restaurant action to perform" - }, - "query": { - "type": "string", - "description": "Search query (cuisine type, restaurant name, etc.)" - }, - "location": { - "type": "object", - "properties": { - "city": { "type": "string" }, - "neighborhood": { "type": "string" }, - "latitude": { "type": "number" }, - "longitude": { "type": "number" } - }, - "description": "Location to search near" - }, - "date": { - "type": "string", - "description": "Reservation date (YYYY-MM-DD)" - }, - "time": { - "type": "string", - "description": "Preferred time (HH:MM)" - }, - "party_size": { - "type": "integer", - "description": "Number of guests" - }, - "restaurant_id": { - "type": "string", - "description": "Restaurant ID (for check_availability, make_reservation)" - }, - "reservation_id": { - "type": "string", - "description": "Reservation ID (for cancel_reservation, get_reservation)" - }, - "guest_name": { - "type": "string", - "description": "Name for the reservation" - }, - "guest_phone": { - "type": "string", - "description": "Phone number for the reservation" - }, - "guest_email": { - "type": "string", - "description": "Email for the reservation" - } - }, - "required": ["action"] - }) - } - - async fn execute( - &self, - params: serde_json::Value, - _ctx: &JobContext, - ) -> Result { - let start = std::time::Instant::now(); - - let action = params - .get("action") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'action' parameter".to_string()) - })?; - - // TODO: Implement actual restaurant reservation API integrations - let result = match action { - "search" => { - let query = params.get("query").and_then(|v| v.as_str()).unwrap_or(""); - - serde_json::json!({ - "query": query, - "restaurants": [], - "message": "Restaurant integration not yet implemented" - }) - } - "check_availability" => { - let restaurant_id = params - .get("restaurant_id") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters( - "missing 'restaurant_id' parameter".to_string(), - ) - })?; - - serde_json::json!({ - "restaurant_id": restaurant_id, - "available_times": [], - "message": "Restaurant integration not yet implemented" - }) - } - "make_reservation" => { - serde_json::json!({ - "success": false, - "message": "Restaurant integration not yet implemented" - }) - } - "cancel_reservation" => { - serde_json::json!({ - "cancelled": false, - "message": "Restaurant integration not yet implemented" - }) - } - "get_reservation" => { - let reservation_id = params.get("reservation_id").and_then(|v| v.as_str()); - - serde_json::json!({ - "reservation_id": reservation_id, - "found": false, - "message": "Restaurant integration not yet implemented" - }) - } - _ => { - return Err(ToolError::InvalidParameters(format!( - "unknown action: {}", - action - ))); - } - }; - - Ok(ToolOutput::success(result, start.elapsed())) - } - - fn requires_sanitization(&self) -> bool { - true // External restaurant data - } -} diff --git a/src/tools/builtin/routine.rs b/src/tools/builtin/routine.rs index 338601bc..6a0abce9 100644 --- a/src/tools/builtin/routine.rs +++ b/src/tools/builtin/routine.rs @@ -20,7 +20,7 @@ use crate::agent::routine::{ use crate::agent::routine_engine::RoutineEngine; use crate::context::JobContext; use crate::db::Database; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; // ==================== routine_create ==================== @@ -106,25 +106,16 @@ impl Tool for RoutineCreateTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'name'".to_string()))?; + let name = require_str(¶ms, "name")?; let description = params .get("description") .and_then(|v| v.as_str()) .unwrap_or(""); - let trigger_type = params - .get("trigger_type") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'trigger_type'".to_string()))?; + let trigger_type = require_str(¶ms, "trigger_type")?; - let prompt = params - .get("prompt") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'prompt'".to_string()))?; + let prompt = require_str(¶ms, "prompt")?; // Build trigger let trigger = match trigger_type { @@ -408,10 +399,7 @@ impl Tool for RoutineUpdateTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'name'".to_string()))?; + let name = require_str(¶ms, "name")?; let mut routine = self .store @@ -514,10 +502,7 @@ impl Tool for RoutineDeleteTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'name'".to_string()))?; + let name = require_str(¶ms, "name")?; let routine = self .store @@ -595,10 +580,7 @@ impl Tool for RoutineHistoryTool { ) -> Result { let start = std::time::Instant::now(); - let name = params - .get("name") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'name'".to_string()))?; + let name = require_str(¶ms, "name")?; let limit = params .get("limit") diff --git a/src/tools/builtin/shell.rs b/src/tools/builtin/shell.rs index 56caba40..0d883da2 100644 --- a/src/tools/builtin/shell.rs +++ b/src/tools/builtin/shell.rs @@ -30,7 +30,7 @@ use tokio::process::Command; use crate::context::JobContext; use crate::sandbox::{SandboxManager, SandboxPolicy}; -use crate::tools::tool::{Tool, ToolDomain, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolDomain, ToolError, ToolOutput, require_str}; /// Maximum output size before truncation (64KB). const MAX_OUTPUT_SIZE: usize = 64 * 1024; @@ -401,10 +401,7 @@ impl Tool for ShellTool { params: serde_json::Value, _ctx: &JobContext, ) -> Result { - let command = params - .get("command") - .and_then(|v| v.as_str()) - .ok_or_else(|| ToolError::InvalidParameters("missing 'command' parameter".into()))?; + let command = require_str(¶ms, "command")?; let workdir = params.get("workdir").and_then(|v| v.as_str()); let timeout = params.get("timeout").and_then(|v| v.as_u64()); diff --git a/src/tools/builtin/taskrabbit.rs b/src/tools/builtin/taskrabbit.rs deleted file mode 100644 index 17be46e1..00000000 --- a/src/tools/builtin/taskrabbit.rs +++ /dev/null @@ -1,157 +0,0 @@ -//! TaskRabbit tool for real-world task delegation. - -use async_trait::async_trait; -use rust_decimal::Decimal; - -use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; - -/// Tool for delegating real-world tasks via TaskRabbit. -pub struct TaskRabbitTool { - // TODO: Add TaskRabbit API client -} - -impl TaskRabbitTool { - /// Create a new TaskRabbit tool. - pub fn new() -> Self { - Self {} - } -} - -impl Default for TaskRabbitTool { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl Tool for TaskRabbitTool { - fn name(&self) -> &str { - "taskrabbit" - } - - fn description(&self) -> &str { - "Delegate real-world tasks to TaskRabbit taskers (delivery, assembly, cleaning, etc.)." - } - - fn parameters_schema(&self) -> serde_json::Value { - serde_json::json!({ - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["search_taskers", "get_quote", "book_task", "get_status", "cancel_task"], - "description": "The TaskRabbit action to perform" - }, - "task_type": { - "type": "string", - "enum": ["delivery", "assembly", "moving", "cleaning", "handyman", "other"], - "description": "Type of task" - }, - "description": { - "type": "string", - "description": "Detailed description of the task" - }, - "location": { - "type": "object", - "properties": { - "address": { "type": "string" }, - "city": { "type": "string" }, - "state": { "type": "string" }, - "zip": { "type": "string" } - }, - "description": "Location for the task" - }, - "scheduled_time": { - "type": "string", - "description": "ISO 8601 datetime for when the task should be performed" - }, - "budget": { - "type": "number", - "description": "Maximum budget for the task in USD" - }, - "task_id": { - "type": "string", - "description": "Task ID (for get_status, cancel_task)" - } - }, - "required": ["action"] - }) - } - - async fn execute( - &self, - params: serde_json::Value, - _ctx: &JobContext, - ) -> Result { - let start = std::time::Instant::now(); - - let action = params - .get("action") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'action' parameter".to_string()) - })?; - - // TODO: Implement actual TaskRabbit API integration - let result = match action { - "search_taskers" => { - serde_json::json!({ - "taskers": [], - "message": "TaskRabbit integration not yet implemented" - }) - } - "get_quote" => { - serde_json::json!({ - "quotes": [], - "message": "TaskRabbit integration not yet implemented" - }) - } - "book_task" => { - serde_json::json!({ - "booked": false, - "message": "TaskRabbit integration not yet implemented" - }) - } - "get_status" => { - let task_id = params.get("task_id").and_then(|v| v.as_str()); - - serde_json::json!({ - "task_id": task_id, - "status": "unknown", - "message": "TaskRabbit integration not yet implemented" - }) - } - "cancel_task" => { - serde_json::json!({ - "cancelled": false, - "message": "TaskRabbit integration not yet implemented" - }) - } - _ => { - return Err(ToolError::InvalidParameters(format!( - "unknown action: {}", - action - ))); - } - }; - - Ok(ToolOutput::success(result, start.elapsed())) - } - - fn estimated_cost(&self, params: &serde_json::Value) -> Option { - // Booking a task has associated costs - if params.get("action").and_then(|v| v.as_str()) == Some("book_task") { - params - .get("budget") - .and_then(|v| v.as_f64()) - .map(|b| Decimal::try_from(b).unwrap_or_default()) - } else { - None - } - } - - fn requires_sanitization(&self) -> bool { - true // External TaskRabbit data - } -} diff --git a/src/tools/builtin/time.rs b/src/tools/builtin/time.rs index 9df6536b..9388f8c7 100644 --- a/src/tools/builtin/time.rs +++ b/src/tools/builtin/time.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; use crate::context::JobContext; -use crate::tools::tool::{Tool, ToolError, ToolOutput}; +use crate::tools::tool::{Tool, ToolError, ToolOutput, require_str}; /// Tool for getting current time and date operations. pub struct TimeTool; @@ -52,12 +52,7 @@ impl Tool for TimeTool { ) -> Result { let start = std::time::Instant::now(); - let operation = params - .get("operation") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'operation' parameter".to_string()) - })?; + let operation = require_str(¶ms, "operation")?; let result = match operation { "now" => { @@ -69,12 +64,7 @@ impl Tool for TimeTool { }) } "parse" => { - let timestamp = params - .get("timestamp") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'timestamp' parameter".to_string()) - })?; + let timestamp = require_str(¶ms, "timestamp")?; let dt: DateTime = timestamp.parse().map_err(|e| { ToolError::InvalidParameters(format!("invalid timestamp: {}", e)) @@ -87,19 +77,9 @@ impl Tool for TimeTool { }) } "diff" => { - let ts1 = params - .get("timestamp") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'timestamp' parameter".to_string()) - })?; + let ts1 = require_str(¶ms, "timestamp")?; - let ts2 = params - .get("timestamp2") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'timestamp2' parameter".to_string()) - })?; + let ts2 = require_str(¶ms, "timestamp2")?; let dt1: DateTime = ts1.parse().map_err(|e| { ToolError::InvalidParameters(format!("invalid timestamp: {}", e)) diff --git a/src/tools/tool.rs b/src/tools/tool.rs index b2f242bb..ae049ead 100644 --- a/src/tools/tool.rs +++ b/src/tools/tool.rs @@ -199,6 +199,28 @@ pub trait Tool: Send + Sync { } } +/// Extract a required string parameter from a JSON object. +/// +/// Returns `ToolError::InvalidParameters` if the key is missing or not a string. +pub fn require_str<'a>(params: &'a serde_json::Value, name: &str) -> Result<&'a str, ToolError> { + params + .get(name) + .and_then(|v| v.as_str()) + .ok_or_else(|| ToolError::InvalidParameters(format!("missing '{}' parameter", name))) +} + +/// Extract a required parameter of any type from a JSON object. +/// +/// Returns `ToolError::InvalidParameters` if the key is missing. +pub fn require_param<'a>( + params: &'a serde_json::Value, + name: &str, +) -> Result<&'a serde_json::Value, ToolError> { + params + .get(name) + .ok_or_else(|| ToolError::InvalidParameters(format!("missing '{}' parameter", name))) +} + #[cfg(test)] mod tests { use super::*; @@ -235,12 +257,7 @@ mod tests { params: serde_json::Value, _ctx: &JobContext, ) -> Result { - let message = params - .get("message") - .and_then(|v| v.as_str()) - .ok_or_else(|| { - ToolError::InvalidParameters("missing 'message' parameter".to_string()) - })?; + let message = require_str(¶ms, "message")?; Ok(ToolOutput::text(message, Duration::from_millis(1))) } @@ -277,4 +294,40 @@ mod tests { let tool = EchoTool; assert_eq!(tool.execution_timeout(), Duration::from_secs(60)); } + + #[test] + fn test_require_str_present() { + let params = serde_json::json!({"name": "alice"}); + assert_eq!(require_str(¶ms, "name").unwrap(), "alice"); + } + + #[test] + fn test_require_str_missing() { + let params = serde_json::json!({}); + let err = require_str(¶ms, "name").unwrap_err(); + assert!(err.to_string().contains("missing 'name'")); + } + + #[test] + fn test_require_str_wrong_type() { + let params = serde_json::json!({"name": 42}); + let err = require_str(¶ms, "name").unwrap_err(); + assert!(err.to_string().contains("missing 'name'")); + } + + #[test] + fn test_require_param_present() { + let params = serde_json::json!({"data": [1, 2, 3]}); + assert_eq!( + require_param(¶ms, "data").unwrap(), + &serde_json::json!([1, 2, 3]) + ); + } + + #[test] + fn test_require_param_missing() { + let params = serde_json::json!({}); + let err = require_param(¶ms, "data").unwrap_err(); + assert!(err.to_string().contains("missing 'data'")); + } } diff --git a/src/worker/api.rs b/src/worker/api.rs index 2974d21e..fca857bc 100644 --- a/src/worker/api.rs +++ b/src/worker/api.rs @@ -129,11 +129,15 @@ impl WorkerHttpClient { format!("{}/worker/{}/{}", self.orchestrator_url, self.job_id, path) } - /// Fetch the job description from the orchestrator. - pub async fn get_job(&self) -> Result { + /// Send a GET request, check the status, and deserialize the JSON body. + async fn get_json( + &self, + path: &str, + context: &str, + ) -> Result { let resp = self .client - .get(self.url("job")) + .get(self.url(path)) .bearer_auth(&self.token) .send() .await @@ -145,15 +149,51 @@ impl WorkerHttpClient { if !resp.status().is_success() { return Err(WorkerError::OrchestratorRejected { job_id: self.job_id, - reason: format!("GET /job returned {}", resp.status()), + reason: format!("{} returned {}", context, resp.status()), }); } resp.json().await.map_err(|e| WorkerError::LlmProxyFailed { - reason: format!("failed to parse job description: {}", e), + reason: format!("{}: failed to parse response: {}", context, e), }) } + /// Send a POST request with a JSON body, check the status, and deserialize the response. + async fn post_json( + &self, + path: &str, + body: &B, + context: &str, + ) -> Result { + let resp = self + .client + .post(self.url(path)) + .bearer_auth(&self.token) + .json(body) + .send() + .await + .map_err(|e| WorkerError::LlmProxyFailed { + reason: format!("{}: {}", context, e), + })?; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(WorkerError::LlmProxyFailed { + reason: format!("{}: orchestrator returned {}: {}", context, status, body), + }); + } + + resp.json().await.map_err(|e| WorkerError::LlmProxyFailed { + reason: format!("{}: failed to parse response: {}", context, e), + }) + } + + /// Fetch the job description from the orchestrator. + pub async fn get_job(&self) -> Result { + self.get_json("job", "GET /job").await + } + /// Proxy an LLM completion request through the orchestrator. pub async fn llm_complete( &self, @@ -166,29 +206,9 @@ impl WorkerHttpClient { stop_sequences: request.stop_sequences.clone(), }; - let resp = self - .client - .post(self.url("llm/complete")) - .bearer_auth(&self.token) - .json(&proxy_req) - .send() - .await - .map_err(|e| WorkerError::LlmProxyFailed { - reason: e.to_string(), - })?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(WorkerError::LlmProxyFailed { - reason: format!("orchestrator returned {}: {}", status, body), - }); - } - - let proxy_resp: ProxyCompletionResponse = - resp.json().await.map_err(|e| WorkerError::LlmProxyFailed { - reason: format!("failed to parse LLM response: {}", e), - })?; + let proxy_resp: ProxyCompletionResponse = self + .post_json("llm/complete", &proxy_req, "LLM complete") + .await?; Ok(CompletionResponse { content: proxy_resp.content, @@ -212,29 +232,9 @@ impl WorkerHttpClient { tool_choice: request.tool_choice.clone(), }; - let resp = self - .client - .post(self.url("llm/complete_with_tools")) - .bearer_auth(&self.token) - .json(&proxy_req) - .send() - .await - .map_err(|e| WorkerError::LlmProxyFailed { - reason: e.to_string(), - })?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(WorkerError::LlmProxyFailed { - reason: format!("orchestrator returned {}: {}", status, body), - }); - } - - let proxy_resp: ProxyToolCompletionResponse = - resp.json().await.map_err(|e| WorkerError::LlmProxyFailed { - reason: format!("failed to parse tool completion response: {}", e), - })?; + let proxy_resp: ProxyToolCompletionResponse = self + .post_json("llm/complete_with_tools", &proxy_req, "LLM tool complete") + .await?; Ok(ToolCompletionResponse { content: proxy_resp.content, @@ -337,25 +337,9 @@ impl WorkerHttpClient { /// Signal job completion to the orchestrator. pub async fn report_complete(&self, report: &CompletionReport) -> Result<(), WorkerError> { - let resp = self - .client - .post(self.url("complete")) - .bearer_auth(&self.token) - .json(report) - .send() - .await - .map_err(|e| WorkerError::ConnectionFailed { - url: self.orchestrator_url.clone(), - reason: e.to_string(), - })?; - - if !resp.status().is_success() { - return Err(WorkerError::OrchestratorRejected { - job_id: self.job_id, - reason: format!("completion report rejected: {}", resp.status()), - }); - } - + let _: serde_json::Value = self + .post_json("complete", report, "report complete") + .await?; Ok(()) } }