From 5814d77b16ee40506ade4792abe8e9019e609f75 Mon Sep 17 00:00:00 2001 From: Zaki Date: Fri, 13 Feb 2026 08:24:03 -0800 Subject: [PATCH] fix: add missing JobContext fields and resolve fmt/clippy warnings Add total_tokens_used and max_tokens fields to JobContext in libsql_backend.rs, apply cargo fmt, and fix clippy warnings. Co-Authored-By: Claude Opus 4.6 --- src/agent/agent_loop.rs | 2 +- src/agent/scheduler.rs | 2 +- src/agent/self_repair.rs | 2 +- src/agent/worker.rs | 2 +- src/channels/web/mod.rs | 2 +- src/channels/web/server.rs | 2 +- src/cli/config.rs | 4 +- src/cli/mcp.rs | 35 +- src/cli/tool.rs | 36 +- src/db/libsql_backend.rs | 824 +++++++++++++++++++++---------------- src/db/mod.rs | 55 +-- src/db/postgres.rs | 79 ++-- src/extensions/manager.rs | 6 +- src/history/mod.rs | 4 +- src/lib.rs | 2 +- src/main.rs | 47 ++- src/secrets/store.rs | 8 +- src/setup/channels.rs | 2 +- src/setup/wizard.rs | 33 +- src/tools/registry.rs | 2 +- src/tools/wasm/mod.rs | 5 +- src/tools/wasm/storage.rs | 4 +- src/workspace/mod.rs | 4 +- 23 files changed, 637 insertions(+), 525 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 5ea5f01c..912ea1b4 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -19,9 +19,9 @@ use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusU use crate::config::{AgentConfig, HeartbeatConfig, RoutineConfig}; use crate::context::ContextManager; use crate::context::JobContext; +use crate::db::Database; use crate::error::Error; use crate::extensions::ExtensionManager; -use crate::db::Database; use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult}; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index 3e887b0a..a665c2af 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -12,8 +12,8 @@ use crate::agent::task::{Task, TaskContext, TaskOutput}; use crate::agent::worker::{Worker, WorkerDeps}; use crate::config::AgentConfig; use crate::context::{ContextManager, JobContext, JobState}; -use crate::error::{Error, JobError}; use crate::db::Database; +use crate::error::{Error, JobError}; use crate::llm::LlmProvider; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; diff --git a/src/agent/self_repair.rs b/src/agent/self_repair.rs index b110d163..02a7be1d 100644 --- a/src/agent/self_repair.rs +++ b/src/agent/self_repair.rs @@ -8,8 +8,8 @@ use chrono::{DateTime, Utc}; use uuid::Uuid; use crate::context::{ContextManager, JobState}; -use crate::error::RepairError; use crate::db::Database; +use crate::error::RepairError; use crate::tools::{BuildRequirement, Language, SoftwareBuilder, SoftwareType, ToolRegistry}; /// A job that has been detected as stuck. diff --git a/src/agent/worker.rs b/src/agent/worker.rs index 342d695d..44638b5f 100644 --- a/src/agent/worker.rs +++ b/src/agent/worker.rs @@ -10,8 +10,8 @@ use uuid::Uuid; use crate::agent::scheduler::WorkerMessage; use crate::agent::task::TaskOutput; use crate::context::{ContextManager, JobState}; -use crate::error::Error; use crate::db::Database; +use crate::error::Error; use crate::llm::{ ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolSelection, }; diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 226c91ee..356eda2c 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -32,9 +32,9 @@ use tokio_stream::wrappers::ReceiverStream; use crate::agent::SessionManager; use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; use crate::config::GatewayConfig; +use crate::db::Database; use crate::error::ChannelError; use crate::extensions::ExtensionManager; -use crate::db::Database; use crate::orchestrator::job_manager::ContainerJobManager; use crate::tools::ToolRegistry; use crate::workspace::Workspace; diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index a74f097e..15e8099c 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -30,8 +30,8 @@ use crate::channels::web::auth::{AuthState, auth_middleware}; use crate::channels::web::log_layer::LogBroadcaster; use crate::channels::web::sse::SseManager; use crate::channels::web::types::*; -use crate::extensions::ExtensionManager; use crate::db::Database; +use crate::extensions::ExtensionManager; use crate::orchestrator::job_manager::ContainerJobManager; use crate::tools::ToolRegistry; use crate::workspace::Workspace; diff --git a/src/cli/config.rs b/src/cli/config.rs index f6b52516..6059e0cd 100644 --- a/src/cli/config.rs +++ b/src/cli/config.rs @@ -74,7 +74,9 @@ pub async fn run_config_command(cmd: ConfigCommand) -> anyhow::Result<()> { /// Bootstrap a DB connection for config commands (backend-agnostic). async fn connect_db() -> anyhow::Result> { - let config = crate::config::Config::from_env().await.map_err(|e| anyhow::anyhow!("{}", e))?; + let config = crate::config::Config::from_env() + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; crate::db::connect_from_config(&config.database) .await .map_err(|e| anyhow::anyhow!("{}", e)) diff --git a/src/cli/mcp.rs b/src/cli/mcp.rs index a575abf3..f73659fc 100644 --- a/src/cli/mcp.rs +++ b/src/cli/mcp.rs @@ -9,9 +9,9 @@ use clap::Subcommand; use crate::config::Config; use crate::db::Database; -use crate::secrets::{SecretsCrypto, SecretsStore}; #[cfg(feature = "postgres")] use crate::secrets::PostgresSecretsStore; +use crate::secrets::{SecretsCrypto, SecretsStore}; use crate::tools::mcp::{ McpClient, McpServerConfig, McpSessionManager, OAuthConfig, auth::{authorize_mcp_server, is_authenticated}, @@ -510,36 +510,45 @@ async fn get_secrets_store() -> anyhow::Result anyhow::Result, user_id: String) -> anyho } #[cfg(all(feature = "libsql", not(feature = "postgres")))] { - use crate::db::libsql_backend::LibSqlBackend; use crate::db::Database as _; + use crate::db::libsql_backend::LibSqlBackend; use secrecy::ExposeSecret as _; let default_path = crate::config::default_libsql_path(); - let db_path = config.database.libsql_path.as_deref() + let db_path = config + .database + .libsql_path + .as_deref() .unwrap_or(&default_path); let backend = if let Some(ref url) = config.database.libsql_url { - let token = config.database.libsql_auth_token.as_ref() + let token = config + .database + .libsql_auth_token + .as_ref() .expect("LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set"); - LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await + LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()) + .await .map_err(|e| anyhow::anyhow!("{}", e))? } else { - LibSqlBackend::new_local(db_path).await + LibSqlBackend::new_local(db_path) + .await .map_err(|e| anyhow::anyhow!("{}", e))? }; - backend.run_migrations().await + backend + .run_migrations() + .await .map_err(|e| anyhow::anyhow!("{}", e))?; - let conn = backend.connect() - .map_err(|e| anyhow::anyhow!("{}", e))?; + let conn = backend.connect().map_err(|e| anyhow::anyhow!("{}", e))?; - Arc::new(crate::secrets::LibSqlSecretsStore::new(conn, Arc::new(crypto))) + Arc::new(crate::secrets::LibSqlSecretsStore::new( + conn, + Arc::new(crypto), + )) } #[cfg(not(any(feature = "postgres", feature = "libsql")))] { let _ = crypto; - anyhow::bail!("No database backend available for secrets. Enable 'postgres' or 'libsql' feature."); + anyhow::bail!( + "No database backend available for secrets. Enable 'postgres' or 'libsql' feature." + ); } }; diff --git a/src/db/libsql_backend.rs b/src/db/libsql_backend.rs index 4afd37ab..ccae6a45 100644 --- a/src/db/libsql_backend.rs +++ b/src/db/libsql_backend.rs @@ -11,14 +11,14 @@ use std::path::Path; use async_trait::async_trait; use chrono::{DateTime, NaiveDateTime, Utc}; -use libsql::{params, Connection, Database as LibSqlDatabase}; +use libsql::{Connection, Database as LibSqlDatabase, params}; use rust_decimal::Decimal; use uuid::Uuid; +use crate::agent::BrokenTool; use crate::agent::routine::{ NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger, }; -use crate::agent::BrokenTool; use crate::context::{ActionRecord, JobContext, JobState}; use crate::db::Database; use crate::error::{DatabaseError, WorkspaceError}; @@ -75,7 +75,9 @@ impl LibSqlBackend { let db = libsql::Builder::new_local(":memory:") .build() .await - .map_err(|e| DatabaseError::Pool(format!("Failed to create in-memory database: {}", e)))?; + .map_err(|e| { + DatabaseError::Pool(format!("Failed to create in-memory database: {}", e)) + })?; Ok(Self { db }) } @@ -95,9 +97,7 @@ impl LibSqlBackend { let db = libsql::Builder::new_remote_replica(path, url.to_string(), auth_token.to_string()) .build() .await - .map_err(|e| { - DatabaseError::Pool(format!("Failed to open remote replica: {}", e)) - })?; + .map_err(|e| DatabaseError::Pool(format!("Failed to open remote replica: {}", e)))?; Ok(Self { db }) } @@ -289,11 +289,11 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let id = Uuid::new_v4(); conn.execute( - "INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, opt_text(thread_id)], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4)", + params![id.to_string(), channel, user_id, opt_text(thread_id)], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(id) } @@ -301,11 +301,11 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - "UPDATE conversations SET last_activity = ?2 WHERE id = ?1", - params![id.to_string(), now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE conversations SET last_activity = ?2 WHERE id = ?1", + params![id.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -337,15 +337,15 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" INSERT INTO conversations (id, channel, user_id, thread_id) VALUES (?1, ?2, ?3, ?4) ON CONFLICT (id) DO UPDATE SET last_activity = ?5 "#, - params![id.to_string(), channel, user_id, opt_text(thread_id), now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![id.to_string(), channel, user_id, opt_text(thread_id), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -382,14 +382,22 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut results = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { let metadata = get_json(&row, 3); let thread_type = metadata .get("thread_type") .and_then(|v| v.as_str()) .map(String::from); results.push(ConversationSummary { - id: row.get::(0).unwrap_or_default().parse().unwrap_or_default(), + id: row + .get::(0) + .unwrap_or_default() + .parse() + .unwrap_or_default(), started_at: get_ts(&row, 1), last_activity: get_ts(&row, 2), message_count: get_i64(&row, 4), @@ -420,7 +428,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - if let Some(row) = rows.next().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).unwrap_or_default(); return id_str .parse() @@ -431,11 +443,11 @@ impl Database for LibSqlBackend { let id = Uuid::new_v4(); let metadata = serde_json::json!({"thread_type": "assistant", "title": "Assistant"}); conn.execute( - "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, metadata.to_string()], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", + params![id.to_string(), channel, user_id, metadata.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(id) } @@ -448,11 +460,11 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let id = Uuid::new_v4(); conn.execute( - "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", - params![id.to_string(), channel, user_id, metadata.to_string()], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "INSERT INTO conversations (id, channel, user_id, metadata) VALUES (?1, ?2, ?3, ?4)", + params![id.to_string(), channel, user_id, metadata.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(id) } @@ -468,33 +480,37 @@ impl Database for LibSqlBackend { let mut rows = if let Some(before_ts) = before { conn.query( - r#" + r#" SELECT id, role, content, created_at FROM conversation_messages WHERE conversation_id = ?1 AND created_at < ?2 ORDER BY created_at DESC LIMIT ?3 "#, - params![cid, fmt_ts(&before_ts), fetch_limit], - ) - .await + params![cid, fmt_ts(&before_ts), fetch_limit], + ) + .await } else { conn.query( - r#" + r#" SELECT id, role, content, created_at FROM conversation_messages WHERE conversation_id = ?1 ORDER BY created_at DESC LIMIT ?2 "#, - params![cid, fetch_limit], - ) - .await + params![cid, fetch_limit], + ) + .await } .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut all = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { all.push(ConversationMessage { id: get_text(&row, 0).parse().unwrap_or_default(), role: get_text(&row, 1), @@ -519,11 +535,11 @@ impl Database for LibSqlBackend { // SQLite: use json_patch to merge the key let patch = serde_json::json!({ key: value }); conn.execute( - "UPDATE conversations SET metadata = json_patch(metadata, ?2) WHERE id = ?1", - params![id.to_string(), patch.to_string()], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE conversations SET metadata = json_patch(metadata, ?2) WHERE id = ?1", + params![id.to_string(), patch.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -540,7 +556,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(get_json(&row, 0))), None => Ok(None), } @@ -565,7 +585,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut messages = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { messages.push(ConversationMessage { id: get_text(&row, 0).parse().unwrap_or_default(), role: get_text(&row, 1), @@ -663,7 +687,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => { let status_str = get_text(&row, 5); let state = parse_job_state(&status_str); @@ -684,6 +712,8 @@ impl Database for LibSqlBackend { estimated_duration: estimated_time_secs .map(|s| std::time::Duration::from_secs(s as u64)), actual_cost: get_decimal(&row, 12), + total_tokens_used: 0, + max_tokens: 0, repair_attempts: get_i64(&row, 13) as u32, created_at: get_ts(&row, 14), started_at: get_opt_ts(&row, 15), @@ -704,11 +734,11 @@ impl Database for LibSqlBackend { ) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - "UPDATE agent_jobs SET status = ?2, failure_reason = ?3 WHERE id = ?1", - params![id.to_string(), status.to_string(), opt_text(failure_reason)], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE agent_jobs SET status = ?2, failure_reason = ?3 WHERE id = ?1", + params![id.to_string(), status.to_string(), opt_text(failure_reason)], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -716,11 +746,11 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - "UPDATE agent_jobs SET status = 'stuck', stuck_since = ?2 WHERE id = ?1", - params![id.to_string(), now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE agent_jobs SET status = 'stuck', stuck_since = ?2 WHERE id = ?1", + params![id.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -732,7 +762,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut ids = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { if let Ok(id_str) = row.get::(0) { if let Ok(id) = id_str.parse() { ids.push(id); @@ -744,48 +778,41 @@ impl Database for LibSqlBackend { // ==================== Actions ==================== - async fn save_action( - &self, - job_id: Uuid, - action: &ActionRecord, - ) -> Result<(), DatabaseError> { + async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError> { let conn = self.connect()?; let duration_ms = action.duration.as_millis() as i64; let warnings_json = serde_json::to_string(&action.sanitization_warnings) .map_err(|e| DatabaseError::Serialization(e.to_string()))?; conn.execute( - r#" + r#" INSERT INTO job_actions ( id, job_id, sequence_num, tool_name, input, output_raw, output_sanitized, sanitization_warnings, cost, duration_ms, success, error_message, created_at ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13) "#, - params![ - action.id.to_string(), - job_id.to_string(), - action.sequence as i64, - action.tool_name.as_str(), - action.input.to_string(), - opt_text(action.output_raw.as_deref()), - opt_text_owned(action.output_sanitized.as_ref().map(|v| v.to_string())), - warnings_json, - opt_text_owned(action.cost.map(|d| d.to_string())), - duration_ms, - action.success as i64, - opt_text(action.error.as_deref()), - fmt_ts(&action.executed_at), - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + action.id.to_string(), + job_id.to_string(), + action.sequence as i64, + action.tool_name.as_str(), + action.input.to_string(), + opt_text(action.output_raw.as_deref()), + opt_text_owned(action.output_sanitized.as_ref().map(|v| v.to_string())), + warnings_json, + opt_text_owned(action.cost.map(|d| d.to_string())), + duration_ms, + action.success as i64, + opt_text(action.error.as_deref()), + fmt_ts(&action.executed_at), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn get_job_actions( - &self, - job_id: Uuid, - ) -> Result, DatabaseError> { + async fn get_job_actions(&self, job_id: Uuid) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -800,16 +827,20 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut actions = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { - let warnings: Vec = serde_json::from_str(&get_text(&row, 6)).unwrap_or_default(); + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + let warnings: Vec = + serde_json::from_str(&get_text(&row, 6)).unwrap_or_default(); actions.push(ActionRecord { id: get_text(&row, 0).parse().unwrap_or_default(), sequence: get_i64(&row, 1) as u32, tool_name: get_text(&row, 2), input: get_json(&row, 3), output_raw: get_opt_text(&row, 4), - output_sanitized: get_opt_text(&row, 5) - .and_then(|s| serde_json::from_str(&s).ok()), + output_sanitized: get_opt_text(&row, 5).and_then(|s| serde_json::from_str(&s).ok()), sanitization_warnings: warnings, cost: get_opt_decimal(&row, 7), duration: std::time::Duration::from_millis(get_i64(&row, 8) as u64), @@ -911,7 +942,7 @@ impl Database for LibSqlBackend { async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - r#" + r#" INSERT INTO agent_jobs ( id, title, description, status, source, user_id, project_dir, success, failure_reason, created_at, started_at, completed_at @@ -923,28 +954,25 @@ impl Database for LibSqlBackend { started_at = excluded.started_at, completed_at = excluded.completed_at "#, - params![ - job.id.to_string(), - job.task.as_str(), - job.status.as_str(), - job.user_id.as_str(), - job.project_dir.as_str(), - job.success.map(|b| b as i64), - opt_text(job.failure_reason.as_deref()), - fmt_ts(&job.created_at), - fmt_opt_ts(&job.started_at), - fmt_opt_ts(&job.completed_at), - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + job.id.to_string(), + job.task.as_str(), + job.status.as_str(), + job.user_id.as_str(), + job.project_dir.as_str(), + job.success.map(|b| b as i64), + opt_text(job.failure_reason.as_deref()), + fmt_ts(&job.created_at), + fmt_opt_ts(&job.started_at), + fmt_opt_ts(&job.completed_at), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn get_sandbox_job( - &self, - id: Uuid, - ) -> Result, DatabaseError> { + async fn get_sandbox_job(&self, id: Uuid) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -958,7 +986,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(SandboxJobRecord { id: get_text(&row, 0).parse().unwrap_or_default(), task: get_text(&row, 1), @@ -991,7 +1023,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut jobs = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { jobs.push(SandboxJobRecord { id: get_text(&row, 0).parse().unwrap_or_default(), task: get_text(&row, 1), @@ -1019,7 +1055,7 @@ impl Database for LibSqlBackend { ) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - r#" + r#" UPDATE agent_jobs SET status = ?2, success = COALESCE(?3, success), @@ -1028,17 +1064,17 @@ impl Database for LibSqlBackend { completed_at = COALESCE(?6, completed_at) WHERE id = ?1 AND source = 'sandbox' "#, - params![ - id.to_string(), - status, - success.map(|b| b as i64), - message, - fmt_opt_ts(&started_at), - fmt_opt_ts(&completed_at), - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + id.to_string(), + status, + success.map(|b| b as i64), + message, + fmt_opt_ts(&started_at), + fmt_opt_ts(&completed_at), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1075,7 +1111,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut summary = SandboxJobSummary::default(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { let status = get_text(&row, 0); let count = get_i64(&row, 1) as usize; summary.total += count; @@ -1110,7 +1150,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut jobs = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { jobs.push(SandboxJobRecord { id: get_text(&row, 0).parse().unwrap_or_default(), task: get_text(&row, 1), @@ -1141,7 +1185,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut summary = SandboxJobSummary::default(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { let status = get_text(&row, 0); let count = get_i64(&row, 1) as usize; summary.total += count; @@ -1177,25 +1225,18 @@ impl Database for LibSqlBackend { Ok(found.is_some()) } - async fn update_sandbox_job_mode( - &self, - id: Uuid, - mode: &str, - ) -> Result<(), DatabaseError> { + async fn update_sandbox_job_mode(&self, id: Uuid, mode: &str) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - "UPDATE agent_jobs SET job_mode = ?2 WHERE id = ?1", - params![id.to_string(), mode], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE agent_jobs SET job_mode = ?2 WHERE id = ?1", + params![id.to_string(), mode], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn get_sandbox_job_mode( - &self, - id: Uuid, - ) -> Result, DatabaseError> { + async fn get_sandbox_job_mode(&self, id: Uuid) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -1205,7 +1246,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(get_text(&row, 0))), None => Ok(None), } @@ -1221,18 +1266,15 @@ impl Database for LibSqlBackend { ) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - "INSERT INTO job_events (job_id, event_type, data) VALUES (?1, ?2, ?3)", - params![job_id.to_string(), event_type, data.to_string()], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "INSERT INTO job_events (job_id, event_type, data) VALUES (?1, ?2, ?3)", + params![job_id.to_string(), event_type, data.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn list_job_events( - &self, - job_id: Uuid, - ) -> Result, DatabaseError> { + async fn list_job_events(&self, job_id: Uuid) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -1246,7 +1288,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut events = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { events.push(JobEventRecord { id: get_i64(&row, 0), job_id: get_text(&row, 1).parse().unwrap_or_default(), @@ -1268,10 +1314,7 @@ impl Database for LibSqlBackend { let action_config = routine.action.to_config_json(); let cooldown_secs = routine.guardrails.cooldown.as_secs() as i64; let max_concurrent = routine.guardrails.max_concurrent as i64; - let dedup_window_secs = routine - .guardrails - .dedup_window - .map(|d| d.as_secs() as i64); + let dedup_window_secs = routine.guardrails.dedup_window.map(|d| d.as_secs() as i64); conn.execute( r#" @@ -1328,7 +1371,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(row_to_routine_libsql(&row)?)), None => Ok(None), } @@ -1342,13 +1389,20 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let mut rows = conn .query( - &format!("SELECT {} FROM routines WHERE user_id = ?1 AND name = ?2", ROUTINE_COLUMNS), + &format!( + "SELECT {} FROM routines WHERE user_id = ?1 AND name = ?2", + ROUTINE_COLUMNS + ), params![user_id, name], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(row_to_routine_libsql(&row)?)), None => Ok(None), } @@ -1358,14 +1412,21 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let mut rows = conn .query( - &format!("SELECT {} FROM routines WHERE user_id = ?1 ORDER BY name", ROUTINE_COLUMNS), + &format!( + "SELECT {} FROM routines WHERE user_id = ?1 ORDER BY name", + ROUTINE_COLUMNS + ), params![user_id], ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut routines = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { routines.push(row_to_routine_libsql(&row)?); } Ok(routines) @@ -1375,14 +1436,21 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let mut rows = conn .query( - &format!("SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'event'", ROUTINE_COLUMNS), + &format!( + "SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'event'", + ROUTINE_COLUMNS + ), (), ) .await .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut routines = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { routines.push(row_to_routine_libsql(&row)?); } Ok(routines) @@ -1403,7 +1471,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut routines = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { routines.push(row_to_routine_libsql(&row)?); } Ok(routines) @@ -1417,14 +1489,11 @@ impl Database for LibSqlBackend { let action_config = routine.action.to_config_json(); let cooldown_secs = routine.guardrails.cooldown.as_secs() as i64; let max_concurrent = routine.guardrails.max_concurrent as i64; - let dedup_window_secs = routine - .guardrails - .dedup_window - .map(|d| d.as_secs() as i64); + let dedup_window_secs = routine.guardrails.dedup_window.map(|d| d.as_secs() as i64); let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" UPDATE routines SET name = ?2, description = ?3, enabled = ?4, trigger_type = ?5, trigger_config = ?6, @@ -1436,30 +1505,30 @@ impl Database for LibSqlBackend { updated_at = ?19 WHERE id = ?1 "#, - params![ - routine.id.to_string(), - routine.name.as_str(), - routine.description.as_str(), - routine.enabled as i64, - trigger_type, - trigger_config.to_string(), - action_type, - action_config.to_string(), - cooldown_secs, - max_concurrent, - dedup_window_secs, - opt_text(routine.notify.channel.as_deref()), - routine.notify.user.as_str(), - routine.notify.on_success as i64, - routine.notify.on_failure as i64, - routine.notify.on_attention as i64, - routine.state.to_string(), - fmt_opt_ts(&routine.next_fire_at), - now, - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + routine.id.to_string(), + routine.name.as_str(), + routine.description.as_str(), + routine.enabled as i64, + trigger_type, + trigger_config.to_string(), + action_type, + action_config.to_string(), + cooldown_secs, + max_concurrent, + dedup_window_secs, + opt_text(routine.notify.channel.as_deref()), + routine.notify.user.as_str(), + routine.notify.on_success as i64, + routine.notify.on_failure as i64, + routine.notify.on_attention as i64, + routine.state.to_string(), + fmt_opt_ts(&routine.next_fire_at), + now, + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1475,25 +1544,25 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" UPDATE routines SET last_run_at = ?2, next_fire_at = ?3, run_count = ?4, consecutive_failures = ?5, state = ?6, updated_at = ?7 WHERE id = ?1 "#, - params![ - id.to_string(), - fmt_ts(&last_run_at), - fmt_opt_ts(&next_fire_at), - run_count as i64, - consecutive_failures as i64, - state.to_string(), - now, - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + id.to_string(), + fmt_ts(&last_run_at), + fmt_opt_ts(&next_fire_at), + run_count as i64, + consecutive_failures as i64, + state.to_string(), + now, + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1514,24 +1583,24 @@ impl Database for LibSqlBackend { async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - r#" + r#" INSERT INTO routine_runs ( id, routine_id, trigger_type, trigger_detail, started_at, status, job_id ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) "#, - params![ - run.id.to_string(), - run.routine_id.to_string(), - run.trigger_type.as_str(), - opt_text(run.trigger_detail.as_deref()), - fmt_ts(&run.started_at), - run.status.to_string(), - opt_text_owned(run.job_id.map(|id| id.to_string())), - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + run.id.to_string(), + run.routine_id.to_string(), + run.trigger_type.as_str(), + opt_text(run.trigger_detail.as_deref()), + fmt_ts(&run.started_at), + run.status.to_string(), + opt_text_owned(run.job_id.map(|id| id.to_string())), + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1545,22 +1614,22 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" UPDATE routine_runs SET completed_at = ?5, status = ?2, result_summary = ?3, tokens_used = ?4 WHERE id = ?1 "#, - params![ - id.to_string(), - status.to_string(), - opt_text(result_summary), - tokens_used.map(|t| t as i64), - now, - ], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![ + id.to_string(), + status.to_string(), + opt_text(result_summary), + tokens_used.map(|t| t as i64), + now, + ], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1582,16 +1651,17 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut runs = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { runs.push(row_to_routine_run_libsql(&row)?); } Ok(runs) } - async fn count_running_routine_runs( - &self, - routine_id: Uuid, - ) -> Result { + async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result { let conn = self.connect()?; let mut rows = conn .query( @@ -1601,7 +1671,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(get_i64(&row, 0)), None => Ok(0), } @@ -1617,7 +1691,7 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" INSERT INTO tool_failures (id, tool_name, error_message, error_count, last_failure) VALUES (?1, ?2, ?3, 1, ?4) ON CONFLICT (tool_name) DO UPDATE SET @@ -1625,17 +1699,14 @@ impl Database for LibSqlBackend { error_count = tool_failures.error_count + 1, last_failure = ?4 "#, - params![Uuid::new_v4().to_string(), tool_name, error_message, now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![Uuid::new_v4().to_string(), tool_name, error_message, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn get_broken_tools( - &self, - threshold: i32, - ) -> Result, DatabaseError> { + async fn get_broken_tools(&self, threshold: i32) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -1652,7 +1723,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut tools = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { tools.push(BrokenTool { name: get_text(&row, 0), last_error: get_opt_text(&row, 1), @@ -1671,22 +1746,22 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - "UPDATE tool_failures SET repaired_at = ?2, error_count = 0 WHERE tool_name = ?1", - params![tool_name, now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE tool_failures SET repaired_at = ?2, error_count = 0 WHERE tool_name = ?1", + params![tool_name, now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError> { let conn = self.connect()?; conn.execute( - "UPDATE tool_failures SET repair_attempts = repair_attempts + 1 WHERE tool_name = ?1", - params![tool_name], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + "UPDATE tool_failures SET repair_attempts = repair_attempts + 1 WHERE tool_name = ?1", + params![tool_name], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } @@ -1706,7 +1781,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(get_json(&row, 0))), None => Ok(None), } @@ -1726,7 +1805,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(Some(SettingRow { key: get_text(&row, 0), value: get_json(&row, 1), @@ -1745,25 +1828,21 @@ impl Database for LibSqlBackend { let conn = self.connect()?; let now = fmt_ts(&Utc::now()); conn.execute( - r#" + r#" INSERT INTO settings (user_id, key, value, updated_at) VALUES (?1, ?2, ?3, ?4) ON CONFLICT (user_id, key) DO UPDATE SET value = excluded.value, updated_at = ?4 "#, - params![user_id, key, value.to_string(), now], - ) - .await - .map_err(|e| DatabaseError::Query(e.to_string()))?; + params![user_id, key, value.to_string(), now], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } - async fn delete_setting( - &self, - user_id: &str, - key: &str, - ) -> Result { + async fn delete_setting(&self, user_id: &str, key: &str) -> Result { let conn = self.connect()?; let count = conn .execute( @@ -1775,10 +1854,7 @@ impl Database for LibSqlBackend { Ok(count > 0) } - async fn list_settings( - &self, - user_id: &str, - ) -> Result, DatabaseError> { + async fn list_settings(&self, user_id: &str) -> Result, DatabaseError> { let conn = self.connect()?; let mut rows = conn .query( @@ -1789,7 +1865,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut settings = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { settings.push(SettingRow { key: get_text(&row, 0), value: get_json(&row, 1), @@ -1813,7 +1893,11 @@ impl Database for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; let mut map = HashMap::new(); - while let Some(row) = rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { map.insert(get_text(&row, 0), get_json(&row, 1)); } Ok(map) @@ -1865,7 +1949,11 @@ impl Database for LibSqlBackend { .await .map_err(|e| DatabaseError::Query(e.to_string()))?; - match rows.next().await.map_err(|e| DatabaseError::Query(e.to_string()))? { + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { Some(row) => Ok(get_i64(&row, 0) > 0), None => Ok(false), } @@ -1898,9 +1986,12 @@ impl Database for LibSqlBackend { reason: format!("Query failed: {}", e), })?; - match rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + match rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? { Some(row) => Ok(row_to_memory_document(&row)), None => Err(WorkspaceError::DocumentNotFound { doc_type: path.to_string(), @@ -1927,9 +2018,12 @@ impl Database for LibSqlBackend { reason: format!("Query failed: {}", e), })?; - match rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + match rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? { Some(row) => Ok(row_to_memory_document(&row)), None => Err(WorkspaceError::DocumentNotFound { doc_type: "unknown".to_string(), @@ -1958,17 +2052,17 @@ impl Database for LibSqlBackend { let id = Uuid::new_v4(); let agent_id_str = agent_id.map(|id| id.to_string()); conn.execute( - r#" + r#" INSERT INTO memory_documents (id, user_id, agent_id, path, content, metadata) VALUES (?1, ?2, ?3, ?4, '', '{}') ON CONFLICT (user_id, agent_id, path) DO NOTHING "#, - params![id.to_string(), user_id, agent_id_str.as_deref(), path], - ) - .await - .map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Insert failed: {}", e), - })?; + params![id.to_string(), user_id, agent_id_str.as_deref(), path], + ) + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Insert failed: {}", e), + })?; self.get_document_by_path(user_id, agent_id, path).await } @@ -1979,13 +2073,13 @@ impl Database for LibSqlBackend { })?; let now = fmt_ts(&Utc::now()); conn.execute( - "UPDATE memory_documents SET content = ?2, updated_at = ?3 WHERE id = ?1", - params![id.to_string(), content, now], - ) - .await - .map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Update failed: {}", e), - })?; + "UPDATE memory_documents SET content = ?2, updated_at = ?3 WHERE id = ?1", + params![id.to_string(), content, now], + ) + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Update failed: {}", e), + })?; Ok(()) } @@ -2003,13 +2097,13 @@ impl Database for LibSqlBackend { })?; let agent_id_str = agent_id.map(|id| id.to_string()); conn.execute( - "DELETE FROM memory_documents WHERE user_id = ?1 AND agent_id IS ?2 AND path = ?3", - params![user_id, agent_id_str.as_deref(), path], - ) - .await - .map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Delete failed: {}", e), - })?; + "DELETE FROM memory_documents WHERE user_id = ?1 AND agent_id IS ?2 AND path = ?3", + params![user_id, agent_id_str.as_deref(), path], + ) + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Delete failed: {}", e), + })?; Ok(()) } @@ -2054,9 +2148,13 @@ impl Database for LibSqlBackend { let mut entries_map: HashMap = HashMap::new(); - while let Some(row) = rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? + { let full_path = get_text(&row, 0); let updated_at = get_opt_ts(&row, 1); let content_preview = get_opt_text(&row, 2); @@ -2135,9 +2233,13 @@ impl Database for LibSqlBackend { })?; let mut paths = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? + { paths.push(get_text(&row, 0)); } Ok(paths) @@ -2169,9 +2271,13 @@ impl Database for LibSqlBackend { })?; let mut docs = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? + { docs.push(row_to_memory_document(&row)); } Ok(docs) @@ -2184,13 +2290,13 @@ impl Database for LibSqlBackend { reason: e.to_string(), })?; conn.execute( - "DELETE FROM memory_chunks WHERE document_id = ?1", - params![document_id.to_string()], - ) - .await - .map_err(|e| WorkspaceError::ChunkingFailed { - reason: format!("Delete failed: {}", e), - })?; + "DELETE FROM memory_chunks WHERE document_id = ?1", + params![document_id.to_string()], + ) + .await + .map_err(|e| WorkspaceError::ChunkingFailed { + reason: format!("Delete failed: {}", e), + })?; Ok(()) } @@ -2212,22 +2318,22 @@ impl Database for LibSqlBackend { }); conn.execute( - r#" + r#" INSERT INTO memory_chunks (id, document_id, chunk_index, content, embedding) VALUES (?1, ?2, ?3, ?4, ?5) "#, - params![ - id.to_string(), - document_id.to_string(), - chunk_index as i64, - content, - embedding_blob.map(libsql::Value::Blob), - ], - ) - .await - .map_err(|e| WorkspaceError::ChunkingFailed { - reason: format!("Insert failed: {}", e), - })?; + params![ + id.to_string(), + document_id.to_string(), + chunk_index as i64, + content, + embedding_blob.map(libsql::Value::Blob), + ], + ) + .await + .map_err(|e| WorkspaceError::ChunkingFailed { + reason: format!("Insert failed: {}", e), + })?; Ok(id) } @@ -2236,19 +2342,21 @@ impl Database for LibSqlBackend { chunk_id: Uuid, embedding: &[f32], ) -> Result<(), WorkspaceError> { - let conn = self.connect().map_err(|e| WorkspaceError::EmbeddingFailed { - reason: e.to_string(), - })?; + let conn = self + .connect() + .map_err(|e| WorkspaceError::EmbeddingFailed { + reason: e.to_string(), + })?; let bytes: Vec = embedding.iter().flat_map(|f| f.to_le_bytes()).collect(); conn.execute( - "UPDATE memory_chunks SET embedding = ?2 WHERE id = ?1", - params![chunk_id.to_string(), libsql::Value::Blob(bytes)], - ) - .await - .map_err(|e| WorkspaceError::EmbeddingFailed { - reason: format!("Update failed: {}", e), - })?; + "UPDATE memory_chunks SET embedding = ?2 WHERE id = ?1", + params![chunk_id.to_string(), libsql::Value::Blob(bytes)], + ) + .await + .map_err(|e| WorkspaceError::EmbeddingFailed { + reason: format!("Update failed: {}", e), + })?; Ok(()) } @@ -2280,9 +2388,13 @@ impl Database for LibSqlBackend { })?; let mut chunks = Vec::new(); - while let Some(row) = rows.next().await.map_err(|e| WorkspaceError::SearchFailed { - reason: format!("Query failed: {}", e), - })? { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { + reason: format!("Query failed: {}", e), + })? + { chunks.push(MemoryChunk { id: get_text(&row, 0).parse().unwrap_or_default(), document_id: get_text(&row, 1).parse().unwrap_or_default(), @@ -2333,8 +2445,10 @@ impl Database for LibSqlBackend { })?; let mut results = Vec::new(); - while let Some(row) = - rows.next().await.map_err(|e| WorkspaceError::SearchFailed { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { reason: format!("FTS row fetch failed: {}", e), })? { @@ -2380,8 +2494,10 @@ impl Database for LibSqlBackend { })?; let mut results = Vec::new(); - while let Some(row) = - rows.next().await.map_err(|e| WorkspaceError::SearchFailed { + while let Some(row) = rows + .next() + .await + .map_err(|e| WorkspaceError::SearchFailed { reason: format!("Vector row fetch failed: {}", e), })? { @@ -2398,7 +2514,9 @@ impl Database for LibSqlBackend { }; if embedding.is_some() && !config.use_vector { - tracing::warn!("Embedding provided but vector search is disabled in config; using FTS-only results"); + tracing::warn!( + "Embedding provided but vector search is disabled in config; using FTS-only results" + ); } Ok(reciprocal_rank_fusion(fts_results, vector_results, config)) diff --git a/src/db/mod.rs b/src/db/mod.rs index 95cb33d1..cf684141 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -26,8 +26,8 @@ use chrono::{DateTime, Utc}; use rust_decimal::Decimal; use uuid::Uuid; -use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::agent::BrokenTool; +use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::context::{ActionRecord, JobContext, JobState}; use crate::error::DatabaseError; use crate::error::WorkspaceError; @@ -214,17 +214,10 @@ pub trait Database: Send + Sync { // ==================== Actions ==================== /// Save a job action. - async fn save_action( - &self, - job_id: Uuid, - action: &ActionRecord, - ) -> Result<(), DatabaseError>; + async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError>; /// Get actions for a job. - async fn get_job_actions( - &self, - job_id: Uuid, - ) -> Result, DatabaseError>; + async fn get_job_actions(&self, job_id: Uuid) -> Result, DatabaseError>; // ==================== LLM Calls ==================== @@ -259,10 +252,7 @@ pub trait Database: Send + Sync { async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError>; /// Get a sandbox job by ID. - async fn get_sandbox_job( - &self, - id: Uuid, - ) -> Result, DatabaseError>; + async fn get_sandbox_job(&self, id: Uuid) -> Result, DatabaseError>; /// List all sandbox jobs, most recent first. async fn list_sandbox_jobs(&self) -> Result, DatabaseError>; @@ -304,17 +294,10 @@ pub trait Database: Send + Sync { ) -> Result; /// Update sandbox job mode. - async fn update_sandbox_job_mode( - &self, - id: Uuid, - mode: &str, - ) -> Result<(), DatabaseError>; + async fn update_sandbox_job_mode(&self, id: Uuid, mode: &str) -> Result<(), DatabaseError>; /// Get sandbox job mode. - async fn get_sandbox_job_mode( - &self, - id: Uuid, - ) -> Result, DatabaseError>; + async fn get_sandbox_job_mode(&self, id: Uuid) -> Result, DatabaseError>; // ==================== Job Events ==================== @@ -327,10 +310,7 @@ pub trait Database: Send + Sync { ) -> Result<(), DatabaseError>; /// Load all job events. - async fn list_job_events( - &self, - job_id: Uuid, - ) -> Result, DatabaseError>; + async fn list_job_events(&self, job_id: Uuid) -> Result, DatabaseError>; // ==================== Routines ==================== @@ -395,10 +375,7 @@ pub trait Database: Send + Sync { ) -> Result, DatabaseError>; /// Count currently running runs for a routine. - async fn count_running_routine_runs( - &self, - routine_id: Uuid, - ) -> Result; + async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result; // ==================== Tool Failures ==================== @@ -410,10 +387,7 @@ pub trait Database: Send + Sync { ) -> Result<(), DatabaseError>; /// Get broken tools exceeding threshold. - async fn get_broken_tools( - &self, - threshold: i32, - ) -> Result, DatabaseError>; + async fn get_broken_tools(&self, threshold: i32) -> Result, DatabaseError>; /// Mark a tool as repaired. async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError>; @@ -446,17 +420,10 @@ pub trait Database: Send + Sync { ) -> Result<(), DatabaseError>; /// Delete a single setting. - async fn delete_setting( - &self, - user_id: &str, - key: &str, - ) -> Result; + async fn delete_setting(&self, user_id: &str, key: &str) -> Result; /// List all settings for a user. - async fn list_settings( - &self, - user_id: &str, - ) -> Result, DatabaseError>; + async fn list_settings(&self, user_id: &str) -> Result, DatabaseError>; /// Get all settings as a flat map. async fn get_all_settings( diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 5cc516af..9676144c 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -11,8 +11,8 @@ use deadpool_postgres::Pool; use rust_decimal::Decimal; use uuid::Uuid; -use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::agent::BrokenTool; +use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::config::DatabaseConfig; use crate::context::{ActionRecord, JobContext, JobState}; use crate::db::Database; @@ -65,7 +65,9 @@ impl Database for PgBackend { user_id: &str, thread_id: Option<&str>, ) -> Result { - self.store.create_conversation(channel, user_id, thread_id).await + self.store + .create_conversation(channel, user_id, thread_id) + .await } async fn touch_conversation(&self, id: Uuid) -> Result<(), DatabaseError> { @@ -160,9 +162,7 @@ impl Database for PgBackend { &self, conversation_id: Uuid, ) -> Result, DatabaseError> { - self.store - .list_conversation_messages(conversation_id) - .await + self.store.list_conversation_messages(conversation_id).await } async fn conversation_belongs_to_user( @@ -191,7 +191,9 @@ impl Database for PgBackend { status: JobState, failure_reason: Option<&str>, ) -> Result<(), DatabaseError> { - self.store.update_job_status(id, status, failure_reason).await + self.store + .update_job_status(id, status, failure_reason) + .await } async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError> { @@ -204,18 +206,11 @@ impl Database for PgBackend { // ==================== Actions ==================== - async fn save_action( - &self, - job_id: Uuid, - action: &ActionRecord, - ) -> Result<(), DatabaseError> { + async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError> { self.store.save_action(job_id, action).await } - async fn get_job_actions( - &self, - job_id: Uuid, - ) -> Result, DatabaseError> { + async fn get_job_actions(&self, job_id: Uuid) -> Result, DatabaseError> { self.store.get_job_actions(job_id).await } @@ -266,10 +261,7 @@ impl Database for PgBackend { self.store.save_sandbox_job(job).await } - async fn get_sandbox_job( - &self, - id: Uuid, - ) -> Result, DatabaseError> { + async fn get_sandbox_job(&self, id: Uuid) -> Result, DatabaseError> { self.store.get_sandbox_job(id).await } @@ -318,21 +310,16 @@ impl Database for PgBackend { job_id: Uuid, user_id: &str, ) -> Result { - self.store.sandbox_job_belongs_to_user(job_id, user_id).await + self.store + .sandbox_job_belongs_to_user(job_id, user_id) + .await } - async fn update_sandbox_job_mode( - &self, - id: Uuid, - mode: &str, - ) -> Result<(), DatabaseError> { + async fn update_sandbox_job_mode(&self, id: Uuid, mode: &str) -> Result<(), DatabaseError> { self.store.update_sandbox_job_mode(id, mode).await } - async fn get_sandbox_job_mode( - &self, - id: Uuid, - ) -> Result, DatabaseError> { + async fn get_sandbox_job_mode(&self, id: Uuid) -> Result, DatabaseError> { self.store.get_sandbox_job_mode(id).await } @@ -347,10 +334,7 @@ impl Database for PgBackend { self.store.save_job_event(job_id, event_type, data).await } - async fn list_job_events( - &self, - job_id: Uuid, - ) -> Result, DatabaseError> { + async fn list_job_events(&self, job_id: Uuid) -> Result, DatabaseError> { self.store.list_job_events(job_id).await } @@ -439,10 +423,7 @@ impl Database for PgBackend { self.store.list_routine_runs(routine_id, limit).await } - async fn count_running_routine_runs( - &self, - routine_id: Uuid, - ) -> Result { + async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result { self.store.count_running_routine_runs(routine_id).await } @@ -453,13 +434,12 @@ impl Database for PgBackend { tool_name: &str, error_message: &str, ) -> Result<(), DatabaseError> { - self.store.record_tool_failure(tool_name, error_message).await + self.store + .record_tool_failure(tool_name, error_message) + .await } - async fn get_broken_tools( - &self, - threshold: i32, - ) -> Result, DatabaseError> { + async fn get_broken_tools(&self, threshold: i32) -> Result, DatabaseError> { self.store.get_broken_tools(threshold).await } @@ -498,18 +478,11 @@ impl Database for PgBackend { self.store.set_setting(user_id, key, value).await } - async fn delete_setting( - &self, - user_id: &str, - key: &str, - ) -> Result { + async fn delete_setting(&self, user_id: &str, key: &str) -> Result { self.store.delete_setting(user_id, key).await } - async fn list_settings( - &self, - user_id: &str, - ) -> Result, DatabaseError> { + async fn list_settings(&self, user_id: &str) -> Result, DatabaseError> { self.store.list_settings(user_id).await } @@ -540,7 +513,9 @@ impl Database for PgBackend { agent_id: Option, path: &str, ) -> Result { - self.repo.get_document_by_path(user_id, agent_id, path).await + self.repo + .get_document_by_path(user_id, agent_id, path) + .await } async fn get_document_by_id(&self, id: Uuid) -> Result { diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 735759ee..ef8ee6a3 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -375,7 +375,8 @@ impl ExtensionManager { ) -> Result<(), crate::tools::mcp::config::ConfigError> { config.validate()?; if let Some(ref store) = self.store { - crate::tools::mcp::config::add_mcp_server_db(store.as_ref(), &self.user_id, config).await + crate::tools::mcp::config::add_mcp_server_db(store.as_ref(), &self.user_id, config) + .await } else { crate::tools::mcp::config::add_mcp_server(config).await } @@ -386,7 +387,8 @@ impl ExtensionManager { name: &str, ) -> Result<(), crate::tools::mcp::config::ConfigError> { if let Some(ref store) = self.store { - crate::tools::mcp::config::remove_mcp_server_db(store.as_ref(), &self.user_id, name).await + crate::tools::mcp::config::remove_mcp_server_db(store.as_ref(), &self.user_id, name) + .await } else { crate::tools::mcp::config::remove_mcp_server(name).await } diff --git a/src/history/mod.rs b/src/history/mod.rs index 11776d7d..4f448cfc 100644 --- a/src/history/mod.rs +++ b/src/history/mod.rs @@ -11,9 +11,9 @@ mod store; #[cfg(feature = "postgres")] pub use analytics::{JobStats, ToolStats}; +#[cfg(feature = "postgres")] +pub use store::Store; pub use store::{ ConversationMessage, ConversationSummary, JobEventRecord, LlmCallRecord, SandboxJobRecord, SandboxJobSummary, SettingRow, }; -#[cfg(feature = "postgres")] -pub use store::Store; diff --git a/src/lib.rs b/src/lib.rs index 539d0e5a..8eaefaa9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -43,8 +43,8 @@ pub mod bootstrap; pub mod channels; pub mod cli; pub mod config; -pub mod db; pub mod context; +pub mod db; pub mod error; pub mod estimation; pub mod evaluation; diff --git a/src/main.rs b/src/main.rs index b5f3a954..dafe4113 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,6 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitEx use ironclaw::{ agent::{Agent, AgentDeps, SessionManager}, - pairing::PairingStore, channels::{ ChannelManager, GatewayChannel, HttpChannel, ReplChannel, WebhookServer, WebhookServerConfig, @@ -18,8 +17,7 @@ use ironclaw::{ web::log_layer::{LogBroadcaster, WebLogLayer}, }, cli::{ - Cli, Command, run_mcp_command, run_pairing_command, run_status_command, - run_tool_command, + Cli, Command, run_mcp_command, run_pairing_command, run_status_command, run_tool_command, }, config::Config, context::ContextManager, @@ -29,6 +27,7 @@ use ironclaw::{ ContainerJobConfig, ContainerJobManager, OrchestratorApi, TokenStore, api::OrchestratorState, }, + pairing::PairingStore, safety::SafetyLayer, secrets::SecretsStore, tools::{ @@ -39,11 +38,9 @@ use ironclaw::{ workspace::{EmbeddingProvider, NearAiEmbeddings, OpenAiEmbeddings, Workspace}, }; -use ironclaw::secrets::SecretsCrypto; #[cfg(feature = "postgres")] use ironclaw::secrets::PostgresSecretsStore; -#[cfg(feature = "libsql")] -use ironclaw::secrets::LibSqlSecretsStore; +use ironclaw::secrets::SecretsCrypto; #[cfg(any(feature = "postgres", feature = "libsql"))] use ironclaw::setup::{SetupConfig, SetupWizard}; @@ -92,7 +89,9 @@ async fn main() -> anyhow::Result<()> { // Memory commands need database (and optionally embeddings) let _ = dotenvy::dotenv(); - let config = Config::from_env().await.map_err(|e| anyhow::anyhow!("{}", e))?; + let config = Config::from_env() + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; // Set up embeddings if available let session = ironclaw::llm::create_session_manager(ironclaw::llm::SessionConfig { @@ -138,7 +137,8 @@ async fn main() -> anyhow::Result<()> { .await .map_err(|e| anyhow::anyhow!("{}", e))?; - return ironclaw::cli::run_memory_command_with_db(mem_cmd.clone(), db, embeddings).await; + return ironclaw::cli::run_memory_command_with_db(mem_cmd.clone(), db, embeddings) + .await; } Some(Command::Pairing(pairing_cmd)) => { tracing_subscriber::fmt() @@ -358,16 +358,22 @@ async fn main() -> anyhow::Result<()> { match config.database.backend { #[cfg(feature = "libsql")] ironclaw::config::DatabaseBackend::LibSql => { - use ironclaw::db::libsql_backend::LibSqlBackend; use ironclaw::db::Database as _; + use ironclaw::db::libsql_backend::LibSqlBackend; use secrecy::ExposeSecret as _; let default_path = ironclaw::config::default_libsql_path(); - let db_path = config.database.libsql_path.as_deref() + let db_path = config + .database + .libsql_path + .as_deref() .unwrap_or(&default_path); let backend = if let Some(ref url) = config.database.libsql_url { - let token = config.database.libsql_auth_token.as_ref() + let token = config + .database + .libsql_auth_token + .as_ref() .expect("LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set"); LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret()).await? } else { @@ -384,9 +390,12 @@ async fn main() -> anyhow::Result<()> { #[cfg(feature = "postgres")] _ => { use ironclaw::db::Database as _; - let pg = ironclaw::db::postgres::PgBackend::new(&config.database).await + let pg = ironclaw::db::postgres::PgBackend::new(&config.database) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + pg.run_migrations() + .await .map_err(|e| anyhow::anyhow!("{}", e))?; - pg.run_migrations().await.map_err(|e| anyhow::anyhow!("{}", e))?; tracing::info!("PostgreSQL database connected and migrations applied"); pg_pool = Some(pg.pool()); @@ -394,7 +403,9 @@ async fn main() -> anyhow::Result<()> { } #[cfg(not(feature = "postgres"))] _ => { - anyhow::bail!("No database backend available. Enable 'postgres' or 'libsql' feature."); + anyhow::bail!( + "No database backend available. Enable 'postgres' or 'libsql' feature." + ); } } }; @@ -526,11 +537,11 @@ async fn main() -> anyhow::Result<()> { } #[cfg(all(feature = "libsql", not(feature = "postgres")))] { - if let (Some(conn), Some(master_key)) = (libsql_conn.take(), config.secrets.master_key()) { + if let (Some(conn), Some(master_key)) = + (libsql_conn.take(), config.secrets.master_key()) + { match SecretsCrypto::new(master_key.clone()) { - Ok(crypto) => Some(Arc::new( - LibSqlSecretsStore::new(conn, Arc::new(crypto)), - ) + Ok(crypto) => Some(Arc::new(LibSqlSecretsStore::new(conn, Arc::new(crypto))) as Arc), Err(e) => { tracing::warn!("Failed to initialize secrets crypto: {}", e); diff --git a/src/secrets/store.rs b/src/secrets/store.rs index 34675bb1..04d896c6 100644 --- a/src/secrets/store.rs +++ b/src/secrets/store.rs @@ -340,7 +340,10 @@ impl SecretsStore for LibSqlSecretsStore { .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)); // Start transaction for atomic upsert + read-back - let tx = self.conn.transaction().await + let tx = self + .conn + .transaction() + .await .map_err(|e| SecretError::Database(e.to_string()))?; tx.execute( @@ -390,7 +393,8 @@ impl SecretsStore for LibSqlSecretsStore { let secret = libsql_row_to_secret(&row)?; - tx.commit().await + tx.commit() + .await .map_err(|e| SecretError::Database(e.to_string()))?; Ok(secret) diff --git a/src/setup/channels.rs b/src/setup/channels.rs index 38596527..d8918c7f 100644 --- a/src/setup/channels.rs +++ b/src/setup/channels.rs @@ -12,9 +12,9 @@ use reqwest::Client; use secrecy::{ExposeSecret, SecretString}; use serde::Deserialize; -use crate::secrets::{CreateSecretParams, SecretsStore}; #[cfg(feature = "postgres")] use crate::secrets::SecretsCrypto; +use crate::secrets::{CreateSecretParams, SecretsStore}; use crate::settings::Settings; use crate::setup::prompts::{ confirm, input, optional_input, print_error, print_info, print_success, secret_input, diff --git a/src/setup/wizard.rs b/src/setup/wizard.rs index b0752dc6..6fbb2c4d 100644 --- a/src/setup/wizard.rs +++ b/src/setup/wizard.rs @@ -270,12 +270,17 @@ impl SetupWizard { if let Some(ref path) = existing_path { print_info(&format!("Existing database path: {}", path)); if confirm("Use this database?", true).map_err(SetupError::Io)? { - let turso_url = std::env::var("LIBSQL_URL").ok() + let turso_url = std::env::var("LIBSQL_URL") + .ok() .or_else(|| self.settings.libsql_url.clone()); let turso_token = std::env::var("LIBSQL_AUTH_TOKEN").ok(); match self - .test_database_connection_libsql(path, turso_url.as_deref(), turso_token.as_deref()) + .test_database_connection_libsql( + path, + turso_url.as_deref(), + turso_token.as_deref(), + ) .await { Ok(()) => { @@ -309,8 +314,8 @@ impl SetupWizard { // Ask about Turso cloud sync println!(); - let use_turso = confirm("Enable Turso cloud sync (remote replica)?", false) - .map_err(SetupError::Io)?; + let use_turso = + confirm("Enable Turso cloud sync (remote replica)?", false).map_err(SetupError::Io)?; let (turso_url, turso_token) = if use_turso { print_info("Enter your Turso database URL and auth token."); @@ -349,11 +354,9 @@ impl SetupWizard { if let Some(url) = turso_url { self.settings.libsql_url = Some(url); } - return Ok(()); - } - Err(e) => { - return Err(SetupError::Database(format!("Connection failed: {}", e))); + Ok(()) } + Err(e) => Err(SetupError::Database(format!("Connection failed: {}", e))), } } @@ -786,9 +789,10 @@ impl SetupWizard { } }; - let store: Arc = Arc::new( - crate::secrets::PostgresSecretsStore::new(pool, Arc::clone(crypto)), - ); + let store: Arc = Arc::new(crate::secrets::PostgresSecretsStore::new( + pool, + Arc::clone(crypto), + )); Ok(Some(store)) } @@ -802,9 +806,10 @@ impl SetupWizard { let conn = backend .connect() .map_err(|e| SetupError::Database(format!("Failed to create connection: {}", e)))?; - let store: Arc = Arc::new( - crate::secrets::LibSqlSecretsStore::new(conn, Arc::clone(crypto)), - ); + let store: Arc = Arc::new(crate::secrets::LibSqlSecretsStore::new( + conn, + Arc::clone(crypto), + )); Ok(Some(store)) } else { Ok(None) diff --git a/src/tools/registry.rs b/src/tools/registry.rs index 6b5ca651..8e7c1904 100644 --- a/src/tools/registry.rs +++ b/src/tools/registry.rs @@ -6,8 +6,8 @@ use std::sync::Arc; use tokio::sync::RwLock; use crate::context::ContextManager; -use crate::extensions::ExtensionManager; use crate::db::Database; +use crate::extensions::ExtensionManager; use crate::llm::{LlmProvider, ToolDefinition}; use crate::orchestrator::job_manager::ContainerJobManager; use crate::safety::SafetyLayer; diff --git a/src/tools/wasm/mod.rs b/src/tools/wasm/mod.rs index 6166fd20..c80c511e 100644 --- a/src/tools/wasm/mod.rs +++ b/src/tools/wasm/mod.rs @@ -113,9 +113,8 @@ pub use storage::LibSqlWasmToolStore; #[cfg(feature = "postgres")] pub use storage::PostgresWasmToolStore; pub use storage::{ - StoreToolParams, StoredCapabilities, StoredWasmTool, - StoredWasmToolWithBinary, ToolStatus, TrustLevel, WasmStorageError, WasmToolStore, - compute_binary_hash, verify_binary_integrity, + StoreToolParams, StoredCapabilities, StoredWasmTool, StoredWasmToolWithBinary, ToolStatus, + TrustLevel, WasmStorageError, WasmToolStore, compute_binary_hash, verify_binary_integrity, }; // Loader diff --git a/src/tools/wasm/storage.rs b/src/tools/wasm/storage.rs index 9ee8d333..0b99bd9f 100644 --- a/src/tools/wasm/storage.rs +++ b/src/tools/wasm/storage.rs @@ -643,7 +643,9 @@ impl WasmToolStore for LibSqlWasmToolStore { .next() .await .map_err(|e| WasmStorageError::Database(e.to_string()))? - .ok_or_else(|| WasmStorageError::Database("Insert succeeded but row not found".into()))?; + .ok_or_else(|| { + WasmStorageError::Database("Insert succeeded but row not found".into()) + })?; libsql_row_to_tool(&row) } diff --git a/src/workspace/mod.rs b/src/workspace/mod.rs index b2eb4ca1..9da78bb5 100644 --- a/src/workspace/mod.rs +++ b/src/workspace/mod.rs @@ -433,7 +433,9 @@ impl Workspace { /// List all files recursively (flat list of all paths). pub async fn list_all(&self) -> Result, WorkspaceError> { - self.storage.list_all_paths(&self.user_id, self.agent_id).await + self.storage + .list_all_paths(&self.user_id, self.agent_id) + .await } // ==================== Convenience Methods ====================