feat: add bulk job removal endpoint (#1310)

Author: paulclindo

shinkai-bin/shinkai-node/src/network/handle_commands_list.rs

@@ -809,6 +809,12 @@ impl Node {
                     let _ = Node::v2_remove_job(db_clone, bearer, job_id, res).await;
                 });
             }
+            NodeCommand::V2ApiRemoveJobs { bearer, job_ids, res } => {
+                let db_clone = self.db.clone();
+                tokio::spawn(async move {
+                    let _ = Node::v2_remove_jobs(db_clone, bearer, job_ids, res).await;
+                });
+            }
             NodeCommand::V2ApiVecFSRetrievePathSimplifiedJson { bearer, payload, res } => {
                 let db_clone = Arc::clone(&self.db);
 

shinkai-bin/shinkai-node/src/network/v2_api/api_v2_commands_jobs.rs

@@ -1697,6 +1697,80 @@ impl Node {
         Ok(())
     }
 
+    pub async fn v2_remove_jobs(
+        db: Arc<SqliteManager>,
+        bearer: String,
+        job_ids: Vec<String>,
+        res: Sender<Result<serde_json::Value, APIError>>,
+    ) -> Result<(), NodeError> {
+        // Validate the bearer token
+        if Self::validate_bearer_token(&bearer, db.clone(), &res).await.is_err() {
+            return Ok(());
+        }
+
+        let mut succeeded: Vec<String> = Vec::new();
+        let mut failed: Vec<serde_json::Value> = Vec::new();
+
+        for job_id in job_ids {
+            // Get the job folder name BEFORE removing from DB (get_job_folder_name queries the jobs table)
+            let job_chat_folder = match db.get_job_folder_name(&job_id) {
+                Ok(folder) => Some(folder),
+                Err(_) => {
+                    // If we can't get the folder name, log but continue - the job might not have a folder yet
+                    None
+                }
+            };
+
+            // Remove the job from database
+            match db.remove_job(&job_id) {
+                Ok(_) => {
+                    // Remove the chat files folder (filesystem/Chat Files/{formatted_name})
+                    if let Some(chat_folder) = job_chat_folder {
+                        use shinkai_fs::shinkai_file_manager::ShinkaiFileManager;
+                        if chat_folder.exists() {
+                            if let Err(err) = ShinkaiFileManager::remove_folder(chat_folder, &db) {
+                                // Log the error but don't fail since DB removal succeeded
+                                eprintln!("Warning: Failed to remove job chat folder for {}: {}", job_id, err);
+                            }
+                        }
+                    }
+
+                    // Remove the tools_storage folder (tools_storage/{job_id}) - used for tool execution files
+                    let node_env = fetch_node_environment();
+                    let node_storage_path = node_env.node_storage_path.unwrap_or_default();
+                    let tools_storage_path = PathBuf::from(&node_storage_path)
+                        .join("tools_storage")
+                        .join(&job_id);
+
+                    if tools_storage_path.exists() {
+                        if let Err(err) = std::fs::remove_dir_all(&tools_storage_path) {
+                            // Log the error but don't fail since DB removal succeeded
+                            eprintln!("Warning: Failed to remove tools_storage folder at {:?}: {}", tools_storage_path, err);
+                        }
+                    }
+
+                    succeeded.push(job_id);
+                }
+                Err(err) => {
+                    failed.push(serde_json::json!({
+                        "job_id": job_id,
+                        "error": format!("Failed to remove job: {}", err)
+                    }));
+                }
+            }
+        }
+
+        let _ = res
+            .send(Ok(serde_json::json!({
+                "status": if failed.is_empty() { "success" } else { "partial" },
+                "message": format!("{} job(s) removed successfully, {} failed", succeeded.len(), failed.len()),
+                "succeeded": succeeded,
+                "failed": failed
+            })))
+            .await;
+        Ok(())
+    }
+
     pub async fn v2_export_messages_from_inbox(
         db: Arc<SqliteManager>,
         bearer: String,

shinkai-libs/shinkai-http-api/src/api_v2/api_v2_handlers_jobs.rs

@@ -175,6 +175,13 @@ pub fn job_routes(
         .and(warp::body::json())
         .and_then(remove_job_handler);
 
+    let remove_jobs_route = warp::path("remove_jobs")
+        .and(warp::post())
+        .and(with_sender(node_commands_sender.clone()))
+        .and(warp::header::<String>("authorization"))
+        .and(warp::body::json())
+        .and_then(remove_jobs_handler);
+
     let export_messages_from_inbox_route = warp::path("export_messages_from_inbox")
         .and(warp::post())
         .and(with_sender(node_commands_sender.clone()))
@@ -216,6 +223,7 @@ pub fn job_routes(
         .or(get_message_traces_route)
         .or(fork_job_messages_route)
         .or(remove_job_route)
+        .or(remove_jobs_route)
         .or(export_messages_from_inbox_route)
         .or(add_messages_god_mode_route)
         .or(get_job_provider_route)
@@ -285,6 +293,11 @@ pub struct RemoveJobRequest {
     pub job_id: String,
 }
 
+#[derive(Deserialize, ToSchema)]
+pub struct RemoveJobsRequest {
+    pub job_ids: Vec<String>,
+}
+
 #[derive(Deserialize, ToSchema)]
 pub struct ExportInboxMessagesRequest {
     pub inbox_name: String,
@@ -1249,6 +1262,45 @@ pub async fn remove_job_handler(
     }
 }
 
+#[utoipa::path(
+    post,
+    path = "/v2/remove_jobs",
+    request_body = RemoveJobsRequest,
+    responses(
+        (status = 200, description = "Successfully removed jobs in bulk", body = Value),
+        (status = 400, description = "Bad request", body = APIError),
+        (status = 500, description = "Internal server error", body = APIError)
+    )
+)]
+pub async fn remove_jobs_handler(
+    node_commands_sender: Sender<NodeCommand>,
+    authorization: String,
+    payload: RemoveJobsRequest,
+) -> Result<impl warp::Reply, warp::Rejection> {
+    let bearer = authorization.strip_prefix("Bearer ").unwrap_or("").to_string();
+    let (res_sender, res_receiver) = async_channel::bounded(1);
+    node_commands_sender
+        .send(NodeCommand::V2ApiRemoveJobs {
+            bearer,
+            job_ids: payload.job_ids,
+            res: res_sender,
+        })
+        .await
+        .map_err(|_| warp::reject::reject())?;
+    let result = res_receiver.recv().await.map_err(|_| warp::reject::reject())?;
+
+    match result {
+        Ok(response) => {
+            let response = create_success_response(response);
+            Ok(warp::reply::with_status(warp::reply::json(&response), StatusCode::OK))
+        }
+        Err(error) => Ok(warp::reply::with_status(
+            warp::reply::json(&error),
+            StatusCode::from_u16(error.code).unwrap(),
+        )),
+    }
+}
+
 #[utoipa::path(
     post,
     path = "/v2/export_messages_from_inbox",
@@ -1398,14 +1450,15 @@ pub async fn get_job_provider_handler(
         get_message_traces_handler,
         fork_job_messages_handler,
         remove_job_handler,
+        remove_jobs_handler,
     ),
     components(
         schemas(AddFileToFolder, V2SmartInbox, APIChangeJobAgentRequest, CreateJobRequest, JobConfig,
             JobMessageRequest, GetLastMessagesRequest, V2ChatMessage, GetLastMessagesWithBranchesRequest,
             UpdateJobConfigRequest, UpdateSmartInboxNameRequest, SerializedLLMProvider, JobCreationInfo,
             JobMessage, NodeApiData, LLMProviderSubset, AssociatedUI, MinimalJobScope, CallbackAction, ShinkaiName,
             LLMProviderInterface, RetryMessageRequest, UpdateJobScopeRequest, ExportInboxMessagesFormat, ExportInboxMessagesRequest,
-            ShinkaiSubidentityType, OpenAI, Ollama, Groq, Gemini, Exo, ShinkaiBackend, SendResponseBody, SendResponseBodyData, APIError, GetToolingLogsRequest, GetMessageTracesRequest, ForkJobMessagesRequest, RemoveJobRequest)
+            ShinkaiSubidentityType, OpenAI, Ollama, Groq, Gemini, Exo, ShinkaiBackend, SendResponseBody, SendResponseBodyData, APIError, GetToolingLogsRequest, GetMessageTracesRequest, ForkJobMessagesRequest, RemoveJobRequest, RemoveJobsRequest)
     ),
     tags(
         (name = "jobs", description = "Job API endpoints")

shinkai-libs/shinkai-http-api/src/node_commands.rs

@@ -289,6 +289,11 @@ pub enum NodeCommand {
         job_id: String,
         res: Sender<Result<SendResponseBody, APIError>>,
     },
+    V2ApiRemoveJobs {
+        bearer: String,
+        job_ids: Vec<String>,
+        res: Sender<Result<Value, APIError>>,
+    },
     V2ApiVecFSRetrievePathSimplifiedJson {
         bearer: String,
         payload: APIVecFsRetrievePathSimplifiedJson,