003-knowledge-memory #55

Merged
cameron merged 12 commits from 003-knowledge-memory into master 2026-04-21 01:54:34 +00:00
22 changed files with 2965 additions and 192 deletions

View File

@@ -0,0 +1,19 @@
-- Restore original schema, retaining only the current insight per file.
CREATE TABLE photo_insights_old (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
file_path TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
summary TEXT NOT NULL,
generated_at BIGINT NOT NULL,
model_version TEXT NOT NULL
);
INSERT INTO photo_insights_old (id, file_path, title, summary, generated_at, model_version)
SELECT id, file_path, title, summary, generated_at, model_version
FROM photo_insights
WHERE is_current = 1;
DROP TABLE photo_insights;
ALTER TABLE photo_insights_old RENAME TO photo_insights;
CREATE INDEX IF NOT EXISTS idx_photo_insights_path ON photo_insights(file_path);

View File

@@ -0,0 +1,25 @@
-- Convert photo_insights to an append-only history table.
-- SQLite cannot drop a UNIQUE constraint via ALTER TABLE, so we recreate the table.
-- This preserves existing insight IDs so that future entity_facts.source_insight_id
-- FK references remain valid.
CREATE TABLE photo_insights_new (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
file_path TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL,
generated_at BIGINT NOT NULL,
model_version TEXT NOT NULL,
is_current BOOLEAN NOT NULL DEFAULT 0
);
-- Migrate existing rows; mark them all as current (one row per path currently).
INSERT INTO photo_insights_new (id, file_path, title, summary, generated_at, model_version, is_current)
SELECT id, file_path, title, summary, generated_at, model_version, 1
FROM photo_insights;
DROP TABLE photo_insights;
ALTER TABLE photo_insights_new RENAME TO photo_insights;
CREATE INDEX idx_photo_insights_file_path ON photo_insights(file_path);
CREATE INDEX idx_photo_insights_current ON photo_insights(file_path, is_current);

View File

@@ -0,0 +1,3 @@
DROP TABLE IF EXISTS entity_photo_links;
DROP TABLE IF EXISTS entity_facts;
DROP TABLE IF EXISTS entities;

View File

@@ -0,0 +1,55 @@
-- Entity-relationship knowledge memory tables.
-- Entities are the nodes (people, places, events, things).
-- entity_facts are typed claims about or between entities.
-- entity_photo_links connect entities to specific photos.
CREATE TABLE entities (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
name TEXT NOT NULL,
entity_type TEXT NOT NULL, -- 'person' | 'place' | 'event' | 'thing'
description TEXT NOT NULL DEFAULT '',
embedding BLOB, -- 768-dim f32 vector; nullable if embedding service was unavailable
confidence REAL NOT NULL DEFAULT 0.5,
status TEXT NOT NULL DEFAULT 'active', -- 'active' | 'reviewed' | 'rejected'
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
UNIQUE(name, entity_type)
);
CREATE INDEX idx_entities_type ON entities(entity_type);
CREATE INDEX idx_entities_status ON entities(status);
CREATE INDEX idx_entities_name ON entities(name);
CREATE TABLE entity_facts (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
subject_entity_id INTEGER NOT NULL,
predicate TEXT NOT NULL,
object_entity_id INTEGER, -- nullable: entity-to-entity relationship target
object_value TEXT, -- nullable: free-text attribute value
source_photo TEXT, -- photo path that prompted extraction (injected server-side)
source_insight_id INTEGER, -- backfilled after insight is stored
confidence REAL NOT NULL DEFAULT 0.6,
status TEXT NOT NULL DEFAULT 'active', -- 'active' | 'reviewed' | 'rejected'
created_at BIGINT NOT NULL,
CONSTRAINT fk_ef_subject FOREIGN KEY (subject_entity_id) REFERENCES entities(id) ON DELETE CASCADE,
CONSTRAINT fk_ef_object FOREIGN KEY (object_entity_id) REFERENCES entities(id) ON DELETE SET NULL,
CONSTRAINT fk_ef_insight FOREIGN KEY (source_insight_id) REFERENCES photo_insights(id) ON DELETE SET NULL,
CHECK (object_entity_id IS NOT NULL OR object_value IS NOT NULL)
);
CREATE INDEX idx_entity_facts_subject ON entity_facts(subject_entity_id);
CREATE INDEX idx_entity_facts_predicate ON entity_facts(predicate);
CREATE INDEX idx_entity_facts_status ON entity_facts(status);
CREATE INDEX idx_entity_facts_source_photo ON entity_facts(source_photo);
CREATE TABLE entity_photo_links (
id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
entity_id INTEGER NOT NULL,
file_path TEXT NOT NULL,
role TEXT NOT NULL, -- 'subject' | 'location' | 'event' | 'thing'
CONSTRAINT fk_epl_entity FOREIGN KEY (entity_id) REFERENCES entities(id) ON DELETE CASCADE,
UNIQUE(entity_id, file_path, role)
);
CREATE INDEX idx_entity_photo_links_entity ON entity_photo_links(entity_id);
CREATE INDEX idx_entity_photo_links_photo ON entity_photo_links(file_path);

View File

@@ -0,0 +1,14 @@
-- SQLite doesn't support DROP COLUMN directly, so we recreate the table
CREATE TABLE photo_insights_backup AS SELECT id, file_path, title, summary, generated_at, model_version, is_current FROM photo_insights;
DROP TABLE photo_insights;
CREATE TABLE photo_insights (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
file_path TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL,
generated_at BIGINT NOT NULL,
model_version TEXT NOT NULL,
is_current BOOLEAN NOT NULL DEFAULT TRUE
);
INSERT INTO photo_insights SELECT * FROM photo_insights_backup;
DROP TABLE photo_insights_backup;

View File

@@ -0,0 +1,2 @@
ALTER TABLE photo_insights ADD COLUMN training_messages TEXT;
ALTER TABLE photo_insights ADD COLUMN approved BOOLEAN;

View File

@@ -18,6 +18,14 @@ pub struct GeneratePhotoInsightRequest {
pub system_prompt: Option<String>, pub system_prompt: Option<String>,
#[serde(default)] #[serde(default)]
pub num_ctx: Option<i32>, pub num_ctx: Option<i32>,
#[serde(default)]
pub temperature: Option<f32>,
#[serde(default)]
pub top_p: Option<f32>,
#[serde(default)]
pub top_k: Option<i32>,
#[serde(default)]
pub min_p: Option<f32>,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
@@ -25,6 +33,18 @@ pub struct GetPhotoInsightQuery {
pub path: String, pub path: String,
} }
#[derive(Debug, Deserialize)]
pub struct RateInsightRequest {
pub file_path: String,
pub approved: bool,
}
#[derive(Debug, Deserialize)]
pub struct ExportTrainingDataQuery {
#[serde(default)]
pub approved_only: Option<bool>,
}
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
pub struct PhotoInsightResponse { pub struct PhotoInsightResponse {
pub id: i32, pub id: i32,
@@ -33,6 +53,12 @@ pub struct PhotoInsightResponse {
pub summary: String, pub summary: String,
pub generated_at: i64, pub generated_at: i64,
pub model_version: String, pub model_version: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub prompt_eval_count: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub eval_count: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub approved: Option<bool>,
} }
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
@@ -90,6 +116,10 @@ pub async fn generate_insight_handler(
request.model.clone(), request.model.clone(),
request.system_prompt.clone(), request.system_prompt.clone(),
request.num_ctx, request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
) )
.await; .await;
@@ -133,6 +163,9 @@ pub async fn get_insight_handler(
summary: insight.summary, summary: insight.summary,
generated_at: insight.generated_at, generated_at: insight.generated_at,
model_version: insight.model_version, model_version: insight.model_version,
prompt_eval_count: None,
eval_count: None,
approved: insight.approved,
}; };
HttpResponse::Ok().json(response) HttpResponse::Ok().json(response)
} }
@@ -197,6 +230,9 @@ pub async fn get_all_insights_handler(
summary: insight.summary, summary: insight.summary,
generated_at: insight.generated_at, generated_at: insight.generated_at,
model_version: insight.model_version, model_version: insight.model_version,
prompt_eval_count: None,
eval_count: None,
approved: insight.approved,
}) })
.collect(); .collect();
@@ -241,7 +277,7 @@ pub async fn generate_agentic_insight_handler(
let max_iterations: usize = std::env::var("AGENTIC_MAX_ITERATIONS") let max_iterations: usize = std::env::var("AGENTIC_MAX_ITERATIONS")
.ok() .ok()
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
.unwrap_or(10); .unwrap_or(12);
span.set_attribute(KeyValue::new("max_iterations", max_iterations as i64)); span.set_attribute(KeyValue::new("max_iterations", max_iterations as i64));
@@ -258,12 +294,16 @@ pub async fn generate_agentic_insight_handler(
request.model.clone(), request.model.clone(),
request.system_prompt.clone(), request.system_prompt.clone(),
request.num_ctx, request.num_ctx,
request.temperature,
request.top_p,
request.top_k,
request.min_p,
max_iterations, max_iterations,
) )
.await; .await;
match result { match result {
Ok(()) => { Ok((prompt_eval_count, eval_count)) => {
span.set_status(Status::Ok); span.set_status(Status::Ok);
// Fetch the stored insight to return it // Fetch the stored insight to return it
let otel_context = opentelemetry::Context::new(); let otel_context = opentelemetry::Context::new();
@@ -277,6 +317,9 @@ pub async fn generate_agentic_insight_handler(
summary: insight.summary, summary: insight.summary,
generated_at: insight.generated_at, generated_at: insight.generated_at,
model_version: insight.model_version, model_version: insight.model_version,
prompt_eval_count,
eval_count,
approved: insight.approved,
}; };
HttpResponse::Ok().json(response) HttpResponse::Ok().json(response)
} }
@@ -367,3 +410,86 @@ pub async fn get_available_models_handler(
HttpResponse::Ok().json(response) HttpResponse::Ok().json(response)
} }
/// POST /insights/rate - Rate an insight (thumbs up/down for training data)
#[post("/insights/rate")]
pub async fn rate_insight_handler(
_claims: Claims,
request: web::Json<RateInsightRequest>,
insight_dao: web::Data<std::sync::Mutex<Box<dyn InsightDao>>>,
) -> impl Responder {
let normalized_path = normalize_path(&request.file_path);
log::info!(
"Rating insight for {}: approved={}",
normalized_path,
request.approved
);
let otel_context = opentelemetry::Context::new();
let mut dao = insight_dao.lock().expect("Unable to lock InsightDao");
match dao.rate_insight(&otel_context, &normalized_path, request.approved) {
Ok(()) => HttpResponse::Ok().json(serde_json::json!({
"success": true,
"message": "Insight rated successfully"
})),
Err(e) => {
log::error!("Failed to rate insight: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({
"error": format!("Failed to rate insight: {:?}", e)
}))
}
}
}
/// GET /insights/training-data - Export approved training data as JSONL
#[get("/insights/training-data")]
pub async fn export_training_data_handler(
_claims: Claims,
query: web::Query<ExportTrainingDataQuery>,
insight_dao: web::Data<std::sync::Mutex<Box<dyn InsightDao>>>,
) -> impl Responder {
let approved_only = query.approved_only.unwrap_or(true);
log::info!("Exporting training data (approved_only={})", approved_only);
let otel_context = opentelemetry::Context::new();
let mut dao = insight_dao.lock().expect("Unable to lock InsightDao");
let insights = if approved_only {
dao.get_approved_insights(&otel_context)
} else {
dao.get_all_insights(&otel_context)
};
match insights {
Ok(insights) => {
let mut jsonl = String::new();
for insight in &insights {
if let Some(ref messages) = insight.training_messages {
let entry = serde_json::json!({
"file_path": insight.file_path,
"model_version": insight.model_version,
"generated_at": insight.generated_at,
"title": insight.title,
"summary": insight.summary,
"messages": serde_json::from_str::<serde_json::Value>(messages)
.unwrap_or(serde_json::Value::Null),
});
jsonl.push_str(&entry.to_string());
jsonl.push('\n');
}
}
HttpResponse::Ok()
.content_type("application/jsonl")
.insert_header(("Content-Disposition", "attachment; filename=\"training_data.jsonl\""))
.body(jsonl)
}
Err(e) => {
log::error!("Failed to export training data: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({
"error": format!("Failed to export training data: {:?}", e)
}))
}
}
}

View File

