convergio_knowledge/
hooks.rs

1//! Hooks — auto-embed task summaries and commit messages.
2//!
3//! - `on_task_completed`: fires on TaskCompleted domain event,
4//!   reads the task summary from DB and embeds it.
5//! - `sync_recent_commits`: scheduled task that scans recent git
6//!   commits and embeds their messages.
7
8use std::sync::Arc;
9
10use convergio_db::pool::ConnPool;
11use tracing::{info, warn};
12
13use crate::store::LanceVectorStore;
14
15/// Called when a TaskCompleted event fires.
16/// Reads the task summary and writes it to the knowledge store.
17pub fn on_task_completed(pool: &ConnPool, store: Arc<LanceVectorStore>, task_id: i64) {
18    let conn = match pool.get() {
19        Ok(c) => c,
20        Err(e) => {
21            warn!(task_id, error = %e, "knowledge hook: db error");
22            return;
23        }
24    };
25
26    let row: Result<(String, Option<String>, Option<String>), _> = conn.query_row(
27        "SELECT title, summary, notes FROM tasks WHERE id = ?1",
28        [task_id],
29        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
30    );
31
32    let (title, summary, notes) = match row {
33        Ok(r) => r,
34        Err(e) => {
35            warn!(task_id, error = %e, "task not found for knowledge embed");
36            return;
37        }
38    };
39
40    let content = format!(
41        "Task completed: {title}\n{}{}",
42        summary
43            .as_deref()
44            .map(|s| format!("Summary: {s}\n"))
45            .unwrap_or_default(),
46        notes
47            .as_deref()
48            .map(|n| format!("Notes: {n}\n"))
49            .unwrap_or_default(),
50    );
51
52    let source_id = format!("task-{task_id}");
53    tokio::spawn(async move {
54        let embedding = crate::embedder::embed(&content).await;
55        match store
56            .insert(
57                &content, "task", &source_id, None, None, None, "org", &embedding,
58            )
59            .await
60        {
61            Ok(id) => info!(id, task_id, "task knowledge embedded"),
62            Err(e) => warn!(task_id, error = %e, "embed task failed"),
63        }
64    });
65}
66
67/// Sync recent git commits — called on schedule (every 15 min).
68/// Reads the repo's recent commits and embeds their messages.
69pub fn sync_recent_commits(store: Arc<LanceVectorStore>) {
70    let repo_root = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
71    let output = match std::process::Command::new("git")
72        .args(["log", "--oneline", "-20", "--format=%H|%s"])
73        .current_dir(&repo_root)
74        .output()
75    {
76        Ok(o) if o.status.success() => String::from_utf8_lossy(&o.stdout).to_string(),
77        Ok(o) => {
78            let stderr = String::from_utf8_lossy(&o.stderr);
79            warn!(error = %stderr, "git log failed");
80            return;
81        }
82        Err(e) => {
83            warn!(error = %e, "git not available");
84            return;
85        }
86    };
87
88    let commits: Vec<(String, String)> = output
89        .lines()
90        .filter_map(|line| {
91            let (hash, msg) = line.split_once('|')?;
92            Some((hash.to_string(), msg.to_string()))
93        })
94        .collect();
95
96    if commits.is_empty() {
97        return;
98    }
99
100    tokio::spawn(async move {
101        let mut embedded = 0u32;
102        for (hash, msg) in &commits {
103            let exists = store.count_by_source("commit", hash).await.unwrap_or(0);
104            if exists > 0 {
105                continue;
106            }
107            let embedding = crate::embedder::embed(msg).await;
108            if store
109                .insert(msg, "commit", hash, None, None, None, "org", &embedding)
110                .await
111                .is_ok()
112            {
113                embedded += 1;
114            }
115        }
116        if embedded > 0 {
117            info!(embedded, "commit knowledge synced");
118        }
119    });
120}
convergio_knowledge/hooks.rs

convergio_knowledge/
hooks.rs
