mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-09-01 00:59:33 +00:00
refactor: architecture improvements for contributor velocity (#198)
* refactor: split large files and consolidate test stubs for contributor velocity - Extract 7 Database sub-traits (ConversationStore, JobStore, SandboxStore, RoutineStore, ToolFailureStore, SettingsStore, WorkspaceStore) with Database as a supertrait combining them all - Split libsql_backend.rs (2769 lines) into src/db/libsql/ directory with one file per sub-trait implementation - Split config.rs (1753 lines) into src/config/ directory with 16 domain files - Consolidate 3 duplicate test LLM stubs into shared StubLlm in src/testing.rs - Split server.rs handlers into src/channels/web/handlers/ directory - Extract main.rs init phases into AppBuilder (src/app.rs) - Add developer setup script (scripts/dev-setup.sh) Co-Authored-By: Claude Opus 4.6 <[email protected]> * refactor: move heartbeat test from examples/ to tests/ Convert standalone example binary into a proper #[ignore] integration test, matching the convention of the other integration tests. Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: fix rustfmt formatting for CI Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review comments from Copilot - tunnel.rs: replace .ok().flatten() with ? to propagate env var errors - secrets.rs: remove misleading "process-wide cache" comment - database.rs: use uppercase "DATABASE_URL" in error key - testing.rs: gate harness tests with #[cfg(feature = "libsql")] Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Illia Polosukhin <[email protected]> Co-authored-by: Claude Opus 4.6 <[email protected]>
This commit is contained in:
Illia Polosukhin
co-authored by
Illia Polosukhin
Claude Opus 4.6
Illia Polosukhin
Claude Opus 4.6
parent
6330f1b27a
commit
ffb1cc9be8
@@ -0,0 +1,354 @@
|
||||
//! Conversation-related ConversationStore implementation for LibSqlBackend.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use chrono::{DateTime, Utc};
|
||||
use libsql::params;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{LibSqlBackend, fmt_ts, get_i64, get_json, get_opt_text, get_text, get_ts, opt_text};
|
||||
use crate::db::ConversationStore;
|
||||
use crate::error::DatabaseError;
|
||||
use crate::history::{ConversationMessage, ConversationSummary};
|
||||
|
||||
#[async_trait]
|
||||
impl ConversationStore for LibSqlBackend {
|
||||
async fn create_conversation(
|
||||
&self,
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn touch_conversation(&self, id: Uuid) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn add_conversation_message(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
role: &str,
|
||||
content: &str,
|
||||
) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let id = Uuid::new_v4();
|
||||
conn.execute(
|
||||
"INSERT INTO conversation_messages (id, conversation_id, role, content) VALUES (?1, ?2, ?3, ?4)",
|
||||
params![id.to_string(), conversation_id.to_string(), role, content],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
self.touch_conversation(conversation_id).await?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn ensure_conversation(
|
||||
&self,
|
||||
id: Uuid,
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute(
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_conversations_with_preview(
|
||||
&self,
|
||||
user_id: &str,
|
||||
channel: &str,
|
||||
limit: i64,
|
||||
) -> Result<Vec<ConversationSummary>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT
|
||||
c.id,
|
||||
c.started_at,
|
||||
c.last_activity,
|
||||
c.metadata,
|
||||
(SELECT COUNT(*) FROM conversation_messages m WHERE m.conversation_id = c.id) AS message_count,
|
||||
(SELECT substr(m2.content, 1, 100)
|
||||
FROM conversation_messages m2
|
||||
WHERE m2.conversation_id = c.id AND m2.role = 'user'
|
||||
ORDER BY m2.created_at ASC
|
||||
LIMIT 1
|
||||
) AS title
|
||||
FROM conversations c
|
||||
WHERE c.user_id = ?1 AND c.channel = ?2
|
||||
ORDER BY c.last_activity DESC
|
||||
LIMIT ?3
|
||||
"#,
|
||||
params![user_id, channel, limit],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
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::<String>(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),
|
||||
title: get_opt_text(&row, 5),
|
||||
thread_type,
|
||||
});
|
||||
}
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
async fn get_or_create_assistant_conversation(
|
||||
&self,
|
||||
user_id: &str,
|
||||
channel: &str,
|
||||
) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
// Try to find existing
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id FROM conversations
|
||||
WHERE user_id = ?1 AND channel = ?2
|
||||
AND json_extract(metadata, '$.thread_type') = 'assistant'
|
||||
LIMIT 1
|
||||
"#,
|
||||
params![user_id, channel],
|
||||
)
|
||||
.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()
|
||||
.map_err(|_| DatabaseError::Serialization("Invalid UUID".to_string()));
|
||||
}
|
||||
|
||||
// Create new
|
||||
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()))?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn create_conversation_with_metadata(
|
||||
&self,
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
metadata: &serde_json::Value,
|
||||
) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn list_conversation_messages_paginated(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
before: Option<DateTime<Utc>>,
|
||||
limit: i64,
|
||||
) -> Result<(Vec<ConversationMessage>, bool), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let fetch_limit = limit + 1;
|
||||
let cid = conversation_id.to_string();
|
||||
|
||||
let mut rows = if let Some(before_ts) = before {
|
||||
conn.query(
|
||||
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
|
||||
} else {
|
||||
conn.query(
|
||||
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
|
||||
}
|
||||
.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()))?
|
||||
{
|
||||
all.push(ConversationMessage {
|
||||
id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
role: get_text(&row, 1),
|
||||
content: get_text(&row, 2),
|
||||
created_at: get_ts(&row, 3),
|
||||
});
|
||||
}
|
||||
|
||||
let has_more = all.len() as i64 > limit;
|
||||
all.truncate(limit as usize);
|
||||
all.reverse(); // oldest first
|
||||
Ok((all, has_more))
|
||||
}
|
||||
|
||||
async fn update_conversation_metadata_field(
|
||||
&self,
|
||||
id: Uuid,
|
||||
key: &str,
|
||||
value: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
// 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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_conversation_metadata(
|
||||
&self,
|
||||
id: Uuid,
|
||||
) -> Result<Option<serde_json::Value>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT metadata FROM conversations WHERE id = ?1",
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(Some(get_json(&row, 0))),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_conversation_messages(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Vec<ConversationMessage>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, role, content, created_at
|
||||
FROM conversation_messages
|
||||
WHERE conversation_id = ?1
|
||||
ORDER BY created_at ASC
|
||||
"#,
|
||||
params![conversation_id.to_string()],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
messages.push(ConversationMessage {
|
||||
id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
role: get_text(&row, 1),
|
||||
content: get_text(&row, 2),
|
||||
created_at: get_ts(&row, 3),
|
||||
});
|
||||
}
|
||||
Ok(messages)
|
||||
}
|
||||
|
||||
async fn conversation_belongs_to_user(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT 1 FROM conversations WHERE id = ?1 AND user_id = ?2",
|
||||
libsql::params![conversation_id.to_string(), user_id],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
let found = rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,330 @@
|
||||
//! Job-related JobStore implementation for LibSqlBackend.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use libsql::params;
|
||||
use rust_decimal::Decimal;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
LibSqlBackend, fmt_opt_ts, fmt_ts, get_decimal, get_i64, get_json, get_opt_decimal,
|
||||
get_opt_text, get_opt_ts, get_text, get_ts, opt_text, opt_text_owned, parse_job_state,
|
||||
};
|
||||
use crate::context::{ActionRecord, JobContext, JobState};
|
||||
use crate::db::JobStore;
|
||||
use crate::error::DatabaseError;
|
||||
use crate::history::LlmCallRecord;
|
||||
|
||||
use chrono::Utc;
|
||||
|
||||
#[async_trait]
|
||||
impl JobStore for LibSqlBackend {
|
||||
async fn save_job(&self, ctx: &JobContext) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let status = ctx.state.to_string();
|
||||
let estimated_time_secs = ctx.estimated_duration.map(|d| d.as_secs() as i64);
|
||||
|
||||
conn
|
||||
.execute(
|
||||
r#"
|
||||
INSERT INTO agent_jobs (
|
||||
id, conversation_id, title, description, category, status, source,
|
||||
budget_amount, budget_token, bid_amount, estimated_cost, estimated_time_secs,
|
||||
actual_cost, repair_attempts, created_at, started_at, completed_at
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
title = excluded.title,
|
||||
description = excluded.description,
|
||||
category = excluded.category,
|
||||
status = excluded.status,
|
||||
estimated_cost = excluded.estimated_cost,
|
||||
estimated_time_secs = excluded.estimated_time_secs,
|
||||
actual_cost = excluded.actual_cost,
|
||||
repair_attempts = excluded.repair_attempts,
|
||||
started_at = excluded.started_at,
|
||||
completed_at = excluded.completed_at
|
||||
"#,
|
||||
params![
|
||||
ctx.job_id.to_string(),
|
||||
opt_text_owned(ctx.conversation_id.map(|id| id.to_string())),
|
||||
ctx.title.as_str(),
|
||||
ctx.description.as_str(),
|
||||
opt_text(ctx.category.as_deref()),
|
||||
status,
|
||||
"direct",
|
||||
opt_text_owned(ctx.budget.map(|d| d.to_string())),
|
||||
opt_text(ctx.budget_token.as_deref()),
|
||||
opt_text_owned(ctx.bid_amount.map(|d| d.to_string())),
|
||||
opt_text_owned(ctx.estimated_cost.map(|d| d.to_string())),
|
||||
estimated_time_secs,
|
||||
ctx.actual_cost.to_string(),
|
||||
ctx.repair_attempts as i64,
|
||||
fmt_ts(&ctx.created_at),
|
||||
fmt_opt_ts(&ctx.started_at),
|
||||
fmt_opt_ts(&ctx.completed_at),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_job(&self, id: Uuid) -> Result<Option<JobContext>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, conversation_id, title, description, category, status, user_id,
|
||||
budget_amount, budget_token, bid_amount, estimated_cost, estimated_time_secs,
|
||||
actual_cost, repair_attempts, created_at, started_at, completed_at
|
||||
FROM agent_jobs WHERE id = ?1
|
||||
"#,
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => {
|
||||
let status_str = get_text(&row, 5);
|
||||
let state = parse_job_state(&status_str);
|
||||
let estimated_time_secs: Option<i64> = row.get::<i64>(11).ok();
|
||||
|
||||
Ok(Some(JobContext {
|
||||
job_id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
state,
|
||||
user_id: get_text(&row, 6),
|
||||
conversation_id: get_opt_text(&row, 1).and_then(|s| s.parse().ok()),
|
||||
title: get_text(&row, 2),
|
||||
description: get_text(&row, 3),
|
||||
category: get_opt_text(&row, 4),
|
||||
budget: get_opt_decimal(&row, 7),
|
||||
budget_token: get_opt_text(&row, 8),
|
||||
bid_amount: get_opt_decimal(&row, 9),
|
||||
estimated_cost: get_opt_decimal(&row, 10),
|
||||
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),
|
||||
completed_at: get_opt_ts(&row, 16),
|
||||
transitions: Vec::new(),
|
||||
metadata: serde_json::Value::Null,
|
||||
extra_env: std::sync::Arc::new(std::collections::HashMap::new()),
|
||||
}))
|
||||
}
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_job_status(
|
||||
&self,
|
||||
id: Uuid,
|
||||
status: JobState,
|
||||
failure_reason: Option<&str>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_stuck_jobs(&self) -> Result<Vec<Uuid>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query("SELECT id FROM agent_jobs WHERE status = 'stuck'", ())
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
if let Ok(id_str) = row.get::<String>(0)
|
||||
&& let Ok(id) = id_str.parse()
|
||||
{
|
||||
ids.push(id);
|
||||
}
|
||||
}
|
||||
Ok(ids)
|
||||
}
|
||||
|
||||
async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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#"
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_job_actions(&self, job_id: Uuid) -> Result<Vec<ActionRecord>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, sequence_num, tool_name, input, output_raw, output_sanitized,
|
||||
sanitization_warnings, cost, duration_ms, success, error_message, created_at
|
||||
FROM job_actions WHERE job_id = ?1 ORDER BY sequence_num
|
||||
"#,
|
||||
params![job_id.to_string()],
|
||||
)
|
||||
.await
|
||||
.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<String> =
|
||||
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()),
|
||||
sanitization_warnings: warnings,
|
||||
cost: get_opt_decimal(&row, 7),
|
||||
duration: std::time::Duration::from_millis(get_i64(&row, 8) as u64),
|
||||
success: get_i64(&row, 9) != 0,
|
||||
error: get_opt_text(&row, 10),
|
||||
executed_at: get_ts(&row, 11),
|
||||
});
|
||||
}
|
||||
Ok(actions)
|
||||
}
|
||||
|
||||
async fn record_llm_call(&self, record: &LlmCallRecord<'_>) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let id = Uuid::new_v4();
|
||||
conn.execute(
|
||||
r#"
|
||||
INSERT INTO llm_calls (id, job_id, conversation_id, provider, model, input_tokens, output_tokens, cost, purpose)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
|
||||
"#,
|
||||
params![
|
||||
id.to_string(),
|
||||
opt_text_owned(record.job_id.map(|id| id.to_string())),
|
||||
opt_text_owned(record.conversation_id.map(|id| id.to_string())),
|
||||
record.provider,
|
||||
record.model,
|
||||
record.input_tokens as i64,
|
||||
record.output_tokens as i64,
|
||||
record.cost.to_string(),
|
||||
opt_text(record.purpose),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn save_estimation_snapshot(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
category: &str,
|
||||
tool_names: &[String],
|
||||
estimated_cost: Decimal,
|
||||
estimated_time_secs: i32,
|
||||
estimated_value: Decimal,
|
||||
) -> Result<Uuid, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let id = Uuid::new_v4();
|
||||
let tools_json = serde_json::to_string(tool_names)
|
||||
.map_err(|e| DatabaseError::Serialization(e.to_string()))?;
|
||||
|
||||
conn.execute(
|
||||
r#"
|
||||
INSERT INTO estimation_snapshots (id, job_id, category, tool_names, estimated_cost, estimated_time_secs, estimated_value)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
|
||||
"#,
|
||||
params![
|
||||
id.to_string(),
|
||||
job_id.to_string(),
|
||||
category,
|
||||
tools_json,
|
||||
estimated_cost.to_string(),
|
||||
estimated_time_secs as i64,
|
||||
estimated_value.to_string(),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn update_estimation_actuals(
|
||||
&self,
|
||||
id: Uuid,
|
||||
actual_cost: Decimal,
|
||||
actual_time_secs: i32,
|
||||
actual_value: Option<Decimal>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
conn.execute(
|
||||
"UPDATE estimation_snapshots SET actual_cost = ?2, actual_time_secs = ?3, actual_value = ?4 WHERE id = ?1",
|
||||
params![
|
||||
id.to_string(),
|
||||
actual_cost.to_string(),
|
||||
actual_time_secs as i64,
|
||||
actual_value.map(|d| d.to_string()).unwrap_or_default(),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,460 @@
|
||||
//! libSQL/Turso backend for the Database trait.
|
||||
//!
|
||||
//! Provides an embedded SQLite-compatible database using Turso's libSQL fork.
|
||||
//! Supports three modes:
|
||||
//! - Local embedded (file-based, no server needed)
|
||||
//! - Turso cloud with embedded replica (sync to cloud)
|
||||
//! - In-memory (for testing)
|
||||
|
||||
mod conversations;
|
||||
mod jobs;
|
||||
mod routines;
|
||||
mod sandbox;
|
||||
mod settings;
|
||||
mod tool_failures;
|
||||
mod workspace;
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use chrono::{DateTime, NaiveDateTime, Utc};
|
||||
use libsql::{Connection, Database as LibSqlDatabase};
|
||||
use rust_decimal::Decimal;
|
||||
|
||||
use crate::agent::routine::{
|
||||
NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger,
|
||||
};
|
||||
use crate::context::JobState;
|
||||
use crate::db::Database;
|
||||
use crate::error::DatabaseError;
|
||||
use crate::workspace::MemoryDocument;
|
||||
|
||||
use crate::db::libsql_migrations;
|
||||
|
||||
/// Explicit column list for routines table (matches positional access in `row_to_routine_libsql`).
|
||||
pub(crate) const ROUTINE_COLUMNS: &str = "\
|
||||
id, name, description, user_id, enabled, \
|
||||
trigger_type, trigger_config, action_type, action_config, \
|
||||
cooldown_secs, max_concurrent, dedup_window_secs, \
|
||||
notify_channel, notify_user, notify_on_success, notify_on_failure, notify_on_attention, \
|
||||
state, last_run_at, next_fire_at, run_count, consecutive_failures, \
|
||||
created_at, updated_at";
|
||||
|
||||
/// Explicit column list for routine_runs table (matches positional access in `row_to_routine_run_libsql`).
|
||||
pub(crate) const ROUTINE_RUN_COLUMNS: &str = "\
|
||||
id, routine_id, trigger_type, trigger_detail, started_at, \
|
||||
status, completed_at, result_summary, tokens_used, job_id, created_at";
|
||||
|
||||
/// libSQL/Turso database backend.
|
||||
///
|
||||
/// Stores the `Database` handle in an `Arc` so that the same underlying
|
||||
/// database can be shared with stores (SecretsStore, WasmToolStore) that
|
||||
/// create their own connections per-operation.
|
||||
pub struct LibSqlBackend {
|
||||
db: Arc<LibSqlDatabase>,
|
||||
}
|
||||
|
||||
impl LibSqlBackend {
|
||||
/// Create a new local embedded database.
|
||||
pub async fn new_local(path: &Path) -> Result<Self, DatabaseError> {
|
||||
// Ensure parent directory exists
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(|e| {
|
||||
DatabaseError::Pool(format!("Failed to create database directory: {}", e))
|
||||
})?;
|
||||
}
|
||||
|
||||
let db = libsql::Builder::new_local(path)
|
||||
.build()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Pool(format!("Failed to open libSQL database: {}", e)))?;
|
||||
|
||||
Ok(Self { db: Arc::new(db) })
|
||||
}
|
||||
|
||||
/// Create a new in-memory database (for testing).
|
||||
pub async fn new_memory() -> Result<Self, DatabaseError> {
|
||||
let db = libsql::Builder::new_local(":memory:")
|
||||
.build()
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DatabaseError::Pool(format!("Failed to create in-memory database: {}", e))
|
||||
})?;
|
||||
|
||||
Ok(Self { db: Arc::new(db) })
|
||||
}
|
||||
|
||||
/// Create with Turso cloud sync (embedded replica).
|
||||
pub async fn new_remote_replica(
|
||||
path: &Path,
|
||||
url: &str,
|
||||
auth_token: &str,
|
||||
) -> Result<Self, DatabaseError> {
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(|e| {
|
||||
DatabaseError::Pool(format!("Failed to create database directory: {}", e))
|
||||
})?;
|
||||
}
|
||||
|
||||
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)))?;
|
||||
|
||||
Ok(Self { db: Arc::new(db) })
|
||||
}
|
||||
|
||||
/// Get a shared reference to the underlying database handle.
|
||||
///
|
||||
/// Use this to pass the database to stores (SecretsStore, WasmToolStore)
|
||||
/// that need to create their own connections per-operation.
|
||||
pub fn shared_db(&self) -> Arc<LibSqlDatabase> {
|
||||
Arc::clone(&self.db)
|
||||
}
|
||||
|
||||
/// Create a new connection to the database.
|
||||
///
|
||||
/// Sets `PRAGMA busy_timeout = 5000` on every connection so concurrent
|
||||
/// writers wait up to 5 seconds instead of failing instantly with
|
||||
/// "database is locked".
|
||||
pub async fn connect(&self) -> Result<Connection, DatabaseError> {
|
||||
let conn = self
|
||||
.db
|
||||
.connect()
|
||||
.map_err(|e| DatabaseError::Pool(format!("Failed to create connection: {}", e)))?;
|
||||
conn.query("PRAGMA busy_timeout = 5000", ())
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Pool(format!("Failed to set busy_timeout: {}", e)))?;
|
||||
Ok(conn)
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Helper functions ====================
|
||||
|
||||
/// Parse an ISO-8601 timestamp string from SQLite into DateTime<Utc>.
|
||||
///
|
||||
/// Tries multiple formats in order:
|
||||
/// 1. RFC 3339 with timezone (e.g. `2024-01-15T10:30:00.123Z`)
|
||||
/// 2. Naive datetime with fractional seconds (e.g. `2024-01-15 10:30:00.123`)
|
||||
/// 3. Naive datetime without fractional seconds (e.g. `2024-01-15 10:30:00`)
|
||||
///
|
||||
/// Returns an error if none of the formats match.
|
||||
pub(crate) fn parse_timestamp(s: &str) -> Result<DateTime<Utc>, String> {
|
||||
// RFC 3339 (our canonical write format)
|
||||
if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
|
||||
return Ok(dt.with_timezone(&Utc));
|
||||
}
|
||||
// Naive with fractional seconds (legacy or SQLite datetime() output)
|
||||
if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f") {
|
||||
return Ok(ndt.and_utc());
|
||||
}
|
||||
// Naive without fractional seconds (legacy format)
|
||||
if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
|
||||
return Ok(ndt.and_utc());
|
||||
}
|
||||
Err(format!("unparseable timestamp: {:?}", s))
|
||||
}
|
||||
|
||||
/// Format a DateTime<Utc> for SQLite storage (RFC 3339 with millisecond precision).
|
||||
pub(crate) fn fmt_ts(dt: &DateTime<Utc>) -> String {
|
||||
dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
|
||||
}
|
||||
|
||||
/// Format an optional DateTime<Utc>.
|
||||
pub(crate) fn fmt_opt_ts(dt: &Option<DateTime<Utc>>) -> libsql::Value {
|
||||
match dt {
|
||||
Some(dt) => libsql::Value::Text(fmt_ts(dt)),
|
||||
None => libsql::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn parse_job_state(s: &str) -> JobState {
|
||||
match s {
|
||||
"pending" => JobState::Pending,
|
||||
"in_progress" => JobState::InProgress,
|
||||
"completed" => JobState::Completed,
|
||||
"submitted" => JobState::Submitted,
|
||||
"accepted" => JobState::Accepted,
|
||||
"failed" => JobState::Failed,
|
||||
"stuck" => JobState::Stuck,
|
||||
"cancelled" => JobState::Cancelled,
|
||||
_ => JobState::Pending,
|
||||
}
|
||||
}
|
||||
|
||||
/// Extract a text column from a libsql Row, returning empty string for NULL.
|
||||
pub(crate) fn get_text(row: &libsql::Row, idx: i32) -> String {
|
||||
row.get::<String>(idx).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Extract an optional text column.
|
||||
/// Returns None for SQL NULL, preserves empty strings as Some("").
|
||||
pub(crate) fn get_opt_text(row: &libsql::Row, idx: i32) -> Option<String> {
|
||||
row.get::<String>(idx).ok()
|
||||
}
|
||||
|
||||
/// Convert an `Option<&str>` to a `libsql::Value` (Text or Null).
|
||||
/// Use this instead of `.unwrap_or("")` to preserve NULL semantics.
|
||||
pub(crate) fn opt_text(s: Option<&str>) -> libsql::Value {
|
||||
match s {
|
||||
Some(s) => libsql::Value::Text(s.to_string()),
|
||||
None => libsql::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert an `Option<String>` to a `libsql::Value` (Text or Null).
|
||||
pub(crate) fn opt_text_owned(s: Option<String>) -> libsql::Value {
|
||||
match s {
|
||||
Some(s) => libsql::Value::Text(s),
|
||||
None => libsql::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
/// Extract an i64 column, defaulting to 0.
|
||||
pub(crate) fn get_i64(row: &libsql::Row, idx: i32) -> i64 {
|
||||
row.get::<i64>(idx).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Extract an optional bool from an integer column.
|
||||
pub(crate) fn get_opt_bool(row: &libsql::Row, idx: i32) -> Option<bool> {
|
||||
row.get::<i64>(idx).ok().map(|v| v != 0)
|
||||
}
|
||||
|
||||
/// Parse a Decimal from a text column.
|
||||
pub(crate) fn get_decimal(row: &libsql::Row, idx: i32) -> Decimal {
|
||||
row.get::<String>(idx)
|
||||
.ok()
|
||||
.and_then(|s| s.parse::<Decimal>().ok())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Parse an optional Decimal from a text column.
|
||||
pub(crate) fn get_opt_decimal(row: &libsql::Row, idx: i32) -> Option<Decimal> {
|
||||
row.get::<String>(idx)
|
||||
.ok()
|
||||
.and_then(|s| s.parse::<Decimal>().ok())
|
||||
}
|
||||
|
||||
/// Parse a JSON value from a text column.
|
||||
pub(crate) fn get_json(row: &libsql::Row, idx: i32) -> serde_json::Value {
|
||||
row.get::<String>(idx)
|
||||
.ok()
|
||||
.and_then(|s| serde_json::from_str(&s).ok())
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}
|
||||
|
||||
/// Parse a timestamp from a text column.
|
||||
///
|
||||
/// If the column is NULL or the value cannot be parsed, logs a warning and
|
||||
/// returns the Unix epoch (1970-01-01T00:00:00Z) so the error is detectable
|
||||
/// rather than silently replaced by the current time.
|
||||
pub(crate) fn get_ts(row: &libsql::Row, idx: i32) -> DateTime<Utc> {
|
||||
match row.get::<String>(idx) {
|
||||
Ok(s) => match parse_timestamp(&s) {
|
||||
Ok(dt) => dt,
|
||||
Err(e) => {
|
||||
tracing::warn!("Timestamp parse failure at column {}: {}", idx, e);
|
||||
DateTime::UNIX_EPOCH
|
||||
}
|
||||
},
|
||||
Err(_) => DateTime::UNIX_EPOCH,
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse an optional timestamp from a text column.
|
||||
///
|
||||
/// Returns None if the column is NULL. Logs a warning and returns None if the
|
||||
/// value is present but cannot be parsed.
|
||||
pub(crate) fn get_opt_ts(row: &libsql::Row, idx: i32) -> Option<DateTime<Utc>> {
|
||||
match row.get::<String>(idx) {
|
||||
Ok(s) if s.is_empty() => None,
|
||||
Ok(s) => match parse_timestamp(&s) {
|
||||
Ok(dt) => Some(dt),
|
||||
Err(e) => {
|
||||
tracing::warn!("Timestamp parse failure at column {}: {}", idx, e);
|
||||
None
|
||||
}
|
||||
},
|
||||
Err(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Database for LibSqlBackend {
|
||||
async fn run_migrations(&self) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
// WAL mode persists in the database file: all future connections benefit.
|
||||
// Readers no longer block writers and vice versa.
|
||||
conn.query("PRAGMA journal_mode=WAL", ())
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Migration(format!("Failed to enable WAL mode: {}", e)))?;
|
||||
conn.execute_batch(libsql_migrations::SCHEMA)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Migration(format!("libSQL migration failed: {}", e)))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Row conversion helpers ====================
|
||||
|
||||
pub(crate) fn row_to_memory_document(row: &libsql::Row) -> MemoryDocument {
|
||||
MemoryDocument {
|
||||
id: get_text(row, 0).parse().unwrap_or_default(),
|
||||
user_id: get_text(row, 1),
|
||||
agent_id: get_opt_text(row, 2).and_then(|s| s.parse().ok()),
|
||||
path: get_text(row, 3),
|
||||
content: get_text(row, 4),
|
||||
created_at: get_ts(row, 5),
|
||||
updated_at: get_ts(row, 6),
|
||||
metadata: get_json(row, 7),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn row_to_routine_libsql(row: &libsql::Row) -> Result<Routine, DatabaseError> {
|
||||
let trigger_type = get_text(row, 5);
|
||||
let trigger_config = get_json(row, 6);
|
||||
let action_type = get_text(row, 7);
|
||||
let action_config = get_json(row, 8);
|
||||
let cooldown_secs = get_i64(row, 9);
|
||||
let max_concurrent = get_i64(row, 10);
|
||||
let dedup_window_secs: Option<i64> = row.get::<i64>(11).ok();
|
||||
|
||||
let trigger =
|
||||
Trigger::from_db(&trigger_type, trigger_config).map_err(DatabaseError::Serialization)?;
|
||||
let action = RoutineAction::from_db(&action_type, action_config)
|
||||
.map_err(DatabaseError::Serialization)?;
|
||||
|
||||
Ok(Routine {
|
||||
id: get_text(row, 0).parse().unwrap_or_default(),
|
||||
name: get_text(row, 1),
|
||||
description: get_text(row, 2),
|
||||
user_id: get_text(row, 3),
|
||||
enabled: get_i64(row, 4) != 0,
|
||||
trigger,
|
||||
action,
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(cooldown_secs as u64),
|
||||
max_concurrent: max_concurrent as u32,
|
||||
dedup_window: dedup_window_secs.map(|s| std::time::Duration::from_secs(s as u64)),
|
||||
},
|
||||
notify: NotifyConfig {
|
||||
channel: get_opt_text(row, 12),
|
||||
user: get_text(row, 13),
|
||||
on_success: get_i64(row, 14) != 0,
|
||||
on_failure: get_i64(row, 15) != 0,
|
||||
on_attention: get_i64(row, 16) != 0,
|
||||
},
|
||||
state: get_json(row, 17),
|
||||
last_run_at: get_opt_ts(row, 18),
|
||||
next_fire_at: get_opt_ts(row, 19),
|
||||
run_count: get_i64(row, 20) as u64,
|
||||
consecutive_failures: get_i64(row, 21) as u32,
|
||||
created_at: get_ts(row, 22),
|
||||
updated_at: get_ts(row, 23),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn row_to_routine_run_libsql(row: &libsql::Row) -> Result<RoutineRun, DatabaseError> {
|
||||
let status_str = get_text(row, 5);
|
||||
let status: RunStatus = status_str
|
||||
.parse()
|
||||
.map_err(|e: String| DatabaseError::Serialization(e))?;
|
||||
|
||||
Ok(RoutineRun {
|
||||
id: get_text(row, 0).parse().unwrap_or_default(),
|
||||
routine_id: get_text(row, 1).parse().unwrap_or_default(),
|
||||
trigger_type: get_text(row, 2),
|
||||
trigger_detail: get_opt_text(row, 3),
|
||||
started_at: get_ts(row, 4),
|
||||
completed_at: get_opt_ts(row, 6),
|
||||
status,
|
||||
result_summary: get_opt_text(row, 7),
|
||||
tokens_used: row.get::<i64>(8).ok().map(|v| v as i32),
|
||||
job_id: get_opt_text(row, 9).and_then(|s| s.parse().ok()),
|
||||
created_at: get_ts(row, 10),
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::db::Database;
|
||||
use crate::db::libsql::LibSqlBackend;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_wal_mode_after_migrations() {
|
||||
let backend = LibSqlBackend::new_memory().await.unwrap();
|
||||
backend.run_migrations().await.unwrap();
|
||||
|
||||
let conn = backend.connect().await.unwrap();
|
||||
let mut rows = conn.query("PRAGMA journal_mode", ()).await.unwrap();
|
||||
let row = rows.next().await.unwrap().unwrap();
|
||||
let mode: String = row.get(0).unwrap();
|
||||
// In-memory databases use "memory" journal mode (WAL doesn't apply to :memory:),
|
||||
// but the PRAGMA still executes without error. For file-based databases it returns "wal".
|
||||
assert!(
|
||||
mode == "wal" || mode == "memory",
|
||||
"expected wal or memory, got: {}",
|
||||
mode,
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_busy_timeout_set_on_connect() {
|
||||
let backend = LibSqlBackend::new_memory().await.unwrap();
|
||||
backend.run_migrations().await.unwrap();
|
||||
|
||||
let conn = backend.connect().await.unwrap();
|
||||
let mut rows = conn.query("PRAGMA busy_timeout", ()).await.unwrap();
|
||||
let row = rows.next().await.unwrap().unwrap();
|
||||
let timeout: i64 = row.get(0).unwrap();
|
||||
assert_eq!(timeout, 5000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_concurrent_writes_succeed() {
|
||||
// Use a temp file so connections share state (in-memory DBs are connection-local)
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let db_path = dir.path().join("test_concurrent.db");
|
||||
let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
|
||||
backend.run_migrations().await.unwrap();
|
||||
|
||||
// Spawn 20 concurrent inserts into the conversations table
|
||||
let mut handles = Vec::new();
|
||||
for i in 0..20 {
|
||||
let conn = backend.connect().await.unwrap();
|
||||
let handle = tokio::spawn(async move {
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let val = format!("ch_{}", i);
|
||||
conn.execute(
|
||||
"INSERT INTO conversations (id, channel, user_id) VALUES (?1, ?2, ?3)",
|
||||
libsql::params![id, val, "test_user"],
|
||||
)
|
||||
.await
|
||||
});
|
||||
handles.push(handle);
|
||||
}
|
||||
|
||||
for handle in handles {
|
||||
let result = handle.await.unwrap();
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"concurrent write failed: {:?}",
|
||||
result.err()
|
||||
);
|
||||
}
|
||||
|
||||
// Verify all 20 rows landed
|
||||
let conn = backend.connect().await.unwrap();
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT COUNT(*) FROM conversations WHERE user_id = ?1",
|
||||
libsql::params!["test_user"],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let row = rows.next().await.unwrap().unwrap();
|
||||
let count: i64 = row.get(0).unwrap();
|
||||
assert_eq!(count, 20);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,390 @@
|
||||
//! Routine-related RoutineStore implementation for LibSqlBackend.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use chrono::{DateTime, Utc};
|
||||
use libsql::params;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
LibSqlBackend, ROUTINE_COLUMNS, ROUTINE_RUN_COLUMNS, fmt_opt_ts, fmt_ts, get_i64, opt_text,
|
||||
opt_text_owned, row_to_routine_libsql, row_to_routine_run_libsql,
|
||||
};
|
||||
use crate::agent::routine::{Routine, RoutineRun, RunStatus};
|
||||
use crate::db::RoutineStore;
|
||||
use crate::error::DatabaseError;
|
||||
|
||||
#[async_trait]
|
||||
impl RoutineStore for LibSqlBackend {
|
||||
async fn create_routine(&self, routine: &Routine) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let trigger_type = routine.trigger.type_tag();
|
||||
let trigger_config = routine.trigger.to_config_json();
|
||||
let action_type = routine.action.type_tag();
|
||||
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);
|
||||
|
||||
conn.execute(
|
||||
r#"
|
||||
INSERT INTO routines (
|
||||
id, name, description, user_id, enabled,
|
||||
trigger_type, trigger_config, action_type, action_config,
|
||||
cooldown_secs, max_concurrent, dedup_window_secs,
|
||||
notify_channel, notify_user, notify_on_success, notify_on_failure, notify_on_attention,
|
||||
state, next_fire_at, created_at, updated_at
|
||||
) VALUES (
|
||||
?1, ?2, ?3, ?4, ?5,
|
||||
?6, ?7, ?8, ?9,
|
||||
?10, ?11, ?12,
|
||||
?13, ?14, ?15, ?16, ?17,
|
||||
?18, ?19, ?20, ?21
|
||||
)
|
||||
"#,
|
||||
params![
|
||||
routine.id.to_string(),
|
||||
routine.name.as_str(),
|
||||
routine.description.as_str(),
|
||||
routine.user_id.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),
|
||||
fmt_ts(&routine.created_at),
|
||||
fmt_ts(&routine.updated_at),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_routine(&self, id: Uuid) -> Result<Option<Routine>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&format!("SELECT {} FROM routines WHERE id = ?1", ROUTINE_COLUMNS),
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(Some(row_to_routine_libsql(&row)?)),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_routine_by_name(
|
||||
&self,
|
||||
user_id: &str,
|
||||
name: &str,
|
||||
) -> Result<Option<Routine>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&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()))?
|
||||
{
|
||||
Some(row) => Ok(Some(row_to_routine_libsql(&row)?)),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_routines(&self, user_id: &str) -> Result<Vec<Routine>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&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()))?
|
||||
{
|
||||
routines.push(row_to_routine_libsql(&row)?);
|
||||
}
|
||||
Ok(routines)
|
||||
}
|
||||
|
||||
async fn list_event_routines(&self) -> Result<Vec<Routine>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&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()))?
|
||||
{
|
||||
routines.push(row_to_routine_libsql(&row)?);
|
||||
}
|
||||
Ok(routines)
|
||||
}
|
||||
|
||||
async fn list_due_cron_routines(&self) -> Result<Vec<Routine>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&format!(
|
||||
"SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'cron' AND next_fire_at IS NOT NULL AND next_fire_at <= ?1",
|
||||
ROUTINE_COLUMNS
|
||||
),
|
||||
params![now],
|
||||
)
|
||||
.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()))?
|
||||
{
|
||||
routines.push(row_to_routine_libsql(&row)?);
|
||||
}
|
||||
Ok(routines)
|
||||
}
|
||||
|
||||
async fn update_routine(&self, routine: &Routine) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let trigger_type = routine.trigger.type_tag();
|
||||
let trigger_config = routine.trigger.to_config_json();
|
||||
let action_type = routine.action.type_tag();
|
||||
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 now = fmt_ts(&Utc::now());
|
||||
|
||||
conn.execute(
|
||||
r#"
|
||||
UPDATE routines SET
|
||||
name = ?2, description = ?3, enabled = ?4,
|
||||
trigger_type = ?5, trigger_config = ?6,
|
||||
action_type = ?7, action_config = ?8,
|
||||
cooldown_secs = ?9, max_concurrent = ?10, dedup_window_secs = ?11,
|
||||
notify_channel = ?12, notify_user = ?13,
|
||||
notify_on_success = ?14, notify_on_failure = ?15, notify_on_attention = ?16,
|
||||
state = ?17, next_fire_at = ?18,
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn update_routine_runtime(
|
||||
&self,
|
||||
id: Uuid,
|
||||
last_run_at: DateTime<Utc>,
|
||||
next_fire_at: Option<DateTime<Utc>>,
|
||||
run_count: u64,
|
||||
consecutive_failures: u32,
|
||||
state: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute(
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_routine(&self, id: Uuid) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let count = conn
|
||||
.execute(
|
||||
"DELETE FROM routines WHERE id = ?1",
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(count > 0)
|
||||
}
|
||||
|
||||
async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
conn.execute(
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn complete_routine_run(
|
||||
&self,
|
||||
id: Uuid,
|
||||
status: RunStatus,
|
||||
result_summary: Option<&str>,
|
||||
tokens_used: Option<i32>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute(
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_routine_runs(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
limit: i64,
|
||||
) -> Result<Vec<RoutineRun>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
&format!(
|
||||
"SELECT {} FROM routine_runs WHERE routine_id = ?1 ORDER BY started_at DESC LIMIT ?2",
|
||||
ROUTINE_RUN_COLUMNS
|
||||
),
|
||||
params![routine_id.to_string(), limit],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
runs.push(row_to_routine_run_libsql(&row)?);
|
||||
}
|
||||
Ok(runs)
|
||||
}
|
||||
|
||||
async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result<i64, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT COUNT(*) as cnt FROM routine_runs WHERE routine_id = ?1 AND status = 'running'",
|
||||
params![routine_id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(get_i64(&row, 0)),
|
||||
None => Ok(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,405 @@
|
||||
//! Sandbox-related SandboxStore implementation for LibSqlBackend.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use chrono::{DateTime, Utc};
|
||||
use libsql::params;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
LibSqlBackend, fmt_opt_ts, fmt_ts, get_i64, get_json, get_opt_bool, get_opt_text, get_opt_ts,
|
||||
get_text, get_ts, opt_text,
|
||||
};
|
||||
use crate::db::SandboxStore;
|
||||
use crate::error::DatabaseError;
|
||||
use crate::history::{JobEventRecord, SandboxJobRecord, SandboxJobSummary};
|
||||
|
||||
#[async_trait]
|
||||
impl SandboxStore for LibSqlBackend {
|
||||
async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
conn.execute(
|
||||
r#"
|
||||
INSERT INTO agent_jobs (
|
||||
id, title, description, status, source, user_id, project_dir,
|
||||
success, failure_reason, created_at, started_at, completed_at
|
||||
) VALUES (?1, ?2, ?3, ?4, 'sandbox', ?5, ?6, ?7, ?8, ?9, ?10, ?11)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
status = excluded.status,
|
||||
success = excluded.success,
|
||||
failure_reason = excluded.failure_reason,
|
||||
started_at = excluded.started_at,
|
||||
completed_at = excluded.completed_at
|
||||
"#,
|
||||
params![
|
||||
job.id.to_string(),
|
||||
job.task.as_str(),
|
||||
job.credential_grants_json.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<Option<SandboxJobRecord>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, title, description, status, user_id, project_dir,
|
||||
success, failure_reason, created_at, started_at, completed_at
|
||||
FROM agent_jobs WHERE id = ?1 AND source = 'sandbox'
|
||||
"#,
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(Some(SandboxJobRecord {
|
||||
id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
task: get_text(&row, 1),
|
||||
credential_grants_json: get_text(&row, 2),
|
||||
status: get_text(&row, 3),
|
||||
user_id: get_text(&row, 4),
|
||||
project_dir: get_text(&row, 5),
|
||||
success: get_opt_bool(&row, 6),
|
||||
failure_reason: get_opt_text(&row, 7),
|
||||
created_at: get_ts(&row, 8),
|
||||
started_at: get_opt_ts(&row, 9),
|
||||
completed_at: get_opt_ts(&row, 10),
|
||||
})),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_sandbox_jobs(&self) -> Result<Vec<SandboxJobRecord>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, title, description, status, user_id, project_dir,
|
||||
success, failure_reason, created_at, started_at, completed_at
|
||||
FROM agent_jobs WHERE source = 'sandbox'
|
||||
ORDER BY created_at DESC
|
||||
"#,
|
||||
(),
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
jobs.push(SandboxJobRecord {
|
||||
id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
task: get_text(&row, 1),
|
||||
credential_grants_json: get_text(&row, 2),
|
||||
status: get_text(&row, 3),
|
||||
user_id: get_text(&row, 4),
|
||||
project_dir: get_text(&row, 5),
|
||||
success: get_opt_bool(&row, 6),
|
||||
failure_reason: get_opt_text(&row, 7),
|
||||
created_at: get_ts(&row, 8),
|
||||
started_at: get_opt_ts(&row, 9),
|
||||
completed_at: get_opt_ts(&row, 10),
|
||||
});
|
||||
}
|
||||
Ok(jobs)
|
||||
}
|
||||
|
||||
async fn update_sandbox_job_status(
|
||||
&self,
|
||||
id: Uuid,
|
||||
status: &str,
|
||||
success: Option<bool>,
|
||||
message: Option<&str>,
|
||||
started_at: Option<DateTime<Utc>>,
|
||||
completed_at: Option<DateTime<Utc>>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
conn.execute(
|
||||
r#"
|
||||
UPDATE agent_jobs SET
|
||||
status = ?2,
|
||||
success = COALESCE(?3, success),
|
||||
failure_reason = COALESCE(?4, failure_reason),
|
||||
started_at = COALESCE(?5, started_at),
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cleanup_stale_sandbox_jobs(&self) -> Result<u64, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
let count = conn
|
||||
.execute(
|
||||
r#"
|
||||
UPDATE agent_jobs SET
|
||||
status = 'interrupted',
|
||||
failure_reason = 'Process restarted',
|
||||
completed_at = ?1
|
||||
WHERE source = 'sandbox' AND status IN ('running', 'creating')
|
||||
"#,
|
||||
params![now],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
if count > 0 {
|
||||
tracing::info!("Marked {} stale sandbox jobs as interrupted", count);
|
||||
}
|
||||
Ok(count)
|
||||
}
|
||||
|
||||
async fn sandbox_job_summary(&self) -> Result<SandboxJobSummary, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT status, COUNT(*) as cnt FROM agent_jobs WHERE source = 'sandbox' GROUP BY status",
|
||||
(),
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
let status = get_text(&row, 0);
|
||||
let count = get_i64(&row, 1) as usize;
|
||||
summary.total += count;
|
||||
match status.as_str() {
|
||||
"creating" => summary.creating += count,
|
||||
"running" => summary.running += count,
|
||||
"completed" => summary.completed += count,
|
||||
"failed" => summary.failed += count,
|
||||
"interrupted" => summary.interrupted += count,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(summary)
|
||||
}
|
||||
|
||||
async fn list_sandbox_jobs_for_user(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<Vec<SandboxJobRecord>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, title, description, status, user_id, project_dir,
|
||||
success, failure_reason, created_at, started_at, completed_at
|
||||
FROM agent_jobs WHERE source = 'sandbox' AND user_id = ?1
|
||||
ORDER BY created_at DESC
|
||||
"#,
|
||||
libsql::params![user_id],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
jobs.push(SandboxJobRecord {
|
||||
id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
task: get_text(&row, 1),
|
||||
credential_grants_json: get_text(&row, 2),
|
||||
status: get_text(&row, 3),
|
||||
user_id: get_text(&row, 4),
|
||||
project_dir: get_text(&row, 5),
|
||||
success: get_opt_bool(&row, 6),
|
||||
failure_reason: get_opt_text(&row, 7),
|
||||
created_at: get_ts(&row, 8),
|
||||
started_at: get_opt_ts(&row, 9),
|
||||
completed_at: get_opt_ts(&row, 10),
|
||||
});
|
||||
}
|
||||
Ok(jobs)
|
||||
}
|
||||
|
||||
async fn sandbox_job_summary_for_user(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<SandboxJobSummary, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT status, COUNT(*) as cnt FROM agent_jobs WHERE source = 'sandbox' AND user_id = ?1 GROUP BY status",
|
||||
libsql::params![user_id],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
let status = get_text(&row, 0);
|
||||
let count = get_i64(&row, 1) as usize;
|
||||
summary.total += count;
|
||||
match status.as_str() {
|
||||
"creating" => summary.creating += count,
|
||||
"running" => summary.running += count,
|
||||
"completed" => summary.completed += count,
|
||||
"failed" => summary.failed += count,
|
||||
"interrupted" => summary.interrupted += count,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(summary)
|
||||
}
|
||||
|
||||
async fn sandbox_job_belongs_to_user(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT 1 FROM agent_jobs WHERE id = ?1 AND user_id = ?2 AND source = 'sandbox'",
|
||||
libsql::params![job_id.to_string(), user_id],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
let found = rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(found.is_some())
|
||||
}
|
||||
|
||||
async fn update_sandbox_job_mode(&self, id: Uuid, mode: &str) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_sandbox_job_mode(&self, id: Uuid) -> Result<Option<String>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT job_mode FROM agent_jobs WHERE id = ?1",
|
||||
params![id.to_string()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
match rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
{
|
||||
Some(row) => Ok(Some(get_text(&row, 0))),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn save_job_event(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
event_type: &str,
|
||||
data: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_job_events(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
limit: Option<i64>,
|
||||
) -> Result<Vec<JobEventRecord>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = if let Some(n) = limit {
|
||||
conn.query(
|
||||
r#"
|
||||
SELECT id, job_id, event_type, data, created_at
|
||||
FROM (
|
||||
SELECT id, job_id, event_type, data, created_at
|
||||
FROM job_events WHERE job_id = ?1
|
||||
ORDER BY id DESC
|
||||
LIMIT ?2
|
||||
)
|
||||
ORDER BY id ASC
|
||||
"#,
|
||||
params![job_id.to_string(), n],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?
|
||||
} else {
|
||||
conn.query(
|
||||
r#"
|
||||
SELECT id, job_id, event_type, data, created_at
|
||||
FROM job_events WHERE job_id = ?1 ORDER BY id ASC
|
||||
"#,
|
||||
params![job_id.to_string()],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
events.push(JobEventRecord {
|
||||
id: get_i64(&row, 0),
|
||||
job_id: get_text(&row, 1).parse().unwrap_or_default(),
|
||||
event_type: get_text(&row, 2),
|
||||
data: get_json(&row, 3),
|
||||
created_at: get_ts(&row, 4),
|
||||
});
|
||||
}
|
||||
Ok(events)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
//! Settings-related SettingsStore implementation for LibSqlBackend.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use libsql::params;
|
||||
|
||||
use super::{LibSqlBackend, fmt_ts, get_i64, get_json, get_text, get_ts};
|
||||
use crate::db::SettingsStore;
|
||||
use crate::error::DatabaseError;
|
||||
use crate::history::SettingRow;
|
||||
|
||||
use chrono::Utc;
|
||||
|
||||
#[async_trait]
|
||||
impl SettingsStore for LibSqlBackend {
|
||||
async fn get_setting(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
) -> Result<Option<serde_json::Value>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT value FROM settings WHERE user_id = ?1 AND key = ?2",
|
||||
params![user_id, key],
|
||||
)
|
||||
.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),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_setting_full(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
) -> Result<Option<SettingRow>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT key, value, updated_at FROM settings WHERE user_id = ?1 AND key = ?2",
|
||||
params![user_id, key],
|
||||
)
|
||||
.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),
|
||||
updated_at: get_ts(&row, 2),
|
||||
})),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn set_setting(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
value: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute(
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_setting(&self, user_id: &str, key: &str) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let count = conn
|
||||
.execute(
|
||||
"DELETE FROM settings WHERE user_id = ?1 AND key = ?2",
|
||||
params![user_id, key],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(count > 0)
|
||||
}
|
||||
|
||||
async fn list_settings(&self, user_id: &str) -> Result<Vec<SettingRow>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT key, value, updated_at FROM settings WHERE user_id = ?1 ORDER BY key",
|
||||
params![user_id],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
settings.push(SettingRow {
|
||||
key: get_text(&row, 0),
|
||||
value: get_json(&row, 1),
|
||||
updated_at: get_ts(&row, 2),
|
||||
});
|
||||
}
|
||||
Ok(settings)
|
||||
}
|
||||
|
||||
async fn get_all_settings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<HashMap<String, serde_json::Value>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT key, value FROM settings WHERE user_id = ?1",
|
||||
params![user_id],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
map.insert(get_text(&row, 0), get_json(&row, 1));
|
||||
}
|
||||
Ok(map)
|
||||
}
|
||||
|
||||
async fn set_all_settings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
settings: &HashMap<String, serde_json::Value>,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute("BEGIN", ())
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
|
||||
for (key, value) in settings {
|
||||
if let Err(e) = conn
|
||||
.execute(
|
||||
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.as_str(), value.to_string(), now.as_str()],
|
||||
)
|
||||
.await
|
||||
{
|
||||
let _ = conn.execute("ROLLBACK", ()).await;
|
||||
return Err(DatabaseError::Query(e.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
conn.execute("COMMIT", ())
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Query(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn has_settings(&self, user_id: &str) -> Result<bool, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT COUNT(*) as cnt FROM settings WHERE user_id = ?1",
|
||||
params![user_id],
|
||||
)
|
||||
.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),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
//! Tool failure-related ToolFailureStore implementation for LibSqlBackend.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use libsql::params;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{LibSqlBackend, fmt_ts, get_i64, get_opt_text, get_text, get_ts};
|
||||
use crate::agent::BrokenTool;
|
||||
use crate::db::ToolFailureStore;
|
||||
use crate::error::DatabaseError;
|
||||
|
||||
use chrono::Utc;
|
||||
|
||||
#[async_trait]
|
||||
impl ToolFailureStore for LibSqlBackend {
|
||||
async fn record_tool_failure(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
error_message: &str,
|
||||
) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let now = fmt_ts(&Utc::now());
|
||||
conn.execute(
|
||||
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
|
||||
error_message = ?3,
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_broken_tools(&self, threshold: i32) -> Result<Vec<BrokenTool>, DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT tool_name, error_message, error_count, first_failure, last_failure,
|
||||
last_build_result, repair_attempts
|
||||
FROM tool_failures
|
||||
WHERE error_count >= ?1 AND repaired_at IS NULL
|
||||
ORDER BY error_count DESC
|
||||
"#,
|
||||
params![threshold as i64],
|
||||
)
|
||||
.await
|
||||
.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()))?
|
||||
{
|
||||
tools.push(BrokenTool {
|
||||
name: get_text(&row, 0),
|
||||
last_error: get_opt_text(&row, 1),
|
||||
failure_count: get_i64(&row, 2) as u32,
|
||||
first_failure: get_ts(&row, 3),
|
||||
last_failure: get_ts(&row, 4),
|
||||
last_build_result: get_opt_text(&row, 5)
|
||||
.and_then(|s| serde_json::from_str(&s).ok()),
|
||||
repair_attempts: get_i64(&row, 6) as u32,
|
||||
});
|
||||
}
|
||||
Ok(tools)
|
||||
}
|
||||
|
||||
async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError> {
|
||||
let conn = self.connect().await?;
|
||||
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()))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,607 @@
|
||||
//! Workspace-related WorkspaceStore implementation for LibSqlBackend.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use libsql::params;
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::{
|
||||
LibSqlBackend, fmt_ts, get_i64, get_opt_text, get_opt_ts, get_text, get_ts,
|
||||
row_to_memory_document,
|
||||
};
|
||||
use crate::db::WorkspaceStore;
|
||||
use crate::error::WorkspaceError;
|
||||
use crate::workspace::{
|
||||
MemoryChunk, MemoryDocument, RankedResult, SearchConfig, SearchResult, WorkspaceEntry,
|
||||
reciprocal_rank_fusion,
|
||||
};
|
||||
|
||||
use chrono::Utc;
|
||||
|
||||
#[async_trait]
|
||||
impl WorkspaceStore for LibSqlBackend {
|
||||
async fn get_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, user_id, agent_id, path, content,
|
||||
created_at, updated_at, metadata
|
||||
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!("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(),
|
||||
user_id: user_id.to_string(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_document_by_id(&self, id: Uuid) -> Result<MemoryDocument, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, user_id, agent_id, path, content,
|
||||
created_at, updated_at, metadata
|
||||
FROM memory_documents WHERE id = ?1
|
||||
"#,
|
||||
params![id.to_string()],
|
||||
)
|
||||
.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(),
|
||||
user_id: "unknown".to_string(),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_or_create_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError> {
|
||||
// Try get
|
||||
match self.get_document_by_path(user_id, agent_id, path).await {
|
||||
Ok(doc) => return Ok(doc),
|
||||
Err(WorkspaceError::DocumentNotFound { .. }) => {}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
|
||||
// Create
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let id = Uuid::new_v4();
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
conn.execute(
|
||||
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),
|
||||
})?;
|
||||
|
||||
self.get_document_by_path(user_id, agent_id, path).await
|
||||
}
|
||||
|
||||
async fn update_document(&self, id: Uuid, content: &str) -> Result<(), WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
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),
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<(), WorkspaceError> {
|
||||
let doc = self.get_document_by_path(user_id, agent_id, path).await?;
|
||||
self.delete_chunks(doc.id).await?;
|
||||
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
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),
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_directory(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
directory: &str,
|
||||
) -> Result<Vec<WorkspaceEntry>, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let dir = if !directory.is_empty() && !directory.ends_with('/') {
|
||||
format!("{}/", directory)
|
||||
} else {
|
||||
directory.to_string()
|
||||
};
|
||||
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let pattern = if dir.is_empty() {
|
||||
"%".to_string()
|
||||
} else {
|
||||
format!("{}%", dir)
|
||||
};
|
||||
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT path, updated_at, substr(content, 1, 200) as content_preview
|
||||
FROM memory_documents
|
||||
WHERE user_id = ?1 AND agent_id IS ?2
|
||||
AND (?3 = '%' OR path LIKE ?3)
|
||||
ORDER BY path
|
||||
"#,
|
||||
params![user_id, agent_id_str.as_deref(), pattern],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("List directory failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut entries_map: HashMap<String, WorkspaceEntry> = HashMap::new();
|
||||
|
||||
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);
|
||||
|
||||
let relative = if dir.is_empty() {
|
||||
&full_path
|
||||
} else if let Some(stripped) = full_path.strip_prefix(&dir) {
|
||||
stripped
|
||||
} else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let child_name = if let Some(slash_pos) = relative.find('/') {
|
||||
&relative[..slash_pos]
|
||||
} else {
|
||||
relative
|
||||
};
|
||||
|
||||
if child_name.is_empty() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let is_dir = relative.contains('/');
|
||||
let entry_path = if dir.is_empty() {
|
||||
child_name.to_string()
|
||||
} else {
|
||||
format!("{}{}", dir, child_name)
|
||||
};
|
||||
|
||||
entries_map
|
||||
.entry(child_name.to_string())
|
||||
.and_modify(|e| {
|
||||
if is_dir {
|
||||
e.is_directory = true;
|
||||
e.content_preview = None;
|
||||
}
|
||||
if let (Some(existing), Some(new)) = (&e.updated_at, &updated_at)
|
||||
&& new > existing
|
||||
{
|
||||
e.updated_at = Some(*new);
|
||||
}
|
||||
})
|
||||
.or_insert(WorkspaceEntry {
|
||||
path: entry_path,
|
||||
is_directory: is_dir,
|
||||
updated_at,
|
||||
content_preview: if is_dir { None } else { content_preview },
|
||||
});
|
||||
}
|
||||
|
||||
let mut entries: Vec<WorkspaceEntry> = entries_map.into_values().collect();
|
||||
entries.sort_by(|a, b| a.path.cmp(&b.path));
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
async fn list_all_paths(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
) -> Result<Vec<String>, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let mut rows = conn
|
||||
.query(
|
||||
"SELECT path FROM memory_documents WHERE user_id = ?1 AND agent_id IS ?2 ORDER BY path",
|
||||
params![user_id, agent_id_str.as_deref()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("List paths failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut paths = Vec::new();
|
||||
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)
|
||||
}
|
||||
|
||||
async fn list_documents(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
) -> Result<Vec<MemoryDocument>, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT id, user_id, agent_id, path, content,
|
||||
created_at, updated_at, metadata
|
||||
FROM memory_documents
|
||||
WHERE user_id = ?1 AND agent_id IS ?2
|
||||
ORDER BY updated_at DESC
|
||||
"#,
|
||||
params![user_id, agent_id_str.as_deref()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("Query failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut docs = Vec::new();
|
||||
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)
|
||||
}
|
||||
|
||||
async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::ChunkingFailed {
|
||||
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),
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn insert_chunk(
|
||||
&self,
|
||||
document_id: Uuid,
|
||||
chunk_index: i32,
|
||||
content: &str,
|
||||
embedding: Option<&[f32]>,
|
||||
) -> Result<Uuid, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::ChunkingFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let id = Uuid::new_v4();
|
||||
let embedding_blob = embedding.map(|e| {
|
||||
let bytes: Vec<u8> = e.iter().flat_map(|f| f.to_le_bytes()).collect();
|
||||
bytes
|
||||
});
|
||||
|
||||
conn.execute(
|
||||
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),
|
||||
})?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
async fn update_chunk_embedding(
|
||||
&self,
|
||||
chunk_id: Uuid,
|
||||
embedding: &[f32],
|
||||
) -> Result<(), WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::EmbeddingFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let bytes: Vec<u8> = 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),
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_chunks_without_embeddings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<MemoryChunk>, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT c.id, c.document_id, c.chunk_index, c.content, c.created_at
|
||||
FROM memory_chunks c
|
||||
JOIN memory_documents d ON d.id = c.document_id
|
||||
WHERE d.user_id = ?1 AND d.agent_id IS ?2
|
||||
AND c.embedding IS NULL
|
||||
LIMIT ?3
|
||||
"#,
|
||||
params![user_id, agent_id_str.as_deref(), limit as i64],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("Query failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut chunks = Vec::new();
|
||||
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(),
|
||||
chunk_index: get_i64(&row, 2) as i32,
|
||||
content: get_text(&row, 3),
|
||||
embedding: None,
|
||||
created_at: get_ts(&row, 4),
|
||||
});
|
||||
}
|
||||
Ok(chunks)
|
||||
}
|
||||
|
||||
async fn hybrid_search(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
query: &str,
|
||||
embedding: Option<&[f32]>,
|
||||
config: &SearchConfig,
|
||||
) -> Result<Vec<SearchResult>, WorkspaceError> {
|
||||
let conn = self
|
||||
.connect()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: e.to_string(),
|
||||
})?;
|
||||
let agent_id_str = agent_id.map(|id| id.to_string());
|
||||
let pre_limit = config.pre_fusion_limit as i64;
|
||||
|
||||
let fts_results = if config.use_fts {
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT c.id, c.document_id, c.content
|
||||
FROM memory_chunks_fts fts
|
||||
JOIN memory_chunks c ON c._rowid = fts.rowid
|
||||
JOIN memory_documents d ON d.id = c.document_id
|
||||
WHERE d.user_id = ?1 AND d.agent_id IS ?2
|
||||
AND memory_chunks_fts MATCH ?3
|
||||
ORDER BY rank
|
||||
LIMIT ?4
|
||||
"#,
|
||||
params![user_id, agent_id_str.as_deref(), query, pre_limit],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("FTS query failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(row) = rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("FTS row fetch failed: {}", e),
|
||||
})?
|
||||
{
|
||||
results.push(RankedResult {
|
||||
chunk_id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
document_id: get_text(&row, 1).parse().unwrap_or_default(),
|
||||
content: get_text(&row, 2),
|
||||
rank: results.len() as u32 + 1,
|
||||
});
|
||||
}
|
||||
results
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
|
||||
let vector_results = if let (true, Some(emb)) = (config.use_vector, embedding) {
|
||||
let vector_json = format!(
|
||||
"[{}]",
|
||||
emb.iter()
|
||||
.map(|f| f.to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
);
|
||||
|
||||
let mut rows = conn
|
||||
.query(
|
||||
r#"
|
||||
SELECT c.id, c.document_id, c.content
|
||||
FROM vector_top_k('idx_memory_chunks_embedding', vector(?1), ?2) AS top_k
|
||||
JOIN memory_chunks c ON c._rowid = top_k.id
|
||||
JOIN memory_documents d ON d.id = c.document_id
|
||||
WHERE d.user_id = ?3 AND d.agent_id IS ?4
|
||||
"#,
|
||||
params![vector_json, pre_limit, user_id, agent_id_str.as_deref()],
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("Vector query failed: {}", e),
|
||||
})?;
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(row) = rows
|
||||
.next()
|
||||
.await
|
||||
.map_err(|e| WorkspaceError::SearchFailed {
|
||||
reason: format!("Vector row fetch failed: {}", e),
|
||||
})?
|
||||
{
|
||||
results.push(RankedResult {
|
||||
chunk_id: get_text(&row, 0).parse().unwrap_or_default(),
|
||||
document_id: get_text(&row, 1).parse().unwrap_or_default(),
|
||||
content: get_text(&row, 2),
|
||||
rank: results.len() as u32 + 1,
|
||||
});
|
||||
}
|
||||
results
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
|
||||
if embedding.is_some() && !config.use_vector {
|
||||
tracing::warn!(
|
||||
"Embedding provided but vector search is disabled in config; using FTS-only results"
|
||||
);
|
||||
}
|
||||
|
||||
Ok(reciprocal_rank_fusion(fts_results, vector_results, config))
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
+50
-184
@@ -13,7 +13,7 @@
|
||||
pub mod postgres;
|
||||
|
||||
#[cfg(feature = "libsql")]
|
||||
pub mod libsql_backend;
|
||||
pub mod libsql;
|
||||
|
||||
#[cfg(feature = "libsql")]
|
||||
pub mod libsql_migrations;
|
||||
@@ -62,15 +62,11 @@ pub async fn connect_from_config(
|
||||
"LIBSQL_AUTH_TOKEN required when LIBSQL_URL is set".to_string(),
|
||||
)
|
||||
})?;
|
||||
libsql_backend::LibSqlBackend::new_remote_replica(
|
||||
db_path,
|
||||
url,
|
||||
token.expose_secret(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Pool(e.to_string()))?
|
||||
libsql::LibSqlBackend::new_remote_replica(db_path, url, token.expose_secret())
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Pool(e.to_string()))?
|
||||
} else {
|
||||
libsql_backend::LibSqlBackend::new_local(db_path)
|
||||
libsql::LibSqlBackend::new_local(db_path)
|
||||
.await
|
||||
.map_err(|e| DatabaseError::Pool(e.to_string()))?
|
||||
};
|
||||
@@ -92,37 +88,27 @@ pub async fn connect_from_config(
|
||||
}
|
||||
}
|
||||
|
||||
/// Backend-agnostic database trait.
|
||||
///
|
||||
/// Combines all persistence operations from Store, Repository, and related
|
||||
/// stores into a single trait that can be implemented for different backends.
|
||||
// ==================== Sub-traits ====================
|
||||
//
|
||||
// Each sub-trait groups related persistence methods. The `Database` supertrait
|
||||
// combines them all, so existing `Arc<dyn Database>` consumers keep working.
|
||||
// Leaf consumers can depend on a specific sub-trait instead.
|
||||
|
||||
#[async_trait]
|
||||
pub trait Database: Send + Sync {
|
||||
/// Run schema migrations for this backend.
|
||||
async fn run_migrations(&self) -> Result<(), DatabaseError>;
|
||||
|
||||
// ==================== Conversations ====================
|
||||
|
||||
/// Create a new conversation.
|
||||
pub trait ConversationStore: Send + Sync {
|
||||
async fn create_conversation(
|
||||
&self,
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
/// Update conversation last activity.
|
||||
async fn touch_conversation(&self, id: Uuid) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Add a message to a conversation.
|
||||
async fn add_conversation_message(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
role: &str,
|
||||
content: &str,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
/// Ensure a conversation row exists (upsert).
|
||||
async fn ensure_conversation(
|
||||
&self,
|
||||
id: Uuid,
|
||||
@@ -130,103 +116,65 @@ pub trait Database: Send + Sync {
|
||||
user_id: &str,
|
||||
thread_id: Option<&str>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// List conversations with a title preview.
|
||||
async fn list_conversations_with_preview(
|
||||
&self,
|
||||
user_id: &str,
|
||||
channel: &str,
|
||||
limit: i64,
|
||||
) -> Result<Vec<ConversationSummary>, DatabaseError>;
|
||||
|
||||
/// Get or create the singleton assistant conversation.
|
||||
async fn get_or_create_assistant_conversation(
|
||||
&self,
|
||||
user_id: &str,
|
||||
channel: &str,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
/// Create a conversation with specific metadata.
|
||||
async fn create_conversation_with_metadata(
|
||||
&self,
|
||||
channel: &str,
|
||||
user_id: &str,
|
||||
metadata: &serde_json::Value,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
/// Load messages with cursor-based pagination.
|
||||
async fn list_conversation_messages_paginated(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
before: Option<DateTime<Utc>>,
|
||||
limit: i64,
|
||||
) -> Result<(Vec<ConversationMessage>, bool), DatabaseError>;
|
||||
|
||||
/// Merge a single key into conversation metadata.
|
||||
async fn update_conversation_metadata_field(
|
||||
&self,
|
||||
id: Uuid,
|
||||
key: &str,
|
||||
value: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Read conversation metadata.
|
||||
async fn get_conversation_metadata(
|
||||
&self,
|
||||
id: Uuid,
|
||||
) -> Result<Option<serde_json::Value>, DatabaseError>;
|
||||
|
||||
/// Load all messages for a conversation.
|
||||
async fn list_conversation_messages(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
) -> Result<Vec<ConversationMessage>, DatabaseError>;
|
||||
|
||||
/// Check if a conversation belongs to a specific user.
|
||||
async fn conversation_belongs_to_user(
|
||||
&self,
|
||||
conversation_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<bool, DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Jobs ====================
|
||||
|
||||
/// Save a job context.
|
||||
#[async_trait]
|
||||
pub trait JobStore: Send + Sync {
|
||||
async fn save_job(&self, ctx: &JobContext) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Get a job by ID.
|
||||
async fn get_job(&self, id: Uuid) -> Result<Option<JobContext>, DatabaseError>;
|
||||
|
||||
/// Update job status.
|
||||
async fn update_job_status(
|
||||
&self,
|
||||
id: Uuid,
|
||||
status: JobState,
|
||||
failure_reason: Option<&str>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Mark job as stuck.
|
||||
async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Get stuck jobs.
|
||||
async fn get_stuck_jobs(&self) -> Result<Vec<Uuid>, DatabaseError>;
|
||||
|
||||
// ==================== Actions ====================
|
||||
|
||||
/// Save a job action.
|
||||
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<Vec<ActionRecord>, DatabaseError>;
|
||||
|
||||
// ==================== LLM Calls ====================
|
||||
|
||||
/// Record an LLM call.
|
||||
async fn record_llm_call(&self, record: &LlmCallRecord<'_>) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
// ==================== Estimation Snapshots ====================
|
||||
|
||||
/// Save an estimation snapshot.
|
||||
async fn save_estimation_snapshot(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
@@ -236,8 +184,6 @@ pub trait Database: Send + Sync {
|
||||
estimated_time_secs: i32,
|
||||
estimated_value: Decimal,
|
||||
) -> Result<Uuid, DatabaseError>;
|
||||
|
||||
/// Update estimation snapshot with actual values.
|
||||
async fn update_estimation_actuals(
|
||||
&self,
|
||||
id: Uuid,
|
||||
@@ -245,19 +191,13 @@ pub trait Database: Send + Sync {
|
||||
actual_time_secs: i32,
|
||||
actual_value: Option<Decimal>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Sandbox Jobs ====================
|
||||
|
||||
/// Insert a new sandbox job.
|
||||
#[async_trait]
|
||||
pub trait SandboxStore: 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<Option<SandboxJobRecord>, DatabaseError>;
|
||||
|
||||
/// List all sandbox jobs, most recent first.
|
||||
async fn list_sandbox_jobs(&self) -> Result<Vec<SandboxJobRecord>, DatabaseError>;
|
||||
|
||||
/// Update sandbox job status.
|
||||
async fn update_sandbox_job_status(
|
||||
&self,
|
||||
id: Uuid,
|
||||
@@ -267,83 +207,49 @@ pub trait Database: Send + Sync {
|
||||
started_at: Option<DateTime<Utc>>,
|
||||
completed_at: Option<DateTime<Utc>>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Mark stale sandbox jobs as interrupted.
|
||||
async fn cleanup_stale_sandbox_jobs(&self) -> Result<u64, DatabaseError>;
|
||||
|
||||
/// Get sandbox job summary.
|
||||
async fn sandbox_job_summary(&self) -> Result<SandboxJobSummary, DatabaseError>;
|
||||
|
||||
/// List sandbox jobs for a specific user, most recent first.
|
||||
async fn list_sandbox_jobs_for_user(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<Vec<SandboxJobRecord>, DatabaseError>;
|
||||
|
||||
/// Get sandbox job summary for a specific user.
|
||||
async fn sandbox_job_summary_for_user(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<SandboxJobSummary, DatabaseError>;
|
||||
|
||||
/// Check if a sandbox job belongs to a specific user.
|
||||
async fn sandbox_job_belongs_to_user(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
user_id: &str,
|
||||
) -> Result<bool, DatabaseError>;
|
||||
|
||||
/// Update sandbox job mode.
|
||||
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<Option<String>, DatabaseError>;
|
||||
|
||||
// ==================== Job Events ====================
|
||||
|
||||
/// Persist a job event.
|
||||
async fn save_job_event(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
event_type: &str,
|
||||
data: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Load job events, returning the most recent `limit` entries (or all if `None`).
|
||||
async fn list_job_events(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
limit: Option<i64>,
|
||||
) -> Result<Vec<JobEventRecord>, DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Routines ====================
|
||||
|
||||
/// Create a new routine.
|
||||
#[async_trait]
|
||||
pub trait RoutineStore: Send + Sync {
|
||||
async fn create_routine(&self, routine: &Routine) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Get a routine by ID.
|
||||
async fn get_routine(&self, id: Uuid) -> Result<Option<Routine>, DatabaseError>;
|
||||
|
||||
/// Get a routine by user_id and name.
|
||||
async fn get_routine_by_name(
|
||||
&self,
|
||||
user_id: &str,
|
||||
name: &str,
|
||||
) -> Result<Option<Routine>, DatabaseError>;
|
||||
|
||||
/// List routines for a user.
|
||||
async fn list_routines(&self, user_id: &str) -> Result<Vec<Routine>, DatabaseError>;
|
||||
|
||||
/// List all enabled event routines.
|
||||
async fn list_event_routines(&self) -> Result<Vec<Routine>, DatabaseError>;
|
||||
|
||||
/// List due cron routines.
|
||||
async fn list_due_cron_routines(&self) -> Result<Vec<Routine>, DatabaseError>;
|
||||
|
||||
/// Update a routine.
|
||||
async fn update_routine(&self, routine: &Routine) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Update runtime state after a routine fires.
|
||||
async fn update_routine_runtime(
|
||||
&self,
|
||||
id: Uuid,
|
||||
@@ -353,16 +259,8 @@ pub trait Database: Send + Sync {
|
||||
consecutive_failures: u32,
|
||||
state: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Delete a routine.
|
||||
async fn delete_routine(&self, id: Uuid) -> Result<bool, DatabaseError>;
|
||||
|
||||
// ==================== Routine Runs ====================
|
||||
|
||||
/// Record a routine run starting.
|
||||
async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Complete a routine run.
|
||||
async fn complete_routine_run(
|
||||
&self,
|
||||
id: Uuid,
|
||||
@@ -370,141 +268,97 @@ pub trait Database: Send + Sync {
|
||||
result_summary: Option<&str>,
|
||||
tokens_used: Option<i32>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// List recent runs for a routine.
|
||||
async fn list_routine_runs(
|
||||
&self,
|
||||
routine_id: Uuid,
|
||||
limit: i64,
|
||||
) -> Result<Vec<RoutineRun>, DatabaseError>;
|
||||
|
||||
/// Count currently running runs for a routine.
|
||||
async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result<i64, DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Tool Failures ====================
|
||||
|
||||
/// Record a tool failure (upsert).
|
||||
#[async_trait]
|
||||
pub trait ToolFailureStore: Send + Sync {
|
||||
async fn record_tool_failure(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
error_message: &str,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Get broken tools exceeding threshold.
|
||||
async fn get_broken_tools(&self, threshold: i32) -> Result<Vec<BrokenTool>, DatabaseError>;
|
||||
|
||||
/// Mark a tool as repaired.
|
||||
async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Increment repair attempts.
|
||||
async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Settings ====================
|
||||
|
||||
/// Get a single setting.
|
||||
#[async_trait]
|
||||
pub trait SettingsStore: Send + Sync {
|
||||
async fn get_setting(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
) -> Result<Option<serde_json::Value>, DatabaseError>;
|
||||
|
||||
/// Get a single setting with metadata.
|
||||
async fn get_setting_full(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
) -> Result<Option<SettingRow>, DatabaseError>;
|
||||
|
||||
/// Set a single setting (upsert).
|
||||
async fn set_setting(
|
||||
&self,
|
||||
user_id: &str,
|
||||
key: &str,
|
||||
value: &serde_json::Value,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Delete a single setting.
|
||||
async fn delete_setting(&self, user_id: &str, key: &str) -> Result<bool, DatabaseError>;
|
||||
|
||||
/// List all settings for a user.
|
||||
async fn list_settings(&self, user_id: &str) -> Result<Vec<SettingRow>, DatabaseError>;
|
||||
|
||||
/// Get all settings as a flat map.
|
||||
async fn get_all_settings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
) -> Result<HashMap<String, serde_json::Value>, DatabaseError>;
|
||||
|
||||
/// Bulk-write settings atomically.
|
||||
async fn set_all_settings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
settings: &HashMap<String, serde_json::Value>,
|
||||
) -> Result<(), DatabaseError>;
|
||||
|
||||
/// Check if settings exist for a user.
|
||||
async fn has_settings(&self, user_id: &str) -> Result<bool, DatabaseError>;
|
||||
}
|
||||
|
||||
// ==================== Workspace: Documents ====================
|
||||
|
||||
/// Get a document by path.
|
||||
#[async_trait]
|
||||
pub trait WorkspaceStore: Send + Sync {
|
||||
async fn get_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError>;
|
||||
|
||||
/// Get a document by ID.
|
||||
async fn get_document_by_id(&self, id: Uuid) -> Result<MemoryDocument, WorkspaceError>;
|
||||
|
||||
/// Get or create a document by path.
|
||||
async fn get_or_create_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<MemoryDocument, WorkspaceError>;
|
||||
|
||||
/// Update a document's content.
|
||||
async fn update_document(&self, id: Uuid, content: &str) -> Result<(), WorkspaceError>;
|
||||
|
||||
/// Delete a document by path.
|
||||
async fn delete_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
path: &str,
|
||||
) -> Result<(), WorkspaceError>;
|
||||
|
||||
/// List files and directories in a directory path.
|
||||
async fn list_directory(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
directory: &str,
|
||||
) -> Result<Vec<WorkspaceEntry>, WorkspaceError>;
|
||||
|
||||
/// List all file paths in the workspace.
|
||||
async fn list_all_paths(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
) -> Result<Vec<String>, WorkspaceError>;
|
||||
|
||||
/// List all documents for a user.
|
||||
async fn list_documents(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
) -> Result<Vec<MemoryDocument>, WorkspaceError>;
|
||||
|
||||
// ==================== Workspace: Chunks ====================
|
||||
|
||||
/// Delete all chunks for a document.
|
||||
async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError>;
|
||||
|
||||
/// Insert a chunk.
|
||||
async fn insert_chunk(
|
||||
&self,
|
||||
document_id: Uuid,
|
||||
@@ -512,25 +366,17 @@ pub trait Database: Send + Sync {
|
||||
content: &str,
|
||||
embedding: Option<&[f32]>,
|
||||
) -> Result<Uuid, WorkspaceError>;
|
||||
|
||||
/// Update a chunk's embedding.
|
||||
async fn update_chunk_embedding(
|
||||
&self,
|
||||
chunk_id: Uuid,
|
||||
embedding: &[f32],
|
||||
) -> Result<(), WorkspaceError>;
|
||||
|
||||
/// Get chunks without embeddings for backfilling.
|
||||
async fn get_chunks_without_embeddings(
|
||||
&self,
|
||||
user_id: &str,
|
||||
agent_id: Option<Uuid>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<MemoryChunk>, WorkspaceError>;
|
||||
|
||||
// ==================== Workspace: Search ====================
|
||||
|
||||
/// Perform hybrid search combining FTS and vector similarity.
|
||||
async fn hybrid_search(
|
||||
&self,
|
||||
user_id: &str,
|
||||
@@ -540,3 +386,23 @@ pub trait Database: Send + Sync {
|
||||
config: &SearchConfig,
|
||||
) -> Result<Vec<SearchResult>, WorkspaceError>;
|
||||
}
|
||||
|
||||
/// Backend-agnostic database supertrait.
|
||||
///
|
||||
/// Combines all sub-traits into one. Existing `Arc<dyn Database>` consumers
|
||||
/// continue to work; leaf consumers can depend on a specific sub-trait instead.
|
||||
#[async_trait]
|
||||
pub trait Database:
|
||||
ConversationStore
|
||||
+ JobStore
|
||||
+ SandboxStore
|
||||
+ RoutineStore
|
||||
+ ToolFailureStore
|
||||
+ SettingsStore
|
||||
+ WorkspaceStore
|
||||
+ Send
|
||||
+ Sync
|
||||
{
|
||||
/// Run schema migrations for this backend.
|
||||
async fn run_migrations(&self) -> Result<(), DatabaseError>;
|
||||
}
|
||||
|
||||
+34
-22
@@ -15,7 +15,10 @@ 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;
|
||||
use crate::db::{
|
||||
ConversationStore, Database, JobStore, RoutineStore, SandboxStore, SettingsStore,
|
||||
ToolFailureStore, WorkspaceStore,
|
||||
};
|
||||
use crate::error::{DatabaseError, WorkspaceError};
|
||||
use crate::history::{
|
||||
ConversationMessage, ConversationSummary, JobEventRecord, LlmCallRecord, SandboxJobRecord,
|
||||
@@ -51,14 +54,19 @@ impl PgBackend {
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Database (supertrait) ====================
|
||||
|
||||
#[async_trait]
|
||||
impl Database for PgBackend {
|
||||
async fn run_migrations(&self) -> Result<(), DatabaseError> {
|
||||
self.store.run_migrations().await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Conversations ====================
|
||||
// ==================== ConversationStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl ConversationStore for PgBackend {
|
||||
async fn create_conversation(
|
||||
&self,
|
||||
channel: &str,
|
||||
@@ -174,9 +182,12 @@ impl Database for PgBackend {
|
||||
.conversation_belongs_to_user(conversation_id, user_id)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Jobs ====================
|
||||
// ==================== JobStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl JobStore for PgBackend {
|
||||
async fn save_job(&self, ctx: &JobContext) -> Result<(), DatabaseError> {
|
||||
self.store.save_job(ctx).await
|
||||
}
|
||||
@@ -204,8 +215,6 @@ impl Database for PgBackend {
|
||||
self.store.get_stuck_jobs().await
|
||||
}
|
||||
|
||||
// ==================== Actions ====================
|
||||
|
||||
async fn save_action(&self, job_id: Uuid, action: &ActionRecord) -> Result<(), DatabaseError> {
|
||||
self.store.save_action(job_id, action).await
|
||||
}
|
||||
@@ -214,14 +223,10 @@ impl Database for PgBackend {
|
||||
self.store.get_job_actions(job_id).await
|
||||
}
|
||||
|
||||
// ==================== LLM Calls ====================
|
||||
|
||||
async fn record_llm_call(&self, record: &LlmCallRecord<'_>) -> Result<Uuid, DatabaseError> {
|
||||
self.store.record_llm_call(record).await
|
||||
}
|
||||
|
||||
// ==================== Estimation Snapshots ====================
|
||||
|
||||
async fn save_estimation_snapshot(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
@@ -254,9 +259,12 @@ impl Database for PgBackend {
|
||||
.update_estimation_actuals(id, actual_cost, actual_time_secs, actual_value)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Sandbox Jobs ====================
|
||||
// ==================== SandboxStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl SandboxStore for PgBackend {
|
||||
async fn save_sandbox_job(&self, job: &SandboxJobRecord) -> Result<(), DatabaseError> {
|
||||
self.store.save_sandbox_job(job).await
|
||||
}
|
||||
@@ -323,8 +331,6 @@ impl Database for PgBackend {
|
||||
self.store.get_sandbox_job_mode(id).await
|
||||
}
|
||||
|
||||
// ==================== Job Events ====================
|
||||
|
||||
async fn save_job_event(
|
||||
&self,
|
||||
job_id: Uuid,
|
||||
@@ -341,9 +347,12 @@ impl Database for PgBackend {
|
||||
) -> Result<Vec<JobEventRecord>, DatabaseError> {
|
||||
self.store.list_job_events(job_id, limit).await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Routines ====================
|
||||
// ==================== RoutineStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl RoutineStore for PgBackend {
|
||||
async fn create_routine(&self, routine: &Routine) -> Result<(), DatabaseError> {
|
||||
self.store.create_routine(routine).await
|
||||
}
|
||||
@@ -401,8 +410,6 @@ impl Database for PgBackend {
|
||||
self.store.delete_routine(id).await
|
||||
}
|
||||
|
||||
// ==================== Routine Runs ====================
|
||||
|
||||
async fn create_routine_run(&self, run: &RoutineRun) -> Result<(), DatabaseError> {
|
||||
self.store.create_routine_run(run).await
|
||||
}
|
||||
@@ -430,9 +437,12 @@ impl Database for PgBackend {
|
||||
async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result<i64, DatabaseError> {
|
||||
self.store.count_running_routine_runs(routine_id).await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Tool Failures ====================
|
||||
// ==================== ToolFailureStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl ToolFailureStore for PgBackend {
|
||||
async fn record_tool_failure(
|
||||
&self,
|
||||
tool_name: &str,
|
||||
@@ -454,9 +464,12 @@ impl Database for PgBackend {
|
||||
async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError> {
|
||||
self.store.increment_repair_attempts(tool_name).await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Settings ====================
|
||||
// ==================== SettingsStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl SettingsStore for PgBackend {
|
||||
async fn get_setting(
|
||||
&self,
|
||||
user_id: &str,
|
||||
@@ -508,9 +521,12 @@ impl Database for PgBackend {
|
||||
async fn has_settings(&self, user_id: &str) -> Result<bool, DatabaseError> {
|
||||
self.store.has_settings(user_id).await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Workspace: Documents ====================
|
||||
// ==================== WorkspaceStore ====================
|
||||
|
||||
#[async_trait]
|
||||
impl WorkspaceStore for PgBackend {
|
||||
async fn get_document_by_path(
|
||||
&self,
|
||||
user_id: &str,
|
||||
@@ -577,8 +593,6 @@ impl Database for PgBackend {
|
||||
self.repo.list_documents(user_id, agent_id).await
|
||||
}
|
||||
|
||||
// ==================== Workspace: Chunks ====================
|
||||
|
||||
async fn delete_chunks(&self, document_id: Uuid) -> Result<(), WorkspaceError> {
|
||||
self.repo.delete_chunks(document_id).await
|
||||
}
|
||||
@@ -614,8 +628,6 @@ impl Database for PgBackend {
|
||||
.await
|
||||
}
|
||||
|
||||
// ==================== Workspace: Search ====================
|
||||
|
||||
async fn hybrid_search(
|
||||
&self,
|
||||
user_id: &str,
|
||||
|
||||
Reference in New Issue
Block a user