@@ -1,6 +1,6 @@
use anyhow::Result; use anyhow::Result;
use base64::Engine as _; use base64::Engine as _;
use chrono::{DateTime, NaiveDate, Utc}; use chrono::{DateTime, Local, NaiveDate, Utc};
use image::ImageFormat; use image::ImageFormat;
use opentelemetry::KeyValue; use opentelemetry::KeyValue;
use opentelemetry::trace::{Span, Status, TraceContextExt, Tracer}; use opentelemetry::trace::{Span, Status, TraceContextExt, Tracer};
@@ -13,7 +13,8 @@ use crate::ai::ollama::{ChatMessage, OllamaClient, Tool};
use crate::ai::sms_client::SmsApiClient; use crate::ai::sms_client::SmsApiClient;
use crate::database::models::InsertPhotoInsight; use crate::database::models::InsertPhotoInsight;
use crate::database::{ use crate::database::{
CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, LocationHistoryDao, SearchHistoryDao, CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, KnowledgeDao, LocationHistoryDao,
SearchHistoryDao,
}; };
use crate::memories::extract_date_from_filename; use crate::memories::extract_date_from_filename;
use crate::otel::global_tracer; use crate::otel::global_tracer;
@@ -48,6 +49,9 @@ pub struct InsightGenerator {
search_dao: Arc<Mutex<Box<dyn SearchHistoryDao>>>, search_dao: Arc<Mutex<Box<dyn SearchHistoryDao>>>,
tag_dao: Arc<Mutex<Box<dyn TagDao>>>, tag_dao: Arc<Mutex<Box<dyn TagDao>>>,
// Knowledge memory
knowledge_dao: Arc<Mutex<Box<dyn KnowledgeDao>>>,
base_path: String, base_path: String,
} }
@@ -62,6 +66,7 @@ impl InsightGenerator {
location_dao: Arc<Mutex<Box<dyn LocationHistoryDao>>>, location_dao: Arc<Mutex<Box<dyn LocationHistoryDao>>>,
search_dao: Arc<Mutex<Box<dyn SearchHistoryDao>>>, search_dao: Arc<Mutex<Box<dyn SearchHistoryDao>>>,
tag_dao: Arc<Mutex<Box<dyn TagDao>>>, tag_dao: Arc<Mutex<Box<dyn TagDao>>>,
knowledge_dao: Arc<Mutex<Box<dyn KnowledgeDao>>>,
base_path: String, base_path: String,
) -> Self { ) -> Self {
Self { Self {
@@ -74,6 +79,7 @@ impl InsightGenerator {
location_dao, location_dao,
search_dao, search_dao,
tag_dao, tag_dao,
knowledge_dao,
base_path, base_path,
} }
} }
@@ -638,6 +644,10 @@ impl InsightGenerator {
custom_model: Option<String>, custom_model: Option<String>,
custom_system_prompt: Option<String>, custom_system_prompt: Option<String>,
num_ctx: Option<i32>, num_ctx: Option<i32>,
temperature: Option<f32>,
top_p: Option<f32>,
top_k: Option<i32>,
min_p: Option<f32>,
) -> Result<()> { ) -> Result<()> {
let tracer = global_tracer(); let tracer = global_tracer();
let current_cx = opentelemetry::Context::current(); let current_cx = opentelemetry::Context::current();
@@ -671,6 +681,30 @@ impl InsightGenerator {
ollama_client.set_num_ctx(Some(ctx)); ollama_client.set_num_ctx(Some(ctx));
} }
// Apply sampling parameters if any were provided
if temperature.is_some() || top_p.is_some() || top_k.is_some() || min_p.is_some() {
log::info!(
"Using sampling params — temperature: {:?}, top_p: {:?}, top_k: {:?}, min_p: {:?}",
temperature,
top_p,
top_k,
min_p
);
if let Some(t) = temperature {
span.set_attribute(KeyValue::new("temperature", t as f64));
}
if let Some(p) = top_p {
span.set_attribute(KeyValue::new("top_p", p as f64));
}
if let Some(k) = top_k {
span.set_attribute(KeyValue::new("top_k", k as i64));
}
if let Some(m) = min_p {
span.set_attribute(KeyValue::new("min_p", m as f64));
}
ollama_client.set_sampling_params(temperature, top_p, top_k, min_p);
}
// Create context with this span for child operations // Create context with this span for child operations
let insight_cx = current_cx.with_span(span); let insight_cx = current_cx.with_span(span);
@@ -1158,6 +1192,8 @@ impl InsightGenerator {
summary, summary,
generated_at: Utc::now().timestamp(), generated_at: Utc::now().timestamp(),
model_version: ollama_client.primary_model.clone(), model_version: ollama_client.primary_model.clone(),
is_current: true,
training_messages: None,
}; };
let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
@@ -1314,7 +1350,7 @@ Return ONLY the summary, nothing else."#,
- Emotional tone and relationship dynamics - Emotional tone and relationship dynamics
- Any significant details that provide context about what was happening - Any significant details that provide context about what was happening
Be thorough but organized. Use 1-2 paragraphs. Be thorough but organized.
Messages: Messages:
{} {}
@@ -1345,6 +1381,7 @@ Return ONLY the summary, nothing else."#,
arguments: &serde_json::Value, arguments: &serde_json::Value,
ollama: &OllamaClient, ollama: &OllamaClient,
image_base64: &Option<String>, image_base64: &Option<String>,
file_path: &str,
cx: &opentelemetry::Context, cx: &opentelemetry::Context,
) -> String { ) -> String {
let result = match tool_name { let result = match tool_name {
@@ -1355,16 +1392,17 @@ Return ONLY the summary, nothing else."#,
"get_file_tags" => self.tool_get_file_tags(arguments, cx).await, "get_file_tags" => self.tool_get_file_tags(arguments, cx).await,
"describe_photo" => self.tool_describe_photo(ollama, image_base64).await, "describe_photo" => self.tool_describe_photo(ollama, image_base64).await,
"reverse_geocode" => self.tool_reverse_geocode(arguments).await, "reverse_geocode" => self.tool_reverse_geocode(arguments).await,
"recall_entities" => self.tool_recall_entities(arguments, cx).await,
"recall_facts_for_photo" => self.tool_recall_facts_for_photo(arguments, cx).await,
"store_entity" => self.tool_store_entity(arguments, ollama, cx).await,
"store_fact" => self.tool_store_fact(arguments, file_path, cx).await,
"get_current_datetime" => Self::tool_get_current_datetime(),
unknown => format!("Unknown tool: {}", unknown), unknown => format!("Unknown tool: {}", unknown),
}; };
if result.starts_with("Error") || result.starts_with("No ") { if result.starts_with("Error") || result.starts_with("No ") {
log::warn!("Tool '{}' result: {}", tool_name, result); log::warn!("Tool '{}' result: {}", tool_name, result);
} else { } else {
log::info!( log::info!("Tool '{}' result: {} chars", tool_name, result.len());
"Tool '{}' result: {} chars",
tool_name,
result.len()
);
} }
result result
} }
@@ -1679,6 +1717,352 @@ Return ONLY the summary, nothing else."#,
} }
} }
/// Tool: recall_entities — search the knowledge memory for known entities
async fn tool_recall_entities(
&self,
args: &serde_json::Value,
cx: &opentelemetry::Context,
) -> String {
use crate::database::EntityFilter;
let name_search = args
.get("name")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let entity_type = args
.get("entity_type")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
let limit = args.get("limit").and_then(|v| v.as_i64()).unwrap_or(10);
log::info!(
"tool_recall_entities: name={:?}, type={:?}, limit={}",
name_search,
entity_type,
limit
);
let filter = EntityFilter {
entity_type,
status: Some("active".to_string()),
search: name_search,
limit,
offset: 0,
};
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
match kdao.list_entities(cx, filter) {
Ok((entities, _total)) if entities.is_empty() => {
"No known entities found matching the query.".to_string()
}
Ok((entities, _total)) => {
let lines: Vec<String> = entities
.iter()
.map(|e| {
format!(
"ID:{} | {} | {} | {} | confidence:{:.2}",
e.id, e.entity_type, e.name, e.description, e.confidence
)
})
.collect();
format!("Known entities:\n{}", lines.join("\n"))
}
Err(e) => format!("Error recalling entities: {:?}", e),
}
}
/// Tool: recall_facts_for_photo — retrieve facts linked to a specific photo
async fn tool_recall_facts_for_photo(
&self,
args: &serde_json::Value,
cx: &opentelemetry::Context,
) -> String {
let file_path = match args.get("file_path").and_then(|v| v.as_str()) {
Some(p) => p.to_string(),
None => return "Error: missing required parameter 'file_path'".to_string(),
};
log::info!("tool_recall_facts_for_photo: file_path={}", file_path);
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
// Fetch photo links to find which entities appear in this photo
let links = match kdao.get_links_for_photo(cx, &file_path) {
Ok(l) => l,
Err(e) => return format!("Error fetching photo links: {:?}", e),
};
if links.is_empty() {
return "No knowledge facts found for this photo.".to_string();
}
let mut output_lines = Vec::new();
let entity_ids: Vec<i32> = links.iter().map(|l| l.entity_id).collect();
// For each linked entity, fetch its facts
for entity_id in entity_ids {
if let Ok(entity) = kdao.get_entity_by_id(cx, entity_id) {
if let Some(e) = entity {
let role = links
.iter()
.find(|l| l.entity_id == entity_id)
.map(|l| l.role.as_str())
.unwrap_or("subject");
output_lines.push(format!(
"Entity: {} ({}, role: {})",
e.name, e.entity_type, role
));
if let Ok(facts) = kdao.get_facts_for_entity(cx, entity_id) {
for f in facts.iter().filter(|f| f.status == "active") {
let obj = if let Some(ref v) = f.object_value {
v.clone()
} else if let Some(oid) = f.object_entity_id {
kdao.get_entity_by_id(cx, oid)
.ok()
.flatten()
.map(|e| format!("{} (entity ID: {})", e.name, e.id))
.unwrap_or_else(|| format!("entity:{}", oid))
} else {
"(unknown)".to_string()
};
output_lines.push(format!(" - {} {}", f.predicate, obj));
}
}
}
}
}
if output_lines.is_empty() {
"No active knowledge facts found for this photo.".to_string()
} else {
format!("Knowledge for this photo:\n{}", output_lines.join("\n"))
}
}
/// Tool: store_entity — upsert an entity into the knowledge memory
async fn tool_store_entity(
&self,
args: &serde_json::Value,
ollama: &OllamaClient,
cx: &opentelemetry::Context,
) -> String {
use crate::database::models::InsertEntity;
let name = match args.get("name").and_then(|v| v.as_str()) {
Some(n) => n.to_string(),
None => return "Error: missing required parameter 'name'".to_string(),
};
let entity_type = match args.get("entity_type").and_then(|v| v.as_str()) {
Some(t) => t.to_string(),
None => return "Error: missing required parameter 'entity_type'".to_string(),
};
let description = args
.get("description")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
log::info!(
"tool_store_entity: name='{}', type='{}', description='{}'",
name,
entity_type,
description
);
// Pre-flight similarity check — surface near-duplicates to the model
// before it commits to a new entity. Uses the first name token as the
// search term so "Sarah" matches when storing "Sarah Johnson" and vice
// versa. Exact-name matches are excluded (upsert_entity deduplicates
// those already). Results are appended to the tool response so the
// model can choose to use an existing entity's ID instead.
let similar_entities: Vec<String> = {
use crate::database::{EntityFilter, KnowledgeDao};
use crate::database::knowledge_dao::normalize_entity_type;
let normalised_type = normalize_entity_type(&entity_type);
let first_token = name
.split_whitespace()
.next()
.unwrap_or(&name)
.to_string();
let filter = EntityFilter {
entity_type: None, // search all types, filter client-side to avoid case issues
status: Some("active".to_string()),
search: Some(first_token),
limit: 10,
offset: 0,
};
let mut kdao = self.knowledge_dao.lock().expect("Unable to lock KnowledgeDao");
kdao.list_entities(cx, filter)
.unwrap_or_default()
.0
.into_iter()
.filter(|e| {
normalize_entity_type(&e.entity_type) == normalised_type
&& e.name.to_lowercase() != name.to_lowercase()
})
.map(|e| format!(" ID:{} | {} | {}", e.id, e.name, e.description))
.collect()
};
// Generate embedding for name + description (best-effort)
let embed_text = format!("{} {}", name, description);
let embedding: Option<Vec<u8>> = match ollama.generate_embedding(&embed_text).await {
Ok(vec) => {
let bytes: Vec<u8> = vec.iter().flat_map(|f| f.to_le_bytes()).collect();
Some(bytes)
}
Err(e) => {
log::warn!("Embedding generation failed for entity '{}': {}", name, e);
None
}
};
let now = chrono::Utc::now().timestamp();
let insert = InsertEntity {
name,
entity_type,
description,
embedding,
confidence: 0.6,
status: "active".to_string(),
created_at: now,
updated_at: now,
};
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
match kdao.upsert_entity(cx, insert) {
Ok(entity) => {
let mut response = format!(
"Entity stored: ID:{} | {} | {} | confidence:{:.2}",
entity.id, entity.entity_type, entity.name, entity.confidence
);
if !similar_entities.is_empty() {
response.push_str(
"\nSimilar existing entities found — verify this is not a duplicate:\n",
);
response.push_str(&similar_entities.join("\n"));
response.push_str(
"\nIf one of these is the same entity, use their existing ID in store_fact instead of the newly created one.",
);
}
response
}
Err(e) => format!("Error storing entity: {:?}", e),
}
}
/// Tool: store_fact — record a fact about an entity, linked to the current photo
async fn tool_store_fact(
&self,
args: &serde_json::Value,
file_path: &str,
cx: &opentelemetry::Context,
) -> String {
use crate::database::models::{InsertEntityFact, InsertEntityPhotoLink};
let subject_entity_id = match args.get("subject_entity_id").and_then(|v| v.as_i64()) {
Some(id) => id as i32,
None => return "Error: missing required parameter 'subject_entity_id'".to_string(),
};
let predicate = match args.get("predicate").and_then(|v| v.as_str()) {
Some(p) => p.to_string(),
None => return "Error: missing required parameter 'predicate'".to_string(),
};
let object_entity_id = args
.get("object_entity_id")
.and_then(|v| v.as_i64())
.map(|id| id as i32);
let object_value = args
.get("object_value")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
if object_entity_id.is_none() && object_value.is_none() {
return "Error: provide either object_entity_id or object_value".to_string();
}
let photo_role = args
.get("photo_role")
.and_then(|v| v.as_str())
.unwrap_or("subject")
.to_string();
log::info!(
"tool_store_fact: entity_id={}, predicate='{}', object_entity_id={:?}, object_value={:?}, photo='{}'",
subject_entity_id,
predicate,
object_entity_id,
object_value,
file_path
);
let fact = InsertEntityFact {
subject_entity_id,
predicate,
object_entity_id,
object_value,
source_photo: Some(file_path.to_string()),
source_insight_id: None, // will be backfilled after store_insight
confidence: 0.6,
status: "active".to_string(),
created_at: chrono::Utc::now().timestamp(),
};
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
// Upsert the fact (corroboration bumps confidence if duplicate)
let (stored_fact, is_new) = match kdao.upsert_fact(cx, fact) {
Ok(r) => r,
Err(e) => return format!("Error storing fact: {:?}", e),
};
// Upsert a photo link so this entity is associated with this photo
let link = InsertEntityPhotoLink {
entity_id: subject_entity_id,
file_path: file_path.to_string(),
role: photo_role,
};
if let Err(e) = kdao.upsert_photo_link(cx, link) {
log::warn!(
"Failed to upsert photo link for entity {}: {:?}",
subject_entity_id,
e
);
}
let action = if is_new {
"Stored new fact"
} else {
"Corroborated existing fact"
};
format!(
"{}: ID:{} | confidence:{:.2}",
action, stored_fact.id, stored_fact.confidence
)
}
/// Tool: get_current_datetime — returns the current local date and time
fn tool_get_current_datetime() -> String {
let now = Local::now();
format!(
"Current date/time: {} ({})",
now.format("%Y-%m-%d %H:%M:%S %Z"),
now.format("%A")
)
}
// ── Agentic insight generation ────────────────────────────────────── // ── Agentic insight generation ──────────────────────────────────────
/// Build the list of tool definitions for the agentic loop /// Build the list of tool definitions for the agentic loop
@@ -1799,6 +2183,109 @@ Return ONLY the summary, nothing else."#,
}), }),
)); ));
// Knowledge memory tools
tools.push(Tool::function(
"recall_entities",
"Search the knowledge memory for people, places, events, or things previously learned from other photos. Use this to retrieve context about subjects appearing in this photo.",
serde_json::json!({
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Name or partial name to search for (case-insensitive substring match)"
},
"entity_type": {
"type": "string",
"enum": ["person", "place", "event", "thing"],
"description": "Filter by entity type (optional)"
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return (default: 10)"
}
}
}),
));
tools.push(Tool::function(
"recall_facts_for_photo",
"Retrieve all known facts linked to a specific photo from the knowledge memory. Use this at the start of insight generation to load any previously stored knowledge about subjects in this photo.",
serde_json::json!({
"type": "object",
"required": ["file_path"],
"properties": {
"file_path": {
"type": "string",
"description": "The file path of the photo to retrieve facts for"
}
}
}),
));
tools.push(Tool::function(
"store_entity",
"Store or update a person, place, event, or thing in the knowledge memory. Call this when you identify a subject in this photo that should be remembered for future insights.",
serde_json::json!({
"type": "object",
"required": ["name", "entity_type"],
"properties": {
"name": {
"type": "string",
"description": "The canonical name of the entity (e.g. 'John Smith', 'Banff National Park')"
},
"entity_type": {
"type": "string",
"enum": ["person", "place", "event", "thing"],
"description": "The type of entity"
},
"description": {
"type": "string",
"description": "A brief description of the entity"
}
}
}),
));
tools.push(Tool::function(
"store_fact",
"Record a fact about an entity in the knowledge memory. Provide EITHER object_entity_id (when the object is a known entity whose ID you have) OR object_value (for free-text attributes). The fact will be linked to the current photo automatically.",
serde_json::json!({
"type": "object",
"required": ["subject_entity_id", "predicate"],
"properties": {
"subject_entity_id": {
"type": "integer",
"description": "The ID of the entity this fact is about (returned by store_entity or recall_entities)"
},
"predicate": {
"type": "string",
"description": "The relationship or attribute (e.g. 'is_friend_of', 'located_in', 'attended_event', 'is_sibling_of')"
},
"object_entity_id": {
"type": "integer",
"description": "Use when the object is a known entity (e.g. Cameron's entity ID for 'is_friend_of Cameron'). Takes precedence over object_value."
},
"object_value": {
"type": "string",
"description": "Use for free-text attributes where the object is not a stored entity (e.g. 'Portland, Oregon', 'software engineer')"
},
"photo_role": {
"type": "string",
"description": "How this entity appears in the photo (e.g. 'subject', 'background', 'location'). Defaults to 'subject'."
}
}
}),
));
tools.push(Tool::function(
"get_current_datetime",
"Get the current date and time. Useful for understanding how long ago the photo was taken.",
serde_json::json!({
"type": "object",
"properties": {}
}),
));
if has_vision { if has_vision {
tools.push(Tool::function( tools.push(Tool::function(
"describe_photo", "describe_photo",
@@ -1821,8 +2308,12 @@ Return ONLY the summary, nothing else."#,
custom_model: Option<String>, custom_model: Option<String>,
custom_system_prompt: Option<String>, custom_system_prompt: Option<String>,
num_ctx: Option<i32>, num_ctx: Option<i32>,
temperature: Option<f32>,
top_p: Option<f32>,
top_k: Option<i32>,
min_p: Option<f32>,
max_iterations: usize, max_iterations: usize,
) -> Result<()> { ) -> Result<(Option<i32>, Option<i32>)> {
let tracer = global_tracer(); let tracer = global_tracer();
let current_cx = opentelemetry::Context::current(); let current_cx = opentelemetry::Context::current();
let mut span = tracer.start_with_context("ai.insight.generate_agentic", &current_cx); let mut span = tracer.start_with_context("ai.insight.generate_agentic", &current_cx);
@@ -1854,6 +2345,29 @@ Return ONLY the summary, nothing else."#,
ollama_client.set_num_ctx(Some(ctx)); ollama_client.set_num_ctx(Some(ctx));
} }
if temperature.is_some() || top_p.is_some() || top_k.is_some() || min_p.is_some() {
log::info!(
"Using sampling params — temperature: {:?}, top_p: {:?}, top_k: {:?}, min_p: {:?}",
temperature,
top_p,
top_k,
min_p
);
if let Some(t) = temperature {
span.set_attribute(KeyValue::new("temperature", t as f64));
}
if let Some(p) = top_p {
span.set_attribute(KeyValue::new("top_p", p as f64));
}
if let Some(k) = top_k {
span.set_attribute(KeyValue::new("top_k", k as i64));
}
if let Some(m) = min_p {
span.set_attribute(KeyValue::new("min_p", m as f64));
}
ollama_client.set_sampling_params(temperature, top_p, top_k, min_p);
}
let insight_cx = current_cx.with_span(span); let insight_cx = current_cx.with_span(span);
// 2a. Verify the model exists on at least one server before checking capabilities // 2a. Verify the model exists on at least one server before checking capabilities
@@ -1976,7 +2490,46 @@ Return ONLY the summary, nothing else."#,
.collect() .collect()
}; };
// 6. Load image if vision capable // 6. Clear existing entity-photo links for this file so the run starts fresh,
// and ensure the owner entity (Cameron) exists so the agent can reference it.
let cameron_entity_id: Option<i32> = {
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
if let Err(e) = kdao.delete_photo_links_for_file(&insight_cx, &file_path) {
log::warn!(
"Failed to clear entity_photo_links for {}: {:?}",
file_path,
e
);
}
// Upsert the owner entity so the agent always has a stable entity ID to reference.
let owner = crate::database::models::InsertEntity {
name: "Cameron".to_string(),
entity_type: "person".to_string(),
description: "The owner of this photo collection. All memories are written from Cameron's perspective.".to_string(),
embedding: None,
confidence: 1.0,
status: "active".to_string(),
created_at: Utc::now().timestamp(),
updated_at: Utc::now().timestamp(),
};
match kdao.upsert_entity(&insight_cx, owner) {
Ok(e) => {
log::info!("Cameron entity ID: {}", e.id);
Some(e.id)
}
Err(e) => {
log::warn!("Failed to upsert Cameron entity: {:?}", e);
None
}
}
};
// 7. Load image if vision capable
let image_base64 = if has_vision { let image_base64 = if has_vision {
match self.load_image_as_base64(&file_path) { match self.load_image_as_base64(&file_path) {
Ok(b64) => { Ok(b64) => {
@@ -1992,21 +2545,37 @@ Return ONLY the summary, nothing else."#,
None None
}; };
// 7. Build system message // 8. Build system message
let base_system = "You are a personal photo memory assistant helping to reconstruct a memory from a photo.\n\n\ let cameron_id_note = match cameron_entity_id {
Some(id) => format!(
"\n\nYour identity in the knowledge store: Cameron (entity ID: {}). \
When storing facts where you (Cameron) are the object — for example, someone is your friend, \
sibling, or colleague — use subject_entity_id for the other person and set object_value to \
\"Cameron\" (or use store_fact with the other person as subject). When storing facts about \
Cameron directly, use {} as the subject_entity_id.",
id, id
),
None => String::new(),
};
let base_system = format!(
"You are a personal photo memory assistant helping to reconstruct a memory from a photo.{cameron_id_note}\n\n\
IMPORTANT INSTRUCTIONS:\n\ IMPORTANT INSTRUCTIONS:\n\
1. You MUST call multiple tools to gather context BEFORE writing any final insight. Do not produce a final answer after only one or two tool calls.\n\ 1. You MUST call multiple tools to gather context BEFORE writing any final insight. Do not produce a final answer after only one or two tool calls.\n\
2. Always call ALL of the following tools that are relevant: search_rag (search conversation summaries), get_sms_messages (fetch nearby messages), get_calendar_events (check what was happening that day), get_location_history (find where this was taken), get_file_tags (retrieve tags).\n\ 2. When calling get_sms_messages and search_rag, always make at least one call WITHOUT a contact filter to capture what else was happening in Cameron's life around this date — other conversations, events, and activities provide important wider context even when a specific contact is known.\n\
3. Only produce your final insight AFTER you have gathered context from at least 3-4 tools.\n\ 3. Use recall_facts_for_photo to load any previously stored knowledge about subjects in this photo.\n\
4. If a tool returns no results, that is useful information — continue calling the remaining tools anyway.\n\ 4. Use recall_entities to look up known people, places, or things that appear in this photo.\n\
5. Your final insight must be written in first person as Cameron, in a journal/memoir style."; 5. When you identify people, places, events, or notable things in this photo: use store_entity to record them and store_fact to record key facts (relationships, roles, attributes). This builds a persistent memory for future insights.\n\
6. Only produce your final insight AFTER you have gathered context from at least 5-12 tool calls.\n\
7. If a tool returns no results, that is useful information — continue calling the remaining tools anyway.",
cameron_id_note = cameron_id_note
);
let system_content = if let Some(ref custom) = custom_system_prompt { let system_content = if let Some(ref custom) = custom_system_prompt {
format!("{}\n\n{}", custom, base_system) format!("{}\n\n{}", custom, base_system)
} else { } else {
base_system.to_string() base_system.to_string()
}; };
// 8. Build user message // 9. Build user message
let gps_info = exif let gps_info = exif
.as_ref() .as_ref()
.and_then(|e| { .and_then(|e| {
@@ -2030,14 +2599,14 @@ Return ONLY the summary, nothing else."#,
.unwrap_or_else(|| "Contact/Person: unknown".to_string()); .unwrap_or_else(|| "Contact/Person: unknown".to_string());
let user_content = format!( let user_content = format!(
"Please analyze this photo and gather context to write a personal journal-style insight.\n\n\ "Please analyze this photo and gather any relevant context from the surrounding weeks.\n\n\
Photo file path: {}\n\ Photo file path: {}\n\
Date taken: {}\n\ Date taken: {}\n\
{}\n\ {}\n\
{}\n\ {}\n\
{}\n\n\ {}\n\n\
Use the available tools to gather more context about this moment (messages, calendar events, location history, etc.), \ Use the available tools to gather more context about this moment (messages, calendar events, location history, etc.), \
then write a detailed personal insight with a title and summary. Write in first person as Cameron.", then write a detailed insight with a title and summary.",
file_path, file_path,
date_taken.format("%B %d, %Y"), date_taken.format("%B %d, %Y"),
contact_info, contact_info,
@@ -2045,10 +2614,10 @@ Return ONLY the summary, nothing else."#,
tags_info, tags_info,
); );
// 9. Define tools // 10. Define tools
let tools = Self::build_tool_definitions(has_vision); let tools = Self::build_tool_definitions(has_vision);
// 10. Build initial messages // 11. Build initial messages
let system_msg = ChatMessage::system(system_content); let system_msg = ChatMessage::system(system_content);
let mut user_msg = ChatMessage::user(user_content); let mut user_msg = ChatMessage::user(user_content);
if let Some(ref img) = image_base64 { if let Some(ref img) = image_base64 {
@@ -2057,21 +2626,26 @@ Return ONLY the summary, nothing else."#,
let mut messages = vec![system_msg, user_msg]; let mut messages = vec![system_msg, user_msg];
// 11. Agentic loop // 12. Agentic loop
let loop_span = tracer.start_with_context("ai.agentic.loop", &insight_cx); let loop_span = tracer.start_with_context("ai.agentic.loop", &insight_cx);
let loop_cx = insight_cx.with_span(loop_span); let loop_cx = insight_cx.with_span(loop_span);
let mut final_content = String::new(); let mut final_content = String::new();
let mut iterations_used = 0usize; let mut iterations_used = 0usize;
let mut last_prompt_eval_count: Option<i32> = None;
let mut last_eval_count: Option<i32> = None;
for iteration in 0..max_iterations { for iteration in 0..max_iterations {
iterations_used = iteration + 1; iterations_used = iteration + 1;
log::info!("Agentic iteration {}/{}", iteration + 1, max_iterations); log::info!("Agentic iteration {}/{}", iteration + 1, max_iterations);
let response = ollama_client let (response, prompt_tokens, eval_tokens) = ollama_client
.chat_with_tools(messages.clone(), tools.clone()) .chat_with_tools(messages.clone(), tools.clone())
.await?; .await?;
last_prompt_eval_count = prompt_tokens;
last_eval_count = eval_tokens;
// Sanitize tool call arguments before pushing back into history. // Sanitize tool call arguments before pushing back into history.
// Some models occasionally return non-object arguments (bool, string, null) // Some models occasionally return non-object arguments (bool, string, null)
// which Ollama rejects when they are re-sent in a subsequent request. // which Ollama rejects when they are re-sent in a subsequent request.
@@ -2107,6 +2681,7 @@ Return ONLY the summary, nothing else."#,
&tool_call.function.arguments, &tool_call.function.arguments,
&ollama_client, &ollama_client,
&image_base64, &image_base64,
&file_path,
&loop_cx, &loop_cx,
) )
.await; .await;
@@ -2129,8 +2704,14 @@ Return ONLY the summary, nothing else."#,
messages.push(ChatMessage::user( messages.push(ChatMessage::user(
"Based on the context gathered, please write the final photo insight: a title and a detailed personal summary. Write in first person as Cameron.", "Based on the context gathered, please write the final photo insight: a title and a detailed personal summary. Write in first person as Cameron.",
)); ));
let final_response = ollama_client.chat_with_tools(messages, vec![]).await?; let (final_response, prompt_tokens, eval_tokens) =
final_content = final_response.content; ollama_client
.chat_with_tools(messages.clone(), vec![])
.await?;
last_prompt_eval_count = prompt_tokens;
last_eval_count = eval_tokens;
final_content = final_response.content.clone();
messages.push(final_response);
} }
loop_cx loop_cx
@@ -2138,7 +2719,7 @@ Return ONLY the summary, nothing else."#,
.set_attribute(KeyValue::new("iterations_used", iterations_used as i64)); .set_attribute(KeyValue::new("iterations_used", iterations_used as i64));
loop_cx.span().set_status(Status::Ok); loop_cx.span().set_status(Status::Ok);
// 12. Generate title // 13. Generate title
let title = ollama_client let title = ollama_client
.generate_photo_title(&final_content, custom_system_prompt.as_deref()) .generate_photo_title(&final_content, custom_system_prompt.as_deref())
.await?; .await?;
@@ -2150,21 +2731,33 @@ Return ONLY the summary, nothing else."#,
&final_content[..final_content.len().min(200)] &final_content[..final_content.len().min(200)]
); );
// 13. Store // 14. Serialize the full message history for training data
let training_messages = match serde_json::to_string(&messages) {
Ok(json) => Some(json),
Err(e) => {
log::warn!("Failed to serialize training messages: {}", e);
None
}
};
// 15. Store insight (returns the persisted row including its new id)
let insight = InsertPhotoInsight { let insight = InsertPhotoInsight {
file_path: file_path.to_string(), file_path: file_path.to_string(),
title, title,
summary: final_content, summary: final_content,
generated_at: Utc::now().timestamp(), generated_at: Utc::now().timestamp(),
model_version: ollama_client.primary_model.clone(), model_version: ollama_client.primary_model.clone(),
is_current: true,
training_messages,
}; };
let stored = {
let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
let result = dao dao.store_insight(&insight_cx, insight)
.store_insight(&insight_cx, insight) .map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e))
.map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e)); };
match &result { match &stored {
Ok(_) => { Ok(_) => {
log::info!("Successfully stored agentic insight for {}", file_path); log::info!("Successfully stored agentic insight for {}", file_path);
insight_cx.span().set_status(Status::Ok); insight_cx.span().set_status(Status::Ok);
@@ -2175,8 +2768,25 @@ Return ONLY the summary, nothing else."#,
} }
} }
result?; let stored_insight = stored?;
Ok(())
// 16. Backfill source_insight_id on all facts recorded for this photo during the loop
{
let mut kdao = self
.knowledge_dao
.lock()
.expect("Unable to lock KnowledgeDao");
if let Err(e) = kdao.update_facts_insight_id(&insight_cx, &file_path, stored_insight.id)
{
log::warn!(
"Failed to backfill source_insight_id for {}: {:?}",
file_path,
e
);
}
}
Ok((last_prompt_eval_count, last_eval_count))
} }
/// Reverse geocode GPS coordinates to human-readable place names /// Reverse geocode GPS coordinates to human-readable place names

View File

@@ -8,8 +8,9 @@ pub mod sms_client;
#[allow(unused_imports)] #[allow(unused_imports)]
pub use daily_summary_job::{generate_daily_summaries, strip_summary_boilerplate}; pub use daily_summary_job::{generate_daily_summaries, strip_summary_boilerplate};
pub use handlers::{ pub use handlers::{
delete_insight_handler, generate_agentic_insight_handler, generate_insight_handler, delete_insight_handler, export_training_data_handler, generate_agentic_insight_handler,
get_all_insights_handler, get_available_models_handler, get_insight_handler, generate_insight_handler, get_all_insights_handler, get_available_models_handler,
get_insight_handler, rate_insight_handler,
}; };
pub use insight_generator::InsightGenerator; pub use insight_generator::InsightGenerator;
pub use ollama::{ModelCapabilities, OllamaClient}; pub use ollama::{ModelCapabilities, OllamaClient};

View File

@@ -46,6 +46,10 @@ pub struct OllamaClient {
pub primary_model: String, pub primary_model: String,
pub fallback_model: Option<String>, pub fallback_model: Option<String>,
num_ctx: Option<i32>, num_ctx: Option<i32>,
temperature: Option<f32>,
top_p: Option<f32>,
top_k: Option<i32>,
min_p: Option<f32>,
} }
impl OllamaClient { impl OllamaClient {
@@ -66,6 +70,10 @@ impl OllamaClient {
primary_model, primary_model,
fallback_model, fallback_model,
num_ctx: None, num_ctx: None,
temperature: None,
top_p: None,
top_k: None,
min_p: None,
} }
} }
@@ -73,6 +81,54 @@ impl OllamaClient {
self.num_ctx = num_ctx; self.num_ctx = num_ctx;
} }
/// Set sampling parameters for generation. `None` values leave the
/// server-side default in place.
pub fn set_sampling_params(
&mut self,
temperature: Option<f32>,
top_p: Option<f32>,
top_k: Option<i32>,
min_p: Option<f32>,
) {
self.temperature = temperature;
self.top_p = top_p;
self.top_k = top_k;
self.min_p = min_p;
}
/// Build an `OllamaOptions` payload from the currently configured fields.
/// Returns `None` if no options would be set, so the `options` field is
/// omitted from the request entirely.
fn build_options(&self) -> Option<OllamaOptions> {
if self.num_ctx.is_none()
&& self.temperature.is_none()
&& self.top_p.is_none()
&& self.top_k.is_none()
&& self.min_p.is_none()
{
None
} else {
Some(OllamaOptions {
num_ctx: self.num_ctx,
temperature: self.temperature,
top_p: self.top_p,
top_k: self.top_k,
min_p: self.min_p,
})
}
}
/// Replace the HTTP client with one using a custom request timeout.
/// Useful for slow models where the default 120s may be insufficient.
pub fn with_request_timeout(mut self, secs: u64) -> Self {
self.client = Client::builder()
.connect_timeout(Duration::from_secs(5))
.timeout(Duration::from_secs(secs))
.build()
.unwrap_or_else(|_| Client::new());
self
}
/// List available models on an Ollama server (cached for 15 minutes) /// List available models on an Ollama server (cached for 15 minutes)
pub async fn list_models(url: &str) -> Result<Vec<String>> { pub async fn list_models(url: &str) -> Result<Vec<String>> {
// Check cache first // Check cache first
@@ -258,7 +314,7 @@ impl OllamaClient {
prompt: prompt.to_string(), prompt: prompt.to_string(),
stream: false, stream: false,
system: system.map(|s| s.to_string()), system: system.map(|s| s.to_string()),
options: self.num_ctx.map(|ctx| OllamaOptions { num_ctx: Some(ctx) }), options: self.build_options(),
images, images,
}; };
@@ -507,7 +563,7 @@ Analyze the image and use specific details from both the visual content and the
&self, &self,
messages: Vec<ChatMessage>, messages: Vec<ChatMessage>,
tools: Vec<Tool>, tools: Vec<Tool>,
) -> Result<ChatMessage> { ) -> Result<(ChatMessage, Option<i32>, Option<i32>)> {
// Try primary server first // Try primary server first
log::info!( log::info!(
"Attempting chat_with_tools with primary server: {} (model: {})", "Attempting chat_with_tools with primary server: {} (model: {})",
@@ -519,9 +575,9 @@ Analyze the image and use specific details from both the visual content and the
.await; .await;
match primary_result { match primary_result {
Ok(response) => { Ok(result) => {
log::info!("Successfully got chat_with_tools response from primary server"); log::info!("Successfully got chat_with_tools response from primary server");
Ok(response) Ok(result)
} }
Err(e) => { Err(e) => {
log::warn!("Primary server chat_with_tools failed: {}", e); log::warn!("Primary server chat_with_tools failed: {}", e);
@@ -540,11 +596,11 @@ Analyze the image and use specific details from both the visual content and the
.try_chat_with_tools(fallback_url, messages, tools) .try_chat_with_tools(fallback_url, messages, tools)
.await .await
{ {
Ok(response) => { Ok(result) => {
log::info!( log::info!(
"Successfully got chat_with_tools response from fallback server" "Successfully got chat_with_tools response from fallback server"
); );
Ok(response) Ok(result)
} }
Err(fallback_e) => { Err(fallback_e) => {
log::error!( log::error!(
@@ -571,7 +627,7 @@ Analyze the image and use specific details from both the visual content and the
base_url: &str, base_url: &str,
messages: Vec<ChatMessage>, messages: Vec<ChatMessage>,
tools: Vec<Tool>, tools: Vec<Tool>,
) -> Result<ChatMessage> { ) -> Result<(ChatMessage, Option<i32>, Option<i32>)> {
let url = format!("{}/api/chat", base_url); let url = format!("{}/api/chat", base_url);
let model = if base_url == self.primary_url { let model = if base_url == self.primary_url {
&self.primary_model &self.primary_model
@@ -581,7 +637,7 @@ Analyze the image and use specific details from both the visual content and the
.unwrap_or(&self.primary_model) .unwrap_or(&self.primary_model)
}; };
let options = self.num_ctx.map(|ctx| OllamaOptions { num_ctx: Some(ctx) }); let options = self.build_options();
let request_body = OllamaChatRequest { let request_body = OllamaChatRequest {
model, model,
@@ -623,7 +679,11 @@ Analyze the image and use specific details from both the visual content and the
.await .await
.with_context(|| "Failed to parse Ollama chat response")?; .with_context(|| "Failed to parse Ollama chat response")?;
Ok(chat_response.message) Ok((
chat_response.message,
chat_response.prompt_eval_count,
chat_response.eval_count,
))
} }
/// Generate an embedding vector for text using nomic-embed-text:v1.5 /// Generate an embedding vector for text using nomic-embed-text:v1.5
@@ -770,7 +830,16 @@ struct OllamaRequest {
#[derive(Serialize)] #[derive(Serialize)]
struct OllamaOptions { struct OllamaOptions {
#[serde(skip_serializing_if = "Option::is_none")]
num_ctx: Option<i32>, num_ctx: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
temperature: Option<f32>,
#[serde(skip_serializing_if = "Option::is_none")]
top_p: Option<f32>,
#[serde(skip_serializing_if = "Option::is_none")]
top_k: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
min_p: Option<f32>,
} }
/// Tool definition sent in /api/chat requests (OpenAI-compatible format) /// Tool definition sent in /api/chat requests (OpenAI-compatible format)
@@ -876,6 +945,10 @@ struct OllamaChatResponse {
#[serde(default)] #[serde(default)]
#[allow(dead_code)] #[allow(dead_code)]
done_reason: String, done_reason: String,
#[serde(default)]
prompt_eval_count: Option<i32>,
#[serde(default)]
eval_count: Option<i32>,
} }
#[derive(Deserialize)] #[derive(Deserialize)]

View File

@@ -85,7 +85,7 @@ pub async fn login<D: UserDao>(
HttpResponse::Ok().json(Token { token: &token }) HttpResponse::Ok().json(Token { token: &token })
} else { } else {
error!("Failed login attempt for user: '{}'", creds.username); error!("Failed login attempt for user: '{}'", creds.username);
HttpResponse::NotFound().finish() HttpResponse::Unauthorized().finish()
} }
} }
@@ -128,7 +128,7 @@ mod tests {
} }
#[actix_rt::test] #[actix_rt::test]
async fn test_login_reports_404_when_user_does_not_exist() { async fn test_login_reports_401_when_user_does_not_exist() {
let mut dao = TestUserDao::new(); let mut dao = TestUserDao::new();
dao.create_user("user", "password"); dao.create_user("user", "password");
@@ -139,6 +139,6 @@ mod tests {
let response = login::<TestUserDao>(j, web::Data::new(Mutex::new(dao))).await; let response = login::<TestUserDao>(j, web::Data::new(Mutex::new(dao))).await;
assert_eq!(response.status(), 404); assert_eq!(response.status(), 401);
} }
} }

View File

@@ -0,0 +1,267 @@
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use clap::Parser;
use walkdir::WalkDir;
use image_api::ai::{InsightGenerator, OllamaClient, SmsApiClient};
use image_api::database::{
CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, KnowledgeDao, LocationHistoryDao,
SearchHistoryDao, SqliteCalendarEventDao, SqliteDailySummaryDao, SqliteExifDao,
SqliteInsightDao, SqliteKnowledgeDao, SqliteLocationHistoryDao, SqliteSearchHistoryDao,
};
use image_api::file_types::{IMAGE_EXTENSIONS, VIDEO_EXTENSIONS};
use image_api::tags::{SqliteTagDao, TagDao};
#[derive(Parser, Debug)]
#[command(name = "populate_knowledge")]
#[command(
about = "Batch populate the knowledge base by running the agentic insight loop over a folder"
)]
struct Args {
/// Directory to scan. Defaults to BASE_PATH from .env
#[arg(long)]
path: Option<String>,
/// Ollama model override. Defaults to OLLAMA_PRIMARY_MODEL from .env
#[arg(long)]
model: Option<String>,
/// Maximum agentic loop iterations per file
#[arg(long, default_value_t = 12)]
max_iterations: usize,
/// HTTP request timeout in seconds. Increase for large/slow models
#[arg(long, default_value_t = 120)]
timeout_secs: u64,
/// Context window size (num_ctx) passed to the model
#[arg(long)]
num_ctx: Option<i32>,
/// Sampling temperature (e.g. 0.8). Omit for model default
#[arg(long)]
temperature: Option<f32>,
/// Top-p (nucleus) sampling (e.g. 0.9). Omit for model default
#[arg(long)]
top_p: Option<f32>,
/// Top-k sampling (e.g. 40). Omit for model default
#[arg(long)]
top_k: Option<i32>,
/// Min-p sampling (e.g. 0.05). Omit for model default
#[arg(long)]
min_p: Option<f32>,
/// Re-process files that already have an insight stored
#[arg(long, default_value_t = false)]
reprocess: bool,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
env_logger::init();
dotenv::dotenv().ok();
let args = Args::parse();
let base_path = dotenv::var("BASE_PATH")?;
let scan_path = args.path.as_deref().unwrap_or(&base_path).to_string();
// Ollama config from env with CLI overrides
let primary_url = std::env::var("OLLAMA_PRIMARY_URL")
.or_else(|_| std::env::var("OLLAMA_URL"))
.unwrap_or_else(|_| "http://localhost:11434".to_string());
let fallback_url = std::env::var("OLLAMA_FALLBACK_URL").ok();
let primary_model = args
.model
.clone()
.or_else(|| std::env::var("OLLAMA_PRIMARY_MODEL").ok())
.or_else(|| std::env::var("OLLAMA_MODEL").ok())
.unwrap_or_else(|| "nemotron-3-nano:30b".to_string());
let fallback_model = std::env::var("OLLAMA_FALLBACK_MODEL").ok();
let mut ollama = OllamaClient::new(
primary_url.clone(),
fallback_url,
primary_model.clone(),
fallback_model,
)
.with_request_timeout(args.timeout_secs);
if let Some(ctx) = args.num_ctx {
ollama.set_num_ctx(Some(ctx));
}
if args.temperature.is_some()
|| args.top_p.is_some()
|| args.top_k.is_some()
|| args.min_p.is_some()
{
ollama.set_sampling_params(args.temperature, args.top_p, args.top_k, args.min_p);
}
let sms_api_url =
std::env::var("SMS_API_URL").unwrap_or_else(|_| "http://localhost:8000".to_string());
let sms_api_token = std::env::var("SMS_API_TOKEN").ok();
let sms_client = SmsApiClient::new(sms_api_url, sms_api_token);
// Wire up all DAOs
let insight_dao: Arc<Mutex<Box<dyn InsightDao>>> =
Arc::new(Mutex::new(Box::new(SqliteInsightDao::new())));
let exif_dao: Arc<Mutex<Box<dyn ExifDao>>> =
Arc::new(Mutex::new(Box::new(SqliteExifDao::new())));
let daily_summary_dao: Arc<Mutex<Box<dyn DailySummaryDao>>> =
Arc::new(Mutex::new(Box::new(SqliteDailySummaryDao::new())));
let calendar_dao: Arc<Mutex<Box<dyn CalendarEventDao>>> =
Arc::new(Mutex::new(Box::new(SqliteCalendarEventDao::new())));
let location_dao: Arc<Mutex<Box<dyn LocationHistoryDao>>> =
Arc::new(Mutex::new(Box::new(SqliteLocationHistoryDao::new())));
let search_dao: Arc<Mutex<Box<dyn SearchHistoryDao>>> =
Arc::new(Mutex::new(Box::new(SqliteSearchHistoryDao::new())));
let tag_dao: Arc<Mutex<Box<dyn TagDao>>> =
Arc::new(Mutex::new(Box::new(SqliteTagDao::default())));
let knowledge_dao: Arc<Mutex<Box<dyn KnowledgeDao>>> =
Arc::new(Mutex::new(Box::new(SqliteKnowledgeDao::new())));
let generator = InsightGenerator::new(
ollama,
sms_client,
insight_dao.clone(),
exif_dao,
daily_summary_dao,
calendar_dao,
location_dao,
search_dao,
tag_dao,
knowledge_dao,
base_path.clone(),
);
println!("Knowledge Base Population");
println!("=========================");
println!("Scan path: {}", scan_path);
println!("Model: {}", primary_model);
println!("Max iterations: {}", args.max_iterations);
println!("Timeout: {}s", args.timeout_secs);
if let Some(ctx) = args.num_ctx {
println!("Num ctx: {}", ctx);
}
if let Some(t) = args.temperature {
println!("Temperature: {}", t);
}
if let Some(p) = args.top_p {
println!("Top P: {}", p);
}
if let Some(k) = args.top_k {
println!("Top K: {}", k);
}
if let Some(m) = args.min_p {
println!("Min P: {}", m);
}
println!(
"Mode: {}",
if args.reprocess {
"reprocess all"
} else {
"skip existing"
}
);
println!();
// Collect all image and video files
let all_extensions: Vec<&str> = IMAGE_EXTENSIONS
.iter()
.chain(VIDEO_EXTENSIONS.iter())
.copied()
.collect();
println!("Scanning {}...", scan_path);
let files: Vec<PathBuf> = WalkDir::new(&scan_path)
.into_iter()
.filter_map(|e| e.ok())
.filter(|e| e.file_type().is_file())
.filter(|e| {
e.path()
.extension()
.and_then(|ext| ext.to_str())
.map(|ext| all_extensions.contains(&ext.to_lowercase().as_str()))
.unwrap_or(false)
})
.map(|e| e.path().to_path_buf())
.collect();
let total = files.len();
println!("Found {} files\n", total);
if total == 0 {
println!("Nothing to process.");
return Ok(());
}
let cx = opentelemetry::Context::new();
let mut processed = 0usize;
let mut skipped = 0usize;
let mut errors = 0usize;
for (i, path) in files.iter().enumerate() {
let relative = match path.strip_prefix(&base_path) {
Ok(p) => p.to_string_lossy().replace('\\', "/"),
Err(_) => path.to_string_lossy().replace('\\', "/"),
};
let prefix = format!("[{}/{}]", i + 1, total);
// Check for existing insight unless --reprocess
if !args.reprocess {
let has_insight = insight_dao
.lock()
.unwrap()
.get_insight(&cx, &relative)
.unwrap_or(None)
.is_some();
if has_insight {
println!("{} skip {}", prefix, relative);
skipped += 1;
continue;
}
}
println!("{} start {}", prefix, relative);
match generator
.generate_agentic_insight_for_photo(
&relative,
args.model.clone(),
None,
args.num_ctx,
args.temperature,
args.top_p,
args.top_k,
args.min_p,
args.max_iterations,
)
.await
{
Ok(_) => {
println!("{} done {}", prefix, relative);
processed += 1;
}
Err(e) => {
eprintln!("{} error {}{:?}", prefix, relative, e);
errors += 1;
}
}
}
println!();
println!("=========================");
println!("Complete");
println!(" Processed: {}", processed);
println!(" Skipped: {}", skipped);
println!(" Errors: {}", errors);
Ok(())
}

View File

@@ -85,7 +85,7 @@ impl FromRequest for Claims {
) )
.and_then(|header| { .and_then(|header| {
Claims::from_str(header) Claims::from_str(header)
.with_context(|| format!("Unable to decode token from: {}", header)) .with_context(|| "Unable to decode token from Authorization header")
}) })
.map_or_else( .map_or_else(
|e| { |e| {

View File

@@ -21,6 +21,12 @@ pub trait InsightDao: Sync + Send {
file_path: &str, file_path: &str,
) -> Result<Option<PhotoInsight>, DbError>; ) -> Result<Option<PhotoInsight>, DbError>;
fn get_insight_history(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
) -> Result<Vec<PhotoInsight>, DbError>;
fn delete_insight( fn delete_insight(
&mut self, &mut self,
context: &opentelemetry::Context, context: &opentelemetry::Context,
@@ -31,6 +37,18 @@ pub trait InsightDao: Sync + Send {
&mut self, &mut self,
context: &opentelemetry::Context, context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError>; ) -> Result<Vec<PhotoInsight>, DbError>;
fn rate_insight(
&mut self,
context: &opentelemetry::Context,
file_path: &str,
approved: bool,
) -> Result<(), DbError>;
fn get_approved_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError>;
} }
pub struct SqliteInsightDao { pub struct SqliteInsightDao {
@@ -49,6 +67,11 @@ impl SqliteInsightDao {
connection: Arc::new(Mutex::new(connect())), connection: Arc::new(Mutex::new(connect())),
} }
} }
#[cfg(test)]
pub fn from_connection(conn: Arc<Mutex<SqliteConnection>>) -> Self {
SqliteInsightDao { connection: conn }
}
} }
impl InsightDao for SqliteInsightDao { impl InsightDao for SqliteInsightDao {
@@ -62,15 +85,22 @@ impl InsightDao for SqliteInsightDao {
let mut connection = self.connection.lock().expect("Unable to get InsightDao"); let mut connection = self.connection.lock().expect("Unable to get InsightDao");
// Insert or replace on conflict (UNIQUE constraint on file_path) // Mark all existing insights for this file as no longer current
diesel::replace_into(photo_insights) diesel::update(photo_insights.filter(file_path.eq(&insight.file_path)))
.set(is_current.eq(false))
.execute(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Update is_current error"))?;
// Insert the new insight as current
diesel::insert_into(photo_insights)
.values(&insight) .values(&insight)
.execute(connection.deref_mut()) .execute(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Insert error"))?; .map_err(|_| anyhow::anyhow!("Insert error"))?;
// Retrieve the inserted record // Retrieve the inserted record (is_current = true)
photo_insights photo_insights
.filter(file_path.eq(&insight.file_path)) .filter(file_path.eq(&insight.file_path))
.filter(is_current.eq(true))
.first::<PhotoInsight>(connection.deref_mut()) .first::<PhotoInsight>(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Query error")) .map_err(|_| anyhow::anyhow!("Query error"))
}) })
@@ -89,6 +119,7 @@ impl InsightDao for SqliteInsightDao {
photo_insights photo_insights
.filter(file_path.eq(path)) .filter(file_path.eq(path))
.filter(is_current.eq(true))
.first::<PhotoInsight>(connection.deref_mut()) .first::<PhotoInsight>(connection.deref_mut())
.optional() .optional()
.map_err(|_| anyhow::anyhow!("Query error")) .map_err(|_| anyhow::anyhow!("Query error"))
@@ -96,6 +127,25 @@ impl InsightDao for SqliteInsightDao {
.map_err(|_| DbError::new(DbErrorKind::QueryError)) .map_err(|_| DbError::new(DbErrorKind::QueryError))
} }
fn get_insight_history(
&mut self,
context: &opentelemetry::Context,
path: &str,
) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_insight_history", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
photo_insights
.filter(file_path.eq(path))
.order(generated_at.desc())
.load::<PhotoInsight>(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Query error"))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn delete_insight( fn delete_insight(
&mut self, &mut self,
context: &opentelemetry::Context, context: &opentelemetry::Context,
@@ -124,6 +174,50 @@ impl InsightDao for SqliteInsightDao {
let mut connection = self.connection.lock().expect("Unable to get InsightDao"); let mut connection = self.connection.lock().expect("Unable to get InsightDao");
photo_insights photo_insights
.filter(is_current.eq(true))
.order(generated_at.desc())
.load::<PhotoInsight>(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Query error"))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn rate_insight(
&mut self,
context: &opentelemetry::Context,
path: &str,
is_approved: bool,
) -> Result<(), DbError> {
trace_db_call(context, "update", "rate_insight", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
diesel::update(
photo_insights
.filter(file_path.eq(path))
.filter(is_current.eq(true)),
)
.set(approved.eq(Some(is_approved)))
.execute(connection.deref_mut())
.map(|_| ())
.map_err(|_| anyhow::anyhow!("Update error"))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
fn get_approved_insights(
&mut self,
context: &opentelemetry::Context,
) -> Result<Vec<PhotoInsight>, DbError> {
trace_db_call(context, "query", "get_approved_insights", |_span| {
use schema::photo_insights::dsl::*;
let mut connection = self.connection.lock().expect("Unable to get InsightDao");
photo_insights
.filter(approved.eq(true))
.filter(training_messages.is_not_null())
.order(generated_at.desc()) .order(generated_at.desc())
.load::<PhotoInsight>(connection.deref_mut()) .load::<PhotoInsight>(connection.deref_mut())
.map_err(|_| anyhow::anyhow!("Query error")) .map_err(|_| anyhow::anyhow!("Query error"))

View File

@@ -0,0 +1,882 @@
use diesel::prelude::*;
use diesel::sqlite::SqliteConnection;
use std::ops::DerefMut;
use std::sync::{Arc, Mutex};
use crate::database::models::{
Entity, EntityFact, EntityPhotoLink, InsertEntity, InsertEntityFact, InsertEntityPhotoLink,
};
use crate::database::schema;
use crate::database::{DbError, DbErrorKind, connect};
use crate::otel::trace_db_call;
// ---------------------------------------------------------------------------
// Entity type normalisation
// ---------------------------------------------------------------------------
/// Canonicalise a model-supplied entity_type to a consistent lowercase form.
/// Weak models frequently vary capitalisation ("Person" vs "person") or use
/// synonym types ("location" vs "place"). Normalising here prevents duplicate
/// entities that differ only by type spelling.
pub(crate) fn normalize_entity_type(raw: &str) -> String {
match raw.to_lowercase().as_str() {
"person" | "people" | "human" | "individual" | "contact" => "person",
"place" | "location" | "venue" | "site" | "area" | "landmark" => "place",
"event" | "occasion" | "activity" | "celebration" => "event",
"thing" | "object" | "item" | "product" => "thing",
other => other,
}
.to_string()
}
// ---------------------------------------------------------------------------
// Filter / patch types
// ---------------------------------------------------------------------------
pub struct EntityFilter {
pub entity_type: Option<String>,
/// "active" | "reviewed" | "rejected" | "all"
pub status: Option<String>,
/// LIKE match on name and description
pub search: Option<String>,
pub limit: i64,
pub offset: i64,
}
pub struct FactFilter {
pub entity_id: Option<i32>,
/// "active" | "reviewed" | "rejected" | "all"
pub status: Option<String>,
pub predicate: Option<String>,
pub limit: i64,
pub offset: i64,
}
pub struct EntityPatch {
pub name: Option<String>,
pub description: Option<String>,
pub status: Option<String>,
pub confidence: Option<f32>,
}
pub struct FactPatch {
pub predicate: Option<String>,
pub object_value: Option<String>,
pub status: Option<String>,
pub confidence: Option<f32>,
}
pub struct RecentActivity {
pub entities: Vec<Entity>,
pub facts: Vec<EntityFact>,
}
// ---------------------------------------------------------------------------
// Trait
// ---------------------------------------------------------------------------
pub trait KnowledgeDao: Sync + Send {
// --- Entity ---
fn upsert_entity(
&mut self,
cx: &opentelemetry::Context,
entity: InsertEntity,
) -> Result<Entity, DbError>;
fn get_entity_by_id(
&mut self,
cx: &opentelemetry::Context,
id: i32,
) -> Result<Option<Entity>, DbError>;
fn get_entity_by_name(
&mut self,
cx: &opentelemetry::Context,
name: &str,
entity_type: Option<&str>,
) -> Result<Vec<Entity>, DbError>;
fn get_entities_with_embeddings(
&mut self,
cx: &opentelemetry::Context,
entity_type: Option<&str>,
) -> Result<Vec<Entity>, DbError>;
fn list_entities(
&mut self,
cx: &opentelemetry::Context,
filter: EntityFilter,
) -> Result<(Vec<Entity>, i64), DbError>;
fn update_entity_status(
&mut self,
cx: &opentelemetry::Context,
id: i32,
status: &str,
) -> Result<(), DbError>;
fn update_entity(
&mut self,
cx: &opentelemetry::Context,
id: i32,
patch: EntityPatch,
) -> Result<Option<Entity>, DbError>;
fn delete_entity(&mut self, cx: &opentelemetry::Context, id: i32) -> Result<(), DbError>;
fn merge_entities(
&mut self,
cx: &opentelemetry::Context,
source_id: i32,
target_id: i32,
) -> Result<(i64, i64), DbError>;
// --- Facts ---
fn upsert_fact(
&mut self,
cx: &opentelemetry::Context,
fact: InsertEntityFact,
) -> Result<(EntityFact, bool), DbError>;
fn get_facts_for_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
) -> Result<Vec<EntityFact>, DbError>;
fn list_facts(
&mut self,
cx: &opentelemetry::Context,
filter: FactFilter,
) -> Result<(Vec<EntityFact>, i64), DbError>;
fn update_fact(
&mut self,
cx: &opentelemetry::Context,
id: i32,
patch: FactPatch,
) -> Result<Option<EntityFact>, DbError>;
fn update_facts_insight_id(
&mut self,
cx: &opentelemetry::Context,
source_photo: &str,
insight_id: i32,
) -> Result<(), DbError>;
fn delete_fact(&mut self, cx: &opentelemetry::Context, id: i32) -> Result<(), DbError>;
// --- Photo links ---
fn upsert_photo_link(
&mut self,
cx: &opentelemetry::Context,
link: InsertEntityPhotoLink,
) -> Result<(), DbError>;
fn delete_photo_links_for_file(
&mut self,
cx: &opentelemetry::Context,
file_path: &str,
) -> Result<(), DbError>;
fn get_links_for_photo(
&mut self,
cx: &opentelemetry::Context,
file_path: &str,
) -> Result<Vec<EntityPhotoLink>, DbError>;
fn get_links_for_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
) -> Result<Vec<EntityPhotoLink>, DbError>;
// --- Audit ---
fn get_recent_activity(
&mut self,
cx: &opentelemetry::Context,
since: i64,
limit: i64,
) -> Result<RecentActivity, DbError>;
}
// ---------------------------------------------------------------------------
// SQLite implementation
// ---------------------------------------------------------------------------
pub struct SqliteKnowledgeDao {
connection: Arc<Mutex<SqliteConnection>>,
}
impl Default for SqliteKnowledgeDao {
fn default() -> Self {
Self::new()
}
}
impl SqliteKnowledgeDao {
pub fn new() -> Self {
SqliteKnowledgeDao {
connection: Arc::new(Mutex::new(connect())),
}
}
pub fn from_connection(conn: Arc<Mutex<SqliteConnection>>) -> Self {
SqliteKnowledgeDao { connection: conn }
}
fn serialize_embedding(vec: &[f32]) -> Vec<u8> {
vec.iter().flat_map(|f| f.to_le_bytes()).collect()
}
fn deserialize_embedding(bytes: &[u8]) -> Result<Vec<f32>, DbError> {
if bytes.len() % 4 != 0 {
return Err(DbError::new(DbErrorKind::QueryError));
}
Ok(bytes
.chunks_exact(4)
.map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
.collect())
}
pub fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
if a.len() != b.len() || a.is_empty() {
return 0.0;
}
let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
let mag_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
let mag_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
if mag_a == 0.0 || mag_b == 0.0 {
0.0
} else {
dot / (mag_a * mag_b)
}
}
}
impl KnowledgeDao for SqliteKnowledgeDao {
// -----------------------------------------------------------------------
// Entity operations
// -----------------------------------------------------------------------
fn upsert_entity(
&mut self,
cx: &opentelemetry::Context,
entity: InsertEntity,
) -> Result<Entity, DbError> {
trace_db_call(cx, "insert", "upsert_entity", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
// Normalise type before lookup and insert so that model variations
// ("Person" / "person", "location" / "place") collapse to one row.
let entity = InsertEntity {
entity_type: normalize_entity_type(&entity.entity_type),
..entity
};
// Case-insensitive lookup by name + entity_type.
// Use lower() on both sides so existing dirty rows ("Person") still match.
let name_lower = entity.name.to_lowercase();
let type_lower = entity.entity_type.to_lowercase();
let existing: Option<Entity> = entities
.filter(diesel::dsl::sql::<diesel::sql_types::Bool>(&format!(
"lower(name) = '{}' AND lower(entity_type) = '{}'",
name_lower.replace('\'', "''"),
type_lower.replace('\'', "''")
)))
.first::<Entity>(conn.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
if let Some(existing_entity) = existing {
// Update description, embedding, updated_at
diesel::update(entities.filter(id.eq(existing_entity.id)))
.set((
description.eq(&entity.description),
embedding.eq(&entity.embedding),
updated_at.eq(entity.updated_at),
))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
entities
.filter(id.eq(existing_entity.id))
.first::<Entity>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
} else {
diesel::insert_into(entities)
.values(&entity)
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))?;
entities
.order(id.desc())
.first::<Entity>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
}
})
.map_err(|_| DbError::new(DbErrorKind::InsertError))
}
fn get_entity_by_id(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
) -> Result<Option<Entity>, DbError> {
trace_db_call(cx, "query", "get_entity_by_id", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
entities
.filter(id.eq(entity_id))
.first::<Entity>(conn.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn get_entity_by_name(
&mut self,
cx: &opentelemetry::Context,
entity_name: &str,
entity_type_filter: Option<&str>,
) -> Result<Vec<Entity>, DbError> {
trace_db_call(cx, "query", "get_entity_by_name", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let name_lower = entity_name.to_lowercase().replace('\'', "''");
let mut sql = format!("lower(name) = '{}'", name_lower);
if let Some(et) = entity_type_filter {
sql.push_str(&format!(" AND entity_type = '{}'", et.replace('\'', "''")));
}
sql.push_str(" AND status != 'rejected'");
entities
.filter(diesel::dsl::sql::<diesel::sql_types::Bool>(&sql))
.load::<Entity>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn get_entities_with_embeddings(
&mut self,
cx: &opentelemetry::Context,
entity_type_filter: Option<&str>,
) -> Result<Vec<Entity>, DbError> {
trace_db_call(cx, "query", "get_entities_with_embeddings", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let mut query = entities
.filter(embedding.is_not_null())
.filter(status.ne("rejected"))
.into_boxed();
if let Some(et) = entity_type_filter {
query = query.filter(entity_type.eq(et));
}
query
.load::<Entity>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn list_entities(
&mut self,
cx: &opentelemetry::Context,
filter: EntityFilter,
) -> Result<(Vec<Entity>, i64), DbError> {
trace_db_call(cx, "query", "list_entities", |_span| {
use diesel::dsl::count_star;
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let mut query = entities.into_boxed();
if let Some(ref et) = filter.entity_type {
query = query.filter(entity_type.eq(et));
}
let status_val = filter.status.as_deref().unwrap_or("active");
if status_val != "all" {
query = query.filter(status.eq(status_val));
}
if let Some(ref search_term) = filter.search {
let pattern = format!("%{}%", search_term);
query = query.filter(name.like(pattern.clone()).or(description.like(pattern)));
}
// Count with same filters applied (build separately since boxed query is consumed)
let mut count_query = entities.into_boxed();
if let Some(ref et) = filter.entity_type {
count_query = count_query.filter(entity_type.eq(et));
}
let status_val2 = filter.status.as_deref().unwrap_or("active");
if status_val2 != "all" {
count_query = count_query.filter(status.eq(status_val2));
}
if let Some(ref search_term) = filter.search {
let pattern = format!("%{}%", search_term);
count_query =
count_query.filter(name.like(pattern.clone()).or(description.like(pattern)));
}
let total: i64 = count_query
.select(count_star())
.first(conn.deref_mut())
.unwrap_or(0);
let results = query
.order(updated_at.desc())
.limit(filter.limit)
.offset(filter.offset)
.load::<Entity>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
Ok((results, total))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn update_entity_status(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
new_status: &str,
) -> Result<(), DbError> {
trace_db_call(cx, "update", "update_entity_status", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
diesel::update(entities.filter(id.eq(entity_id)))
.set(status.eq(new_status))
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
fn update_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
patch: EntityPatch,
) -> Result<Option<Entity>, DbError> {
trace_db_call(cx, "update", "update_entity", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let now = chrono::Utc::now().timestamp();
if let Some(ref new_name) = patch.name {
diesel::update(entities.filter(id.eq(entity_id)))
.set((name.eq(new_name), updated_at.eq(now)))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update name error: {}", e))?;
}
if let Some(ref new_desc) = patch.description {
diesel::update(entities.filter(id.eq(entity_id)))
.set((description.eq(new_desc), updated_at.eq(now)))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update description error: {}", e))?;
}
if let Some(ref new_status) = patch.status {
diesel::update(entities.filter(id.eq(entity_id)))
.set((status.eq(new_status), updated_at.eq(now)))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update status error: {}", e))?;
}
if let Some(new_confidence) = patch.confidence {
diesel::update(entities.filter(id.eq(entity_id)))
.set((confidence.eq(new_confidence), updated_at.eq(now)))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update confidence error: {}", e))?;
}
entities
.filter(id.eq(entity_id))
.first::<Entity>(conn.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
fn delete_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
) -> Result<(), DbError> {
trace_db_call(cx, "delete", "delete_entity", |_span| {
use schema::entities::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
diesel::delete(entities.filter(id.eq(entity_id)))
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Delete error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn merge_entities(
&mut self,
cx: &opentelemetry::Context,
source_id: i32,
target_id: i32,
) -> Result<(i64, i64), DbError> {
trace_db_call(cx, "update", "merge_entities", |_span| {
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
conn.transaction::<(i64, i64), diesel::result::Error, _>(|conn| {
use schema::entity_facts::dsl as ef;
use schema::entity_photo_links::dsl as epl;
// 1. Re-point facts where source is subject
let facts_updated =
diesel::update(ef::entity_facts.filter(ef::subject_entity_id.eq(source_id)))
.set(ef::subject_entity_id.eq(target_id))
.execute(conn)? as i64;
// 2. Re-point facts where source is object
diesel::update(ef::entity_facts.filter(ef::object_entity_id.eq(source_id)))
.set(ef::object_entity_id.eq(Some(target_id)))
.execute(conn)?;
// 3. Copy photo links to target (INSERT OR IGNORE to skip duplicates)
let links_updated = diesel::sql_query(
"INSERT OR IGNORE INTO entity_photo_links (entity_id, file_path, role) \
SELECT ?, file_path, role FROM entity_photo_links WHERE entity_id = ?",
)
.bind::<diesel::sql_types::Integer, _>(target_id)
.bind::<diesel::sql_types::Integer, _>(source_id)
.execute(conn)? as i64;
// 4. Delete source entity (FK CASCADE removes remaining facts/links)
diesel::delete(
schema::entities::dsl::entities.filter(schema::entities::dsl::id.eq(source_id)),
)
.execute(conn)?;
Ok((facts_updated, links_updated))
})
.map_err(|e| anyhow::anyhow!("Merge transaction error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
// -----------------------------------------------------------------------
// Fact operations
// -----------------------------------------------------------------------
fn upsert_fact(
&mut self,
cx: &opentelemetry::Context,
fact: InsertEntityFact,
) -> Result<(EntityFact, bool), DbError> {
trace_db_call(cx, "insert", "upsert_fact", |_span| {
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
// Look for an identical active fact
let mut dup_query = entity_facts
.filter(subject_entity_id.eq(fact.subject_entity_id))
.filter(predicate.eq(&fact.predicate))
.filter(status.ne("rejected"))
.into_boxed();
match &fact.object_entity_id {
Some(oid) => dup_query = dup_query.filter(object_entity_id.eq(oid)),
None => dup_query = dup_query.filter(object_entity_id.is_null()),
}
match &fact.object_value {
Some(ov) => dup_query = dup_query.filter(object_value.eq(ov)),
None => dup_query = dup_query.filter(object_value.is_null()),
}
let existing: Option<EntityFact> = dup_query
.first::<EntityFact>(conn.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
if let Some(existing_fact) = existing {
// Corroborate: bump confidence by 0.1 capped at 0.95
let new_confidence = (existing_fact.confidence + 0.1).min(0.95);
diesel::update(entity_facts.filter(id.eq(existing_fact.id)))
.set(confidence.eq(new_confidence))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update confidence error: {}", e))?;
let updated = entity_facts
.filter(id.eq(existing_fact.id))
.first::<EntityFact>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
Ok((updated, false)) // false = corroborated, not newly created
} else {
diesel::insert_into(entity_facts)
.values(&fact)
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))?;
let inserted = entity_facts
.order(id.desc())
.first::<EntityFact>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
Ok((inserted, true)) // true = newly created
}
})
.map_err(|_| DbError::new(DbErrorKind::InsertError))
}
fn get_facts_for_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id: i32,
) -> Result<Vec<EntityFact>, DbError> {
trace_db_call(cx, "query", "get_facts_for_entity", |_span| {
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
entity_facts
.filter(subject_entity_id.eq(entity_id))
.filter(status.ne("rejected"))
.load::<EntityFact>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn list_facts(
&mut self,
cx: &opentelemetry::Context,
filter: FactFilter,
) -> Result<(Vec<EntityFact>, i64), DbError> {
trace_db_call(cx, "query", "list_facts", |_span| {
use diesel::dsl::count_star;
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let mut query = entity_facts.into_boxed();
if let Some(eid) = filter.entity_id {
query = query.filter(subject_entity_id.eq(eid));
}
let status_val = filter.status.as_deref().unwrap_or("active");
if status_val != "all" {
query = query.filter(status.eq(status_val));
}
if let Some(ref pred) = filter.predicate {
query = query.filter(predicate.eq(pred));
}
let total: i64 = entity_facts
.select(count_star())
.first(conn.deref_mut())
.unwrap_or(0);
let results = query
.order(created_at.desc())
.limit(filter.limit)
.offset(filter.offset)
.load::<EntityFact>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
Ok((results, total))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn update_fact(
&mut self,
cx: &opentelemetry::Context,
fact_id: i32,
patch: FactPatch,
) -> Result<Option<EntityFact>, DbError> {
trace_db_call(cx, "update", "update_fact", |_span| {
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
if let Some(ref new_predicate) = patch.predicate {
diesel::update(entity_facts.filter(id.eq(fact_id)))
.set(predicate.eq(new_predicate))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
}
if let Some(ref new_value) = patch.object_value {
diesel::update(entity_facts.filter(id.eq(fact_id)))
.set(object_value.eq(new_value))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
}
if let Some(ref new_status) = patch.status {
diesel::update(entity_facts.filter(id.eq(fact_id)))
.set(status.eq(new_status))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
}
if let Some(new_confidence) = patch.confidence {
diesel::update(entity_facts.filter(id.eq(fact_id)))
.set(confidence.eq(new_confidence))
.execute(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
}
entity_facts
.filter(id.eq(fact_id))
.first::<EntityFact>(conn.deref_mut())
.optional()
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
fn update_facts_insight_id(
&mut self,
cx: &opentelemetry::Context,
photo_path: &str,
insight_id: i32,
) -> Result<(), DbError> {
trace_db_call(cx, "update", "update_facts_insight_id", |_span| {
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
diesel::update(
entity_facts
.filter(source_photo.eq(photo_path))
.filter(source_insight_id.is_null()),
)
.set(source_insight_id.eq(insight_id))
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Update error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::UpdateError))
}
fn delete_fact(&mut self, cx: &opentelemetry::Context, fact_id: i32) -> Result<(), DbError> {
trace_db_call(cx, "delete", "delete_fact", |_span| {
use schema::entity_facts::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
diesel::delete(entity_facts.filter(id.eq(fact_id)))
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Delete error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
// -----------------------------------------------------------------------
// Photo link operations
// -----------------------------------------------------------------------
fn upsert_photo_link(
&mut self,
cx: &opentelemetry::Context,
link: InsertEntityPhotoLink,
) -> Result<(), DbError> {
trace_db_call(cx, "insert", "upsert_photo_link", |_span| {
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
// INSERT OR IGNORE respects the UNIQUE(entity_id, file_path, role) constraint
diesel::sql_query(
"INSERT OR IGNORE INTO entity_photo_links (entity_id, file_path, role) VALUES (?, ?, ?)"
)
.bind::<diesel::sql_types::Integer, _>(link.entity_id)
.bind::<diesel::sql_types::Text, _>(&link.file_path)
.bind::<diesel::sql_types::Text, _>(&link.role)
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::InsertError))
}
fn delete_photo_links_for_file(
&mut self,
cx: &opentelemetry::Context,
file_path_val: &str,
) -> Result<(), DbError> {
trace_db_call(cx, "delete", "delete_photo_links_for_file", |_span| {
use schema::entity_photo_links::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
diesel::delete(entity_photo_links.filter(file_path.eq(file_path_val)))
.execute(conn.deref_mut())
.map(|_| ())
.map_err(|e| anyhow::anyhow!("Delete error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn get_links_for_photo(
&mut self,
cx: &opentelemetry::Context,
file_path_val: &str,
) -> Result<Vec<EntityPhotoLink>, DbError> {
trace_db_call(cx, "query", "get_links_for_photo", |_span| {
use schema::entity_photo_links::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
entity_photo_links
.filter(file_path.eq(file_path_val))
.load::<EntityPhotoLink>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn get_links_for_entity(
&mut self,
cx: &opentelemetry::Context,
entity_id_val: i32,
) -> Result<Vec<EntityPhotoLink>, DbError> {
trace_db_call(cx, "query", "get_links_for_entity", |_span| {
use schema::entity_photo_links::dsl::*;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
entity_photo_links
.filter(entity_id.eq(entity_id_val))
.load::<EntityPhotoLink>(conn.deref_mut())
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
// -----------------------------------------------------------------------
// Audit
// -----------------------------------------------------------------------
fn get_recent_activity(
&mut self,
cx: &opentelemetry::Context,
since: i64,
limit: i64,
) -> Result<RecentActivity, DbError> {
trace_db_call(cx, "query", "get_recent_activity", |_span| {
use schema::entities::dsl as e;
use schema::entity_facts::dsl as ef;
let mut conn = self.connection.lock().expect("KnowledgeDao lock");
let recent_entities = e::entities
.filter(e::created_at.gt(since))
.order(e::created_at.desc())
.limit(limit)
.load::<Entity>(conn.deref_mut())
.map_err(|err| anyhow::anyhow!("Query error: {}", err))?;
let recent_facts = ef::entity_facts
.filter(ef::created_at.gt(since))
.order(ef::created_at.desc())
.limit(limit)
.load::<EntityFact>(conn.deref_mut())
.map_err(|err| anyhow::anyhow!("Query error: {}", err))?;
Ok(RecentActivity {
entities: recent_entities,
facts: recent_facts,
})
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
}

View File

@@ -12,6 +12,7 @@ use crate::otel::trace_db_call;
pub mod calendar_dao; pub mod calendar_dao;
pub mod daily_summary_dao; pub mod daily_summary_dao;
pub mod insights_dao; pub mod insights_dao;
pub mod knowledge_dao;
pub mod location_dao; pub mod location_dao;
pub mod models; pub mod models;
pub mod preview_dao; pub mod preview_dao;
@@ -21,6 +22,10 @@ pub mod search_dao;
pub use calendar_dao::{CalendarEventDao, SqliteCalendarEventDao}; pub use calendar_dao::{CalendarEventDao, SqliteCalendarEventDao};
pub use daily_summary_dao::{DailySummaryDao, InsertDailySummary, SqliteDailySummaryDao}; pub use daily_summary_dao::{DailySummaryDao, InsertDailySummary, SqliteDailySummaryDao};
pub use insights_dao::{InsightDao, SqliteInsightDao}; pub use insights_dao::{InsightDao, SqliteInsightDao};
pub use knowledge_dao::{
EntityFilter, EntityPatch, FactFilter, FactPatch, KnowledgeDao, RecentActivity,
SqliteKnowledgeDao,
};
pub use location_dao::{LocationHistoryDao, SqliteLocationHistoryDao}; pub use location_dao::{LocationHistoryDao, SqliteLocationHistoryDao};
pub use preview_dao::{PreviewDao, SqlitePreviewDao}; pub use preview_dao::{PreviewDao, SqlitePreviewDao};
pub use search_dao::{SearchHistoryDao, SqliteSearchHistoryDao}; pub use search_dao::{SearchHistoryDao, SqliteSearchHistoryDao};
@@ -299,17 +304,6 @@ pub trait ExifDao: Sync + Send {
context: &opentelemetry::Context, context: &opentelemetry::Context,
) -> Result<Vec<String>, DbError>; ) -> Result<Vec<String>, DbError>;
/// Get files sorted by date with optional pagination
/// Returns (sorted_file_paths, total_count)
fn get_files_sorted_by_date(
&mut self,
context: &opentelemetry::Context,
file_paths: &[String],
ascending: bool,
limit: Option<i64>,
offset: i64,
) -> Result<(Vec<String>, i64), DbError>;
/// Get all photos with GPS coordinates /// Get all photos with GPS coordinates
/// Returns Vec<(file_path, latitude, longitude, date_taken)> /// Returns Vec<(file_path, latitude, longitude, date_taken)>
fn get_all_with_gps( fn get_all_with_gps(
@@ -604,66 +598,6 @@ impl ExifDao for SqliteExifDao {
.map_err(|_| DbError::new(DbErrorKind::QueryError)) .map_err(|_| DbError::new(DbErrorKind::QueryError))
} }
fn get_files_sorted_by_date(
&mut self,
context: &opentelemetry::Context,
file_paths: &[String],
ascending: bool,
limit: Option<i64>,
offset: i64,
) -> Result<(Vec<String>, i64), DbError> {
trace_db_call(context, "query", "get_files_sorted_by_date", |span| {
use diesel::dsl::count_star;
use opentelemetry::KeyValue;
use opentelemetry::trace::Span;
use schema::image_exif::dsl::*;
span.set_attributes(vec![
KeyValue::new("file_count", file_paths.len() as i64),
KeyValue::new("ascending", ascending.to_string()),
KeyValue::new("limit", limit.map(|l| l.to_string()).unwrap_or_default()),
KeyValue::new("offset", offset.to_string()),
]);
if file_paths.is_empty() {
return Ok((Vec::new(), 0));
}
let connection = &mut *self.connection.lock().unwrap();
// Get total count of files that have EXIF data
let total_count: i64 = image_exif
.filter(file_path.eq_any(file_paths))
.select(count_star())
.first(connection)
.map_err(|_| anyhow::anyhow!("Count query error"))?;
// Build sorted query
let mut query = image_exif.filter(file_path.eq_any(file_paths)).into_boxed();
// Apply sorting
// Note: SQLite NULL handling varies - NULLs appear first for ASC, last for DESC by default
if ascending {
query = query.order(date_taken.asc());
} else {
query = query.order(date_taken.desc());
}
// Apply pagination if requested
if let Some(limit_val) = limit {
query = query.limit(limit_val).offset(offset);
}
// Execute and extract file paths
let results: Vec<String> = query
.select(file_path)
.load::<String>(connection)
.map_err(|_| anyhow::anyhow!("Query error"))?;
Ok((results, total_count))
})
.map_err(|_| DbError::new(DbErrorKind::QueryError))
}
fn get_all_with_gps( fn get_all_with_gps(
&mut self, &mut self,

View File

@@ -1,4 +1,7 @@
use crate::database::schema::{favorites, image_exif, photo_insights, users, video_preview_clips}; use crate::database::schema::{
entities, entity_facts, entity_photo_links, favorites, image_exif, photo_insights, users,
video_preview_clips,
};
use serde::Serialize; use serde::Serialize;
#[derive(Insertable)] #[derive(Insertable)]
@@ -82,6 +85,8 @@ pub struct InsertPhotoInsight {
pub summary: String, pub summary: String,
pub generated_at: i64, pub generated_at: i64,
pub model_version: String, pub model_version: String,
pub is_current: bool,
pub training_messages: Option<String>,
} }
#[derive(Serialize, Queryable, Clone, Debug)] #[derive(Serialize, Queryable, Clone, Debug)]
@@ -92,6 +97,81 @@ pub struct PhotoInsight {
pub summary: String, pub summary: String,
pub generated_at: i64, pub generated_at: i64,
pub model_version: String, pub model_version: String,
pub is_current: bool,
pub training_messages: Option<String>,
pub approved: Option<bool>,
}
// --- Knowledge memory models ---
#[derive(Insertable)]
#[diesel(table_name = entities)]
pub struct InsertEntity {
pub name: String,
pub entity_type: String,
pub description: String,
pub embedding: Option<Vec<u8>>,
pub confidence: f32,
pub status: String,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Serialize, Queryable, Clone, Debug)]
pub struct Entity {
pub id: i32,
pub name: String,
pub entity_type: String,
pub description: String,
pub embedding: Option<Vec<u8>>,
pub confidence: f32,
pub status: String,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Insertable)]
#[diesel(table_name = entity_facts)]
pub struct InsertEntityFact {
pub subject_entity_id: i32,
pub predicate: String,
pub object_entity_id: Option<i32>,
pub object_value: Option<String>,
pub source_photo: Option<String>,
pub source_insight_id: Option<i32>,
pub confidence: f32,
pub status: String,
pub created_at: i64,
}
#[derive(Serialize, Queryable, Clone, Debug)]
pub struct EntityFact {
pub id: i32,
pub subject_entity_id: i32,
pub predicate: String,
pub object_entity_id: Option<i32>,
pub object_value: Option<String>,
pub source_photo: Option<String>,
pub source_insight_id: Option<i32>,
pub confidence: f32,
pub status: String,
pub created_at: i64,
}
#[derive(Insertable)]
#[diesel(table_name = entity_photo_links)]
pub struct InsertEntityPhotoLink {
pub entity_id: i32,
pub file_path: String,
pub role: String,
}
#[derive(Serialize, Queryable, Clone, Debug)]
pub struct EntityPhotoLink {
pub id: i32,
pub entity_id: i32,
pub file_path: String,
pub role: String,
} }
#[derive(Insertable)] #[derive(Insertable)]

View File

@@ -31,6 +31,44 @@ diesel::table! {
} }
} }
diesel::table! {
entities (id) {
id -> Integer,
name -> Text,
entity_type -> Text,
description -> Text,
embedding -> Nullable<Binary>,
confidence -> Float,
status -> Text,
created_at -> BigInt,
updated_at -> BigInt,
}
}
diesel::table! {
entity_facts (id) {
id -> Integer,
subject_entity_id -> Integer,
predicate -> Text,
object_entity_id -> Nullable<Integer>,
object_value -> Nullable<Text>,
source_photo -> Nullable<Text>,
source_insight_id -> Nullable<Integer>,
confidence -> Float,
status -> Text,
created_at -> BigInt,
}
}
diesel::table! {
entity_photo_links (id) {
id -> Integer,
entity_id -> Integer,
file_path -> Text,
role -> Text,
}
}
diesel::table! { diesel::table! {
favorites (id) { favorites (id) {
id -> Integer, id -> Integer,
@@ -112,6 +150,9 @@ diesel::table! {
summary -> Text, summary -> Text,
generated_at -> BigInt, generated_at -> BigInt,
model_version -> Text, model_version -> Text,
is_current -> Bool,
training_messages -> Nullable<Text>,
approved -> Nullable<Bool>,
} }
} }
@@ -165,11 +206,16 @@ diesel::table! {
} }
} }
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
diesel::joinable!(entity_photo_links -> entities (entity_id));
diesel::joinable!(tagged_photo -> tags (tag_id)); diesel::joinable!(tagged_photo -> tags (tag_id));
diesel::allow_tables_to_appear_in_same_query!( diesel::allow_tables_to_appear_in_same_query!(
calendar_events, calendar_events,
daily_conversation_summaries, daily_conversation_summaries,
entities,
entity_facts,
entity_photo_links,
favorites, favorites,
image_exif, image_exif,
knowledge_embeddings, knowledge_embeddings,

View File

@@ -61,33 +61,9 @@ fn apply_sorting_with_exif(
match sort_type { match sort_type {
SortType::DateTakenAsc | SortType::DateTakenDesc => { SortType::DateTakenAsc | SortType::DateTakenDesc => {
info!("Date sorting requested, using database-level sorting"); info!("Date sorting requested, using in-memory sort with EXIF/filename fallback");
// Use in-memory sort so files without EXIF dates are included via
// Collect file paths for batch EXIF query // filename extraction and filesystem metadata fallbacks.
let file_paths: Vec<String> = files.iter().map(|f| f.file_name.clone()).collect();
// Try database-level sorting first (most efficient)
let ascending = sort_type == SortType::DateTakenAsc;
match exif_dao.get_files_sorted_by_date(
span_context,
&file_paths,
ascending,
limit,
offset,
) {
Ok((sorted_files, db_total)) => {
info!(
"Database-level date sorting succeeded, returned {} files",
sorted_files.len()
);
(sorted_files, db_total)
}
Err(e) => {
warn!(
"Database-level sorting failed: {:?}, falling back to in-memory sort",
e
);
// Fallback to in-memory sorting with date extraction
let (sorted, _) = in_memory_date_sort( let (sorted, _) = in_memory_date_sort(
files, files,
sort_type, sort_type,
@@ -99,8 +75,6 @@ fn apply_sorting_with_exif(
); );
(sorted, total_count) (sorted, total_count)
} }
}
}
_ => { _ => {
// Use regular sort for non-date sorting // Use regular sort for non-date sorting
let sorted = sort(files, sort_type); let sorted = sort(files, sort_type);
@@ -958,6 +932,7 @@ pub async fn get_gps_summary(
request: HttpRequest, request: HttpRequest,
req: Query<FilesRequest>, req: Query<FilesRequest>,
exif_dao: Data<Mutex<Box<dyn ExifDao>>>, exif_dao: Data<Mutex<Box<dyn ExifDao>>>,
app_state: Data<AppState>,
) -> Result<HttpResponse, actix_web::Error> { ) -> Result<HttpResponse, actix_web::Error> {
use crate::data::{GpsPhotoSummary, GpsPhotosResponse}; use crate::data::{GpsPhotoSummary, GpsPhotosResponse};
@@ -978,17 +953,17 @@ pub async fn get_gps_summary(
// The database stores relative paths, so we use the path as-is // The database stores relative paths, so we use the path as-is
// Normalize empty path or "/" to return all GPS photos // Normalize empty path or "/" to return all GPS photos
let requested_path = if req.path.is_empty() || req.path == "/" { let requested_path = if req.path.is_empty() || req.path == "/" {
"" String::new()
} else { } else {
// Just do basic validation to prevent path traversal // Validate path using the same check as all other endpoints
if req.path.contains("..") { if is_valid_full_path(&app_state.base_path, &req.path, false).is_none() {
warn!("Path traversal attempt: {}", req.path); warn!("Invalid path for GPS summary: {}", req.path);
cx.span().set_status(Status::error("Invalid path")); cx.span().set_status(Status::error("Invalid path"));
return Ok(HttpResponse::Forbidden().json(serde_json::json!({ return Ok(HttpResponse::BadRequest().json(serde_json::json!({
"error": "Invalid path" "error": "Invalid path"
}))); })));
} }
req.path.as_str() req.path.clone()
}; };
let recursive = req.recursive.unwrap_or(false); let recursive = req.recursive.unwrap_or(false);
@@ -999,7 +974,7 @@ pub async fn get_gps_summary(
// Query database for all photos with GPS // Query database for all photos with GPS
let mut exif_dao_guard = exif_dao.lock().expect("Unable to get ExifDao"); let mut exif_dao_guard = exif_dao.lock().expect("Unable to get ExifDao");
match exif_dao_guard.get_all_with_gps(&cx, requested_path, recursive) { match exif_dao_guard.get_all_with_gps(&cx, &requested_path, recursive) {
Ok(gps_data) => { Ok(gps_data) => {
let mut photos: Vec<GpsPhotoSummary> = gps_data let mut photos: Vec<GpsPhotoSummary> = gps_data
.into_iter() .into_iter()
@@ -1352,19 +1327,6 @@ mod tests {
Ok(Vec::new()) Ok(Vec::new())
} }
fn get_files_sorted_by_date(
&mut self,
_context: &opentelemetry::Context,
file_paths: &[String],
_ascending: bool,
_limit: Option<i64>,
_offset: i64,
) -> Result<(Vec<String>, i64), DbError> {
// For tests, just return all files unsorted
let count = file_paths.len() as i64;
Ok((file_paths.to_vec(), count))
}
fn get_all_with_gps( fn get_all_with_gps(
&mut self, &mut self,
_context: &opentelemetry::Context, _context: &opentelemetry::Context,

567
src/knowledge.rs Normal file
View File

@@ -0,0 +1,567 @@
use actix_web::dev::{ServiceFactory, ServiceRequest};
use actix_web::{App, HttpResponse, Responder, web};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use std::sync::Mutex;
use crate::data::Claims;
use crate::database::models::{Entity, EntityFact, EntityPhotoLink};
use crate::database::{
EntityFilter, EntityPatch, FactFilter, FactPatch, KnowledgeDao, RecentActivity,
};
// ---------------------------------------------------------------------------
// Request / Response types
// ---------------------------------------------------------------------------
#[derive(Serialize)]
pub struct EntitySummary {
pub id: i32,
pub name: String,
pub entity_type: String,
pub description: String,
pub confidence: f32,
pub status: String,
pub created_at: i64,
pub updated_at: i64,
}
impl From<Entity> for EntitySummary {
fn from(e: Entity) -> Self {
EntitySummary {
id: e.id,
name: e.name,
entity_type: e.entity_type,
description: e.description,
confidence: e.confidence,
status: e.status,
created_at: e.created_at,
updated_at: e.updated_at,
}
}
}
#[derive(Serialize)]
pub struct EntityListResponse {
pub entities: Vec<EntitySummary>,
pub total: i64,
pub limit: i64,
pub offset: i64,
}
#[derive(Serialize)]
pub struct FactDetail {
pub id: i32,
pub predicate: String,
pub object_entity_id: Option<i32>,
pub object_entity_name: Option<String>,
pub object_value: Option<String>,
pub confidence: f32,
pub status: String,
pub source_photo: Option<String>,
pub source_insight_id: Option<i32>,
pub created_at: i64,
}
#[derive(Serialize)]
pub struct PhotoLinkDetail {
pub file_path: String,
pub role: String,
}
impl From<EntityPhotoLink> for PhotoLinkDetail {
fn from(l: EntityPhotoLink) -> Self {
PhotoLinkDetail {
file_path: l.file_path,
role: l.role,
}
}
}
#[derive(Serialize)]
pub struct EntityDetailResponse {
pub id: i32,
pub name: String,
pub entity_type: String,
pub description: String,
pub confidence: f32,
pub status: String,
pub created_at: i64,
pub updated_at: i64,
pub facts: Vec<FactDetail>,
pub photo_links: Vec<PhotoLinkDetail>,
}
#[derive(Serialize)]
pub struct FactSummary {
pub id: i32,
pub subject_entity_id: i32,
pub subject_entity_name: Option<String>,
pub predicate: String,
pub object_entity_id: Option<i32>,
pub object_entity_name: Option<String>,
pub object_value: Option<String>,
pub confidence: f32,
pub status: String,
pub source_photo: Option<String>,
pub source_insight_id: Option<i32>,
pub created_at: i64,
}
#[derive(Serialize)]
pub struct FactListResponse {
pub facts: Vec<FactSummary>,
pub total: i64,
pub limit: i64,
pub offset: i64,
}
#[derive(Deserialize)]
pub struct MergeRequest {
pub source_id: i32,
pub target_id: i32,
}
#[derive(Serialize)]
pub struct MergeResponse {
pub merged_entity_id: i32,
pub deleted_entity_id: i32,
pub facts_transferred: i64,
pub links_transferred: i64,
}
#[derive(Deserialize)]
pub struct EntityPatchRequest {
pub name: Option<String>,
pub description: Option<String>,
pub status: Option<String>,
pub confidence: Option<f32>,
}
#[derive(Deserialize)]
pub struct FactPatchRequest {
pub predicate: Option<String>,
pub object_value: Option<String>,
pub status: Option<String>,
pub confidence: Option<f32>,
}
#[derive(Deserialize)]
pub struct EntityListQuery {
#[serde(rename = "type")]
pub entity_type: Option<String>,
pub status: Option<String>,
pub search: Option<String>,
pub limit: Option<i64>,
pub offset: Option<i64>,
}
#[derive(Deserialize)]
pub struct FactListQuery {
pub entity_id: Option<i32>,
pub status: Option<String>,
pub predicate: Option<String>,
pub limit: Option<i64>,
pub offset: Option<i64>,
}
#[derive(Deserialize)]
pub struct RecentQuery {
pub since: Option<i64>,
pub limit: Option<i64>,
}
// ---------------------------------------------------------------------------
// Service registration
// ---------------------------------------------------------------------------
pub fn add_knowledge_services<T, D: KnowledgeDao + 'static>(app: App<T>) -> App<T>
where
T: ServiceFactory<ServiceRequest, Config = (), Error = actix_web::Error, InitError = ()>,
{
app.service(
web::scope("/knowledge")
.service(web::resource("/entities").route(web::get().to(list_entities::<D>)))
.service(web::resource("/entities/merge").route(web::post().to(merge_entities::<D>)))
.service(
web::resource("/entities/{id}")
.route(web::get().to(get_entity::<D>))
.route(web::patch().to(patch_entity::<D>))
.route(web::delete().to(delete_entity::<D>)),
)
.service(web::resource("/facts").route(web::get().to(list_facts::<D>)))
.service(
web::resource("/facts/{id}")
.route(web::patch().to(patch_fact::<D>))
.route(web::delete().to(delete_fact::<D>)),
)
.service(web::resource("/recent").route(web::get().to(get_recent::<D>))),
)
}
// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------
async fn list_entities<D: KnowledgeDao + 'static>(
_claims: Claims,
query: web::Query<EntityListQuery>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let limit = query.limit.unwrap_or(50).min(200);
let offset = query.offset.unwrap_or(0);
let status_filter = match query.status.as_deref() {
None | Some("active") => Some("active".to_string()),
Some("all") => None,
Some(s) => Some(s.to_string()),
};
let filter = EntityFilter {
entity_type: query.entity_type.clone(),
status: status_filter,
search: query.search.clone(),
limit,
offset,
};
let cx = opentelemetry::Context::current();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.list_entities(&cx, filter) {
Ok((entities, total)) => {
let summaries: Vec<EntitySummary> =
entities.into_iter().map(EntitySummary::from).collect();
HttpResponse::Ok().json(EntityListResponse {
entities: summaries,
total,
limit,
offset,
})
}
Err(e) => {
log::error!("list_entities error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn get_entity<D: KnowledgeDao + 'static>(
_claims: Claims,
id: web::Path<i32>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let cx = opentelemetry::Context::current();
let entity_id = id.into_inner();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
let entity = match dao.get_entity_by_id(&cx, entity_id) {
Ok(Some(e)) => e,
Ok(None) => {
return HttpResponse::NotFound().json(serde_json::json!({"error": "Entity not found"}));
}
Err(e) => {
log::error!("get_entity error: {:?}", e);
return HttpResponse::InternalServerError()
.json(serde_json::json!({"error": "Database error"}));
}
};
// Fetch all facts (all statuses for audit)
let raw_facts: Vec<EntityFact> = match dao.get_facts_for_entity(&cx, entity_id) {
Ok(f) => f,
Err(e) => {
log::error!("get_facts_for_entity error: {:?}", e);
return HttpResponse::InternalServerError()
.json(serde_json::json!({"error": "Database error"}));
}
};
// Resolve object entity names
let mut facts = Vec::with_capacity(raw_facts.len());
for f in raw_facts {
let object_entity_name = if let Some(oid) = f.object_entity_id {
dao.get_entity_by_id(&cx, oid)
.ok()
.flatten()
.map(|e| e.name)
} else {
None
};
facts.push(FactDetail {
id: f.id,
predicate: f.predicate,
object_entity_id: f.object_entity_id,
object_entity_name,
object_value: f.object_value,
confidence: f.confidence,
status: f.status,
source_photo: f.source_photo,
source_insight_id: f.source_insight_id,
created_at: f.created_at,
});
}
// Fetch photo links
let photo_links: Vec<PhotoLinkDetail> = match dao.get_links_for_entity(&cx, entity_id) {
Ok(links) => links.into_iter().map(PhotoLinkDetail::from).collect(),
Err(e) => {
log::error!("get_links_for_entity error: {:?}", e);
return HttpResponse::InternalServerError()
.json(serde_json::json!({"error": "Database error"}));
}
};
HttpResponse::Ok().json(EntityDetailResponse {
id: entity.id,
name: entity.name,
entity_type: entity.entity_type,
description: entity.description,
confidence: entity.confidence,
status: entity.status,
created_at: entity.created_at,
updated_at: entity.updated_at,
facts,
photo_links,
})
}
async fn patch_entity<D: KnowledgeDao + 'static>(
_claims: Claims,
id: web::Path<i32>,
body: web::Json<EntityPatchRequest>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let cx = opentelemetry::Context::current();
let entity_id = id.into_inner();
let patch = EntityPatch {
name: body.name.clone(),
description: body.description.clone(),
status: body.status.clone(),
confidence: body.confidence,
};
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.update_entity(&cx, entity_id, patch) {
Ok(Some(entity)) => HttpResponse::Ok().json(EntitySummary::from(entity)),
Ok(None) => HttpResponse::NotFound().json(serde_json::json!({"error": "Entity not found"})),
Err(e) => {
log::error!("patch_entity error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn delete_entity<D: KnowledgeDao + 'static>(
_claims: Claims,
id: web::Path<i32>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let cx = opentelemetry::Context::current();
let entity_id = id.into_inner();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
// Verify entity exists before deleting
match dao.get_entity_by_id(&cx, entity_id) {
Ok(None) => {
return HttpResponse::NotFound().json(serde_json::json!({"error": "Entity not found"}));
}
Err(e) => {
log::error!("delete_entity lookup error: {:?}", e);
return HttpResponse::InternalServerError()
.json(serde_json::json!({"error": "Database error"}));
}
Ok(Some(_)) => {}
}
match dao.delete_entity(&cx, entity_id) {
Ok(()) => HttpResponse::NoContent().finish(),
Err(e) => {
log::error!("delete_entity error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn merge_entities<D: KnowledgeDao + 'static>(
_claims: Claims,
body: web::Json<MergeRequest>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
if body.source_id == body.target_id {
return HttpResponse::BadRequest()
.json(serde_json::json!({"error": "source_id and target_id must be different"}));
}
let cx = opentelemetry::Context::current();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
// Verify both entities exist
for id in [body.source_id, body.target_id] {
match dao.get_entity_by_id(&cx, id) {
Ok(None) => {
return HttpResponse::BadRequest()
.json(serde_json::json!({"error": format!("Entity {} not found", id)}));
}
Err(e) => {
log::error!("merge_entities lookup error: {:?}", e);
return HttpResponse::InternalServerError()
.json(serde_json::json!({"error": "Database error"}));
}
Ok(Some(_)) => {}
}
}
match dao.merge_entities(&cx, body.source_id, body.target_id) {
Ok((facts_transferred, links_transferred)) => HttpResponse::Ok().json(MergeResponse {
merged_entity_id: body.target_id,
deleted_entity_id: body.source_id,
facts_transferred,
links_transferred,
}),
Err(e) => {
log::error!("merge_entities error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn list_facts<D: KnowledgeDao + 'static>(
_claims: Claims,
query: web::Query<FactListQuery>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let limit = query.limit.unwrap_or(50).min(200);
let offset = query.offset.unwrap_or(0);
let status_filter = match query.status.as_deref() {
None | Some("active") => Some("active".to_string()),
Some("all") => None,
Some(s) => Some(s.to_string()),
};
let filter = FactFilter {
entity_id: query.entity_id,
status: status_filter,
predicate: query.predicate.clone(),
limit,
offset,
};
let cx = opentelemetry::Context::current();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.list_facts(&cx, filter) {
Ok((facts, total)) => {
let mut summaries = Vec::with_capacity(facts.len());
for f in facts {
let subject_entity_name = dao
.get_entity_by_id(&cx, f.subject_entity_id)
.ok()
.flatten()
.map(|e| e.name);
let object_entity_name = if let Some(oid) = f.object_entity_id {
dao.get_entity_by_id(&cx, oid)
.ok()
.flatten()
.map(|e| e.name)
} else {
None
};
summaries.push(FactSummary {
id: f.id,
subject_entity_id: f.subject_entity_id,
subject_entity_name,
predicate: f.predicate,
object_entity_id: f.object_entity_id,
object_entity_name,
object_value: f.object_value,
confidence: f.confidence,
status: f.status,
source_photo: f.source_photo,
source_insight_id: f.source_insight_id,
created_at: f.created_at,
});
}
HttpResponse::Ok().json(FactListResponse {
facts: summaries,
total,
limit,
offset,
})
}
Err(e) => {
log::error!("list_facts error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn patch_fact<D: KnowledgeDao + 'static>(
_claims: Claims,
id: web::Path<i32>,
body: web::Json<FactPatchRequest>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let cx = opentelemetry::Context::current();
let fact_id = id.into_inner();
let patch = FactPatch {
predicate: body.predicate.clone(),
object_value: body.object_value.clone(),
status: body.status.clone(),
confidence: body.confidence,
};
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.update_fact(&cx, fact_id, patch) {
Ok(Some(fact)) => HttpResponse::Ok().json(fact),
Ok(None) => HttpResponse::NotFound().json(serde_json::json!({"error": "Fact not found"})),
Err(e) => {
log::error!("patch_fact error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}
async fn delete_fact<D: KnowledgeDao + 'static>(
_claims: Claims,
id: web::Path<i32>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let cx = opentelemetry::Context::current();
let fact_id = id.into_inner();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.delete_fact(&cx, fact_id) {
Ok(()) => HttpResponse::NoContent().finish(),
Err(e) => {
log::warn!("delete_fact({}) error: {:?}", fact_id, e);
HttpResponse::NotFound().json(serde_json::json!({"error": "Fact not found"}))
}
}
}
async fn get_recent<D: KnowledgeDao + 'static>(
_claims: Claims,
query: web::Query<RecentQuery>,
dao: web::Data<Mutex<D>>,
) -> impl Responder {
let since = query
.since
.unwrap_or_else(|| Utc::now().timestamp() - 86400);
let limit = query.limit.unwrap_or(20).min(100);
let cx = opentelemetry::Context::current();
let mut dao = dao.lock().expect("Unable to lock KnowledgeDao");
match dao.get_recent_activity(&cx, since, limit) {
Ok(RecentActivity { entities, facts }) => {
let entity_summaries: Vec<EntitySummary> =
entities.into_iter().map(EntitySummary::from).collect();
HttpResponse::Ok().json(serde_json::json!({
"entities": entity_summaries,
"facts": facts
}))
}
Err(e) => {
log::error!("get_recent error: {:?}", e);
HttpResponse::InternalServerError().json(serde_json::json!({"error": "Database error"}))
}
}
}

View File

@@ -67,6 +67,7 @@ mod tags;
mod utils; mod utils;
mod video; mod video;
mod knowledge;
mod memories; mod memories;
mod otel; mod otel;
mod service; mod service;
@@ -502,14 +503,10 @@ async fn stream_video(
let playlist = &path.path; let playlist = &path.path;
debug!("Playlist: {}", playlist); debug!("Playlist: {}", playlist);
// Extract video playlist dir to dotenv // Only serve files under video_path (HLS playlists) or base_path (source videos)
if !playlist.starts_with(&app_state.video_path) if playlist.starts_with(&app_state.video_path)
&& is_valid_full_path(&app_state.base_path, playlist, false).is_some() || is_valid_full_path(&app_state.base_path, playlist, false).is_some()
{ {
span.set_status(Status::error(format!("playlist not valid {}", playlist)));
HttpResponse::BadRequest().finish()
} else {
match NamedFile::open(playlist) { match NamedFile::open(playlist) {
Ok(file) => { Ok(file) => {
span.set_status(Status::Ok); span.set_status(Status::Ok);
@@ -520,6 +517,9 @@ async fn stream_video(
HttpResponse::NotFound().finish() HttpResponse::NotFound().finish()
} }
} }
} else {
span.set_status(Status::error(format!("playlist not valid {}", playlist)));
HttpResponse::BadRequest().finish()
} }
} }
@@ -1185,7 +1185,10 @@ fn main() -> std::io::Result<()> {
.service(ai::delete_insight_handler) .service(ai::delete_insight_handler)
.service(ai::get_all_insights_handler) .service(ai::get_all_insights_handler)
.service(ai::get_available_models_handler) .service(ai::get_available_models_handler)
.service(ai::rate_insight_handler)
.service(ai::export_training_data_handler)
.add_feature(add_tag_services::<_, SqliteTagDao>) .add_feature(add_tag_services::<_, SqliteTagDao>)
.add_feature(knowledge::add_knowledge_services::<_, SqliteKnowledgeDao>)
.app_data(app_data.clone()) .app_data(app_data.clone())
.app_data::<Data<RealFileSystem>>(Data::new(RealFileSystem::new( .app_data::<Data<RealFileSystem>>(Data::new(RealFileSystem::new(
app_data.base_path.clone(), app_data.base_path.clone(),
@@ -1204,6 +1207,10 @@ fn main() -> std::io::Result<()> {
.app_data::<Data<Mutex<Box<dyn PreviewDao>>>>(Data::new(Mutex::new(Box::new( .app_data::<Data<Mutex<Box<dyn PreviewDao>>>>(Data::new(Mutex::new(Box::new(
preview_dao, preview_dao,
)))) ))))
.app_data::<Data<Mutex<SqliteKnowledgeDao>>>(Data::new(Mutex::new(
SqliteKnowledgeDao::new(),
)))
.app_data(mp::form::MultipartFormConfig::default().total_limit(1024 * 1024 * 1024)) // 1GB upload limit
.app_data(web::JsonConfig::default().error_handler(|err, req| { .app_data(web::JsonConfig::default().error_handler(|err, req| {
let detail = err.to_string(); let detail = err.to_string();
log::warn!( log::warn!(

View File

@@ -1,8 +1,8 @@
use crate::ai::{InsightGenerator, OllamaClient, SmsApiClient}; use crate::ai::{InsightGenerator, OllamaClient, SmsApiClient};
use crate::database::{ use crate::database::{
CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, LocationHistoryDao, SearchHistoryDao, CalendarEventDao, DailySummaryDao, ExifDao, InsightDao, KnowledgeDao, LocationHistoryDao,
SqliteCalendarEventDao, SqliteDailySummaryDao, SqliteExifDao, SqliteInsightDao, SearchHistoryDao, SqliteCalendarEventDao, SqliteDailySummaryDao, SqliteExifDao,
SqliteLocationHistoryDao, SqliteSearchHistoryDao, SqliteInsightDao, SqliteKnowledgeDao, SqliteLocationHistoryDao, SqliteSearchHistoryDao,
}; };
use crate::database::{PreviewDao, SqlitePreviewDao}; use crate::database::{PreviewDao, SqlitePreviewDao};
use crate::tags::{SqliteTagDao, TagDao}; use crate::tags::{SqliteTagDao, TagDao};
@@ -119,6 +119,8 @@ impl Default for AppState {
Arc::new(Mutex::new(Box::new(SqliteSearchHistoryDao::new()))); Arc::new(Mutex::new(Box::new(SqliteSearchHistoryDao::new())));
let tag_dao: Arc<Mutex<Box<dyn TagDao>>> = let tag_dao: Arc<Mutex<Box<dyn TagDao>>> =
Arc::new(Mutex::new(Box::new(SqliteTagDao::default()))); Arc::new(Mutex::new(Box::new(SqliteTagDao::default())));
let knowledge_dao: Arc<Mutex<Box<dyn KnowledgeDao>>> =
Arc::new(Mutex::new(Box::new(SqliteKnowledgeDao::new())));
// Load base path // Load base path
let base_path = env::var("BASE_PATH").expect("BASE_PATH was not set in the env"); let base_path = env::var("BASE_PATH").expect("BASE_PATH was not set in the env");
@@ -134,6 +136,7 @@ impl Default for AppState {
location_dao.clone(), location_dao.clone(),
search_dao.clone(), search_dao.clone(),
tag_dao.clone(), tag_dao.clone(),
knowledge_dao,
base_path.clone(), base_path.clone(),
); );
@@ -200,6 +203,8 @@ impl AppState {
Arc::new(Mutex::new(Box::new(SqliteSearchHistoryDao::new()))); Arc::new(Mutex::new(Box::new(SqliteSearchHistoryDao::new())));
let tag_dao: Arc<Mutex<Box<dyn TagDao>>> = let tag_dao: Arc<Mutex<Box<dyn TagDao>>> =
Arc::new(Mutex::new(Box::new(SqliteTagDao::default()))); Arc::new(Mutex::new(Box::new(SqliteTagDao::default())));
let knowledge_dao: Arc<Mutex<Box<dyn KnowledgeDao>>> =
Arc::new(Mutex::new(Box::new(SqliteKnowledgeDao::new())));
// Initialize test InsightGenerator with all data sources // Initialize test InsightGenerator with all data sources
let base_path_str = base_path.to_string_lossy().to_string(); let base_path_str = base_path.to_string_lossy().to_string();
@@ -213,6 +218,7 @@ impl AppState {
location_dao.clone(), location_dao.clone(),
search_dao.clone(), search_dao.clone(),
tag_dao.clone(), tag_dao.clone(),
knowledge_dao,
base_path_str.clone(), base_path_str.clone(),
); );