feat(flowctl-db): add skills table and skill module for semantic routing

Add skills table (name, description, embedding) with vector index for
semantic skill matching. SkillRepo provides upsert() and match_skills()
using cosine similarity via BGE-small embeddings.

Reuses embed_one()/ensure_embedder() from memory module — zero duplication.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: z23cc

flowctl/crates/flowctl-db/src/lib.rs

@@ -24,12 +24,14 @@ pub mod memory;
 pub mod metrics;
 pub mod pool;
 pub mod repo;
+pub mod skill;
 
 pub use error::DbError;
 pub use indexer::{reindex, ReindexResult};
 pub use events::{EventLog, TaskTokenSummary, TokenRecord, TokenUsageRow};
 pub use memory::{MemoryEntry, MemoryFilter, MemoryRepo};
 pub use metrics::StatsQuery;
+pub use skill::{SkillEntry, SkillMatch, SkillRepo};
 pub use pool::{cleanup, open_async, open_memory_async, resolve_db_path, resolve_libsql_path, resolve_state_dir};
 pub use repo::{
     DepRepo, EpicRepo, EventRepo, EventRow, EvidenceRepo, FileLockRepo, FileOwnershipRepo,

flowctl/crates/flowctl-db/src/memory.rs

@@ -88,7 +88,7 @@ static EMBEDDER: OnceCell<Result<Mutex<TextEmbedding>, String>> = OnceCell::cons
 /// model (~130MB) via fastembed; subsequent calls return the cached
 /// instance. Initialization runs on a blocking thread because fastembed
 /// performs synchronous file I/O.
-async fn ensure_embedder() -> Result<(), DbError> {
+pub(crate) async fn ensure_embedder() -> Result<(), DbError> {
     let res = EMBEDDER
         .get_or_init(|| async {
             match tokio::task::spawn_blocking(|| {
@@ -109,7 +109,7 @@ async fn ensure_embedder() -> Result<(), DbError> {
 }
 
 /// Embed a single passage into a 384-dim vector.
-async fn embed_one(text: &str) -> Result<Vec<f32>, DbError> {
+pub(crate) async fn embed_one(text: &str) -> Result<Vec<f32>, DbError> {
     ensure_embedder().await?;
     let text = text.to_string();
     let result = tokio::task::spawn_blocking(move || {
@@ -132,7 +132,7 @@ async fn embed_one(text: &str) -> Result<Vec<f32>, DbError> {
 }
 
 /// Convert a `Vec<f32>` into a libSQL `vector32()` literal string.
-fn vec_to_literal(v: &[f32]) -> String {
+pub(crate) fn vec_to_literal(v: &[f32]) -> String {
     let parts: Vec<String> = v.iter().map(std::string::ToString::to_string).collect();
     format!("[{}]", parts.join(","))
 }

flowctl/crates/flowctl-db/src/pool.rs

@@ -98,7 +98,7 @@ async fn apply_schema(conn: &Connection) -> Result<(), DbError> {
     .await
     .map_err(|e| DbError::Schema(format!("reverse deps backfill failed: {e}")))?;
 
-    // Try to create the vector index (requires libSQL server extensions).
+    // Try to create vector indexes (requires libSQL server extensions).
     // Gracefully degrade if not available (embedded/core mode).
     let _ = conn
         .execute(
@@ -107,6 +107,13 @@ async fn apply_schema(conn: &Connection) -> Result<(), DbError> {
         )
         .await;
 
+    let _ = conn
+        .execute(
+            "CREATE INDEX IF NOT EXISTS skills_emb_idx ON skills(libsql_vector_idx(embedding))",
+            (),
+        )
+        .await;
+
     Ok(())
 }
 
@@ -215,6 +222,7 @@ mod tests {
             "daily_rollup",
             "monthly_rollup",
             "memory",
+            "skills",
         ] {
             assert!(
                 tables.contains(&expected.to_string()),

flowctl/crates/flowctl-db/src/schema.sql

@@ -217,6 +217,21 @@ CREATE TABLE IF NOT EXISTS memory (
     embedding     BLOB
 );
 
+-- ── Skills with native vector embedding (BGE-small, 384-dim) ───────
+
+CREATE TABLE IF NOT EXISTS skills (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    name        TEXT NOT NULL UNIQUE,
+    description TEXT NOT NULL,
+    plugin_path TEXT,
+    updated_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+    embedding   BLOB
+);
+
+-- Native libSQL vector index for semantic skill matching
+-- NOTE: Applied separately in pool.rs with graceful degradation.
+-- CREATE INDEX IF NOT EXISTS skills_emb_idx ON skills(libsql_vector_idx(embedding));
+
 -- ── Indexes ─────────────────────────────────────────────────────────
 
 CREATE INDEX IF NOT EXISTS idx_gaps_epic ON gaps(epic_id);

flowctl/crates/flowctl-db/src/skill.rs

@@ -0,0 +1,260 @@
+//! Skill repository with native libSQL vector search.
+//!
+//! Stores skill metadata (name, description, plugin path) with a 384-dim
+//! BGE-small embedding for semantic matching via `vector_top_k`.
+//!
+//! Reuses `embed_one()`, `ensure_embedder()`, and `vec_to_literal()` from
+//! the memory module -- zero duplication.
+
+use libsql::{params, Connection};
+
+use crate::error::DbError;
+use crate::memory::{embed_one, vec_to_literal};
+
+// ── Types ───────────────────────────────────────────────────────────
+
+/// A skill match result from semantic search.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SkillMatch {
+    pub name: String,
+    pub description: String,
+    pub score: f64,
+}
+
+/// A registered skill entry.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SkillEntry {
+    pub id: i64,
+    pub name: String,
+    pub description: String,
+    pub plugin_path: Option<String>,
+    pub updated_at: String,
+}
+
+// ── Repository ──────────────────────────────────────────────────────
+
+/// Async repository for skill metadata + semantic vector search.
+pub struct SkillRepo {
+    conn: Connection,
+}
+
+impl SkillRepo {
+    pub fn new(conn: Connection) -> Self {
+        Self { conn }
+    }
+
+    /// Insert or replace a skill. Auto-generates an embedding from
+    /// `description` when the embedder is available; otherwise leaves the
+    /// embedding NULL and logs a warning.
+    pub async fn upsert(
+        &self,
+        name: &str,
+        description: &str,
+        plugin_path: Option<&str>,
+    ) -> Result<(), DbError> {
+        let now = chrono::Utc::now().to_rfc3339();
+
+        self.conn
+            .execute(
+                "INSERT INTO skills (name, description, plugin_path, updated_at)
+                 VALUES (?1, ?2, ?3, ?4)
+                 ON CONFLICT(name) DO UPDATE SET
+                     description = excluded.description,
+                     plugin_path = excluded.plugin_path,
+                     updated_at  = excluded.updated_at",
+                params![
+                    name.to_string(),
+                    description.to_string(),
+                    plugin_path.map(String::from),
+                    now,
+                ],
+            )
+            .await?;
+
+        // Attempt to embed; swallow failures (NULL embedding is fine).
+        match embed_one(description).await {
+            Ok(vec) => {
+                let lit = vec_to_literal(&vec);
+                self.conn
+                    .execute(
+                        "UPDATE skills SET embedding = vector32(?1) WHERE name = ?2",
+                        params![lit, name.to_string()],
+                    )
+                    .await?;
+            }
+            Err(e) => {
+                tracing::warn!(
+                    skill = name,
+                    error = %e,
+                    "embedder unavailable; skill inserted without embedding"
+                );
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Semantic search: embed the query, find nearest skills via
+    /// `vector_top_k`, convert L2 distance to cosine similarity, and
+    /// filter by threshold.
+    ///
+    /// Returns `Ok(vec![])` (not an error) if the embedder or vector
+    /// index is unavailable -- graceful degradation.
+    pub async fn match_skills(
+        &self,
+        query: &str,
+        limit: usize,
+        threshold: f64,
+    ) -> Result<Vec<SkillMatch>, DbError> {
+        let vec = match embed_one(query).await {
+            Ok(v) => v,
+            Err(_) => return Ok(vec![]),
+        };
+        let lit = vec_to_literal(&vec);
+
+        let rows_result = self
+            .conn
+            .query(
+                "SELECT s.name, s.description, top.distance
+                   FROM vector_top_k('skills_emb_idx', vector32(?1), ?2) AS top
+                   JOIN skills s ON s.rowid = top.id",
+                params![lit, limit as i64],
+            )
+            .await;
+
+        let mut rows = match rows_result {
+            Ok(r) => r,
+            Err(_) => return Ok(vec![]), // vector index unavailable
+        };
+
+        let mut out = Vec::new();
+        while let Some(row) = rows.next().await? {
+            let dist: f64 = row.get(2)?;
+            // Convert L2 distance to cosine similarity.
+            // For unit-norm vectors: cos_sim = 1 - (L2^2 / 2).
+            let score = 1.0 - (dist * dist / 2.0);
+            if score >= threshold {
+                out.push(SkillMatch {
+                    name: row.get::<String>(0)?,
+                    description: row.get::<String>(1)?,
+                    score,
+                });
+            }
+        }
+        Ok(out)
+    }
+
+    /// List all registered skills (for debugging / introspection).
+    pub async fn list(&self) -> Result<Vec<SkillEntry>, DbError> {
+        let mut rows = self
+            .conn
+            .query(
+                "SELECT id, name, description, plugin_path, updated_at
+                   FROM skills
+                  ORDER BY name ASC",
+                (),
+            )
+            .await?;
+
+        let mut out = Vec::new();
+        while let Some(row) = rows.next().await? {
+            out.push(SkillEntry {
+                id: row.get::<i64>(0)?,
+                name: row.get::<String>(1)?,
+                description: row.get::<String>(2)?,
+                plugin_path: row.get::<Option<String>>(3)?,
+                updated_at: row.get::<String>(4)?,
+            });
+        }
+        Ok(out)
+    }
+}
+
+// ── Tests ───────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::pool::open_memory_async;
+
+    async fn fresh_repo() -> SkillRepo {
+        let (_db, conn) = open_memory_async().await.expect("open memory db");
+        let _ = Box::leak(Box::new(_db));
+        SkillRepo::new(conn)
+    }
+
+    #[tokio::test]
+    async fn test_upsert_and_list() {
+        let repo = fresh_repo().await;
+        repo.upsert("plan", "Plan and design tasks", Some("/plugins/flow"))
+            .await
+            .expect("upsert");
+        repo.upsert("work", "Execute implementation tasks", None)
+            .await
+            .expect("upsert");
+
+        let skills = repo.list().await.expect("list");
+        assert_eq!(skills.len(), 2);
+        assert_eq!(skills[0].name, "plan");
+        assert_eq!(skills[1].name, "work");
+    }
+
+    #[tokio::test]
+    async fn test_upsert_replaces() {
+        let repo = fresh_repo().await;
+        repo.upsert("plan", "old description", None)
+            .await
+            .expect("upsert");
+        repo.upsert("plan", "new description", Some("/new/path"))
+            .await
+            .expect("upsert");
+
+        let skills = repo.list().await.expect("list");
+        assert_eq!(skills.len(), 1);
+        assert_eq!(skills[0].description, "new description");
+        assert_eq!(skills[0].plugin_path.as_deref(), Some("/new/path"));
+    }
+
+    #[tokio::test]
+    async fn test_match_skills_graceful_no_index() {
+        // In-memory DB won't have vector index; should return empty, not error.
+        let repo = fresh_repo().await;
+        repo.upsert("plan", "Plan tasks", None)
+            .await
+            .expect("upsert");
+
+        let matches = repo
+            .match_skills("planning", 5, 0.5)
+            .await
+            .expect("match_skills should not error");
+        // May be empty if embedder or index is unavailable -- that's fine.
+        assert!(matches.len() <= 5);
+    }
+
+    /// Semantic match end-to-end. Gated behind `#[ignore]` because the
+    /// first run downloads the BGE-small model (~130MB).
+    #[tokio::test]
+    #[ignore = "requires fastembed model (~130MB); run with --ignored"]
+    async fn test_match_skills_semantic() {
+        let repo = fresh_repo().await;
+        repo.upsert("plan", "Design and architect implementation plans", None)
+            .await
+            .expect("upsert");
+        repo.upsert("work", "Execute coding tasks and write code", None)
+            .await
+            .expect("upsert");
+        repo.upsert("review", "Review code changes for quality", None)
+            .await
+            .expect("upsert");
+
+        let matches = repo
+            .match_skills("architecture design", 2, 0.3)
+            .await
+            .expect("match_skills");
+        assert!(!matches.is_empty(), "expected at least one match");
+        assert_eq!(
+            matches[0].name, "plan",
+            "expected 'plan' as best match for architecture query"
+        );
+    }
+}