fix: DB connection pooling + sync.rs cleanup

P0 fixes from design review:

1. DB connection pooling: DaemonState now holds Mutex<Connection>
   instead of opening a new connection per HTTP request. All handlers
   use state.db.lock() for shared access. Connection created once
   in create_state() helper.

2. sync.rs cleanup: Removed all re-exports of dual-write functions
   (write_epic, write_task, check_staleness, etc). Module kept as
   #[allow(dead_code)] for reference. Updated lib.rs docs to reflect
   SQLite-as-single-source-of-truth architecture.

Deferred to next session:
- P0: Leptos SSR integration into daemon (needs cargo-leptos config)
- P1: Split admin.rs (1700 lines) and task.rs (1200 lines)
- P2: Add daemon/MCP/export test coverage

0 clippy warnings, 224/224 tests pass.

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

Author: z23cc

flowctl/crates/flowctl-daemon/src/handlers.rs

@@ -20,10 +20,11 @@ use crate::lifecycle::DaemonRuntime;
 /// Shared application state for all handlers.
 pub type AppState = Arc<DaemonState>;
 
-/// Combined daemon state: runtime + event bus.
+/// Combined daemon state: runtime + event bus + shared DB connection.
 pub struct DaemonState {
     pub runtime: DaemonRuntime,
     pub event_bus: flowctl_scheduler::EventBus,
+    pub db: std::sync::Mutex<rusqlite::Connection>,
 }
 
 /// GET /api/v1/health -- simple liveness check.
@@ -68,35 +69,11 @@ pub async fn status_handler(State(state): State<AppState>) -> impl IntoResponse
 ///
 /// Returns a JSON array of epics. On database errors, returns 500.
 pub async fn epics_handler(State(state): State<AppState>) -> impl IntoResponse {
-    let db_path = state
-        .runtime
-        .paths
-        .state_dir
-        .parent()
-        .map(|flow_dir| flow_dir.join("flowctl.db"));
-
-    let Some(db_path) = db_path else {
-        return (
-            StatusCode::INTERNAL_SERVER_ERROR,
-            Json(serde_json::json!({"error": "cannot resolve db path"})),
-        );
-    };
-
-    match flowctl_db::open(&db_path) {
-        Ok(conn) => {
-            let repo = flowctl_db::EpicRepo::new(&conn);
-            match repo.list(None) {
-                Ok(epics) => (StatusCode::OK, Json(serde_json::to_value(&epics).unwrap())),
-                Err(e) => (
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    Json(serde_json::json!({"error": e.to_string()})),
-                ),
-            }
-        }
-        Err(e) => (
-            StatusCode::INTERNAL_SERVER_ERROR,
-            Json(serde_json::json!({"error": format!("db open failed: {e}")})),
-        ),
+    let conn = state.db.lock().unwrap();
+    let repo = flowctl_db::EpicRepo::new(&conn);
+    match repo.list(None) {
+        Ok(epics) => (StatusCode::OK, Json(serde_json::to_value(&epics).unwrap())),
+        Err(e) => db_error(&e.to_string()),
     }
 }
 
@@ -105,40 +82,16 @@ pub async fn tasks_handler(
     State(state): State<AppState>,
     axum::extract::Query(params): axum::extract::Query<TasksQuery>,
 ) -> impl IntoResponse {
-    let db_path = state
-        .runtime
-        .paths
-        .state_dir
-        .parent()
-        .map(|flow_dir| flow_dir.join("flowctl.db"));
-
-    let Some(db_path) = db_path else {
-        return (
-            StatusCode::INTERNAL_SERVER_ERROR,
-            Json(serde_json::json!({"error": "cannot resolve db path"})),
-        );
+    let conn = state.db.lock().unwrap();
+    let repo = flowctl_db::TaskRepo::new(&conn);
+    let result = if let Some(ref epic_id) = params.epic_id {
+        repo.list_by_epic(epic_id)
+    } else {
+        repo.list_all(None, None)
     };
-
-    match flowctl_db::open(&db_path) {
-        Ok(conn) => {
-            let repo = flowctl_db::TaskRepo::new(&conn);
-            let result = if let Some(ref epic_id) = params.epic_id {
-                repo.list_by_epic(epic_id)
-            } else {
-                repo.list_all(None, None)
-            };
-            match result {
-                Ok(tasks) => (StatusCode::OK, Json(serde_json::to_value(&tasks).unwrap())),
-                Err(e) => (
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    Json(serde_json::json!({"error": e.to_string()})),
-                ),
-            }
-        }
-        Err(e) => (
-            StatusCode::INTERNAL_SERVER_ERROR,
-            Json(serde_json::json!({"error": format!("db open failed: {e}")})),
-        ),
+    match result {
+        Ok(tasks) => (StatusCode::OK, Json(serde_json::to_value(&tasks).unwrap())),
+        Err(e) => db_error(&e.to_string()),
     }
 }
 
@@ -155,10 +108,7 @@ pub async fn create_task_handler(
     State(state): State<AppState>,
     Json(body): Json<CreateTaskRequest>,
 ) -> impl IntoResponse {
-    let conn = match open_db(&state) {
-        Ok(c) => c,
-        Err(resp) => return resp,
-    };
+    let conn = state.db.lock().unwrap();
     let task = flowctl_core::types::Task {
         schema_version: 1,
         id: body.id.clone(),
@@ -188,10 +138,7 @@ pub async fn start_task_handler(
     State(state): State<AppState>,
     Json(body): Json<TaskIdRequest>,
 ) -> impl IntoResponse {
-    let conn = match open_db(&state) {
-        Ok(c) => c,
-        Err(resp) => return resp,
-    };
+    let conn = state.db.lock().unwrap();
     let repo = flowctl_db::TaskRepo::new(&conn);
     match repo.update_status(&body.task_id, flowctl_core::state_machine::Status::InProgress) {
         Ok(()) => (StatusCode::OK, Json(serde_json::json!({"success": true, "id": body.task_id}))),
@@ -204,10 +151,7 @@ pub async fn done_task_handler(
     State(state): State<AppState>,
     Json(body): Json<TaskIdRequest>,
 ) -> impl IntoResponse {
-    let conn = match open_db(&state) {
-        Ok(c) => c,
-        Err(resp) => return resp,
-    };
+    let conn = state.db.lock().unwrap();
     let repo = flowctl_db::TaskRepo::new(&conn);
     match repo.update_status(&body.task_id, flowctl_core::state_machine::Status::Done) {
         Ok(()) => (StatusCode::OK, Json(serde_json::json!({"success": true, "id": body.task_id}))),
@@ -229,16 +173,9 @@ pub struct TaskIdRequest {
     pub task_id: String,
 }
 
-/// Helper: open a DB connection from daemon state.
-fn open_db(state: &DaemonState) -> Result<rusqlite::Connection, (StatusCode, Json<serde_json::Value>)> {
-    let db_path = state.runtime.paths.state_dir.parent()
-        .map(|flow_dir| flow_dir.join("flowctl.db"));
-    let Some(db_path) = db_path else {
-        return Err((StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": "cannot resolve db path"}))));
-    };
-    flowctl_db::open(&db_path).map_err(|e| {
-        (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": format!("db open failed: {e}")})))
-    })
+/// Helper: acquire DB lock from shared state.
+fn db_error(msg: &str) -> (StatusCode, Json<serde_json::Value>) {
+    (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"error": msg})))
 }
 
 /// GET /api/v1/events -- WebSocket upgrade for live event streaming.

flowctl/crates/flowctl-daemon/src/server.rs

@@ -17,6 +17,22 @@ use crate::handlers::{
 };
 use crate::lifecycle::{set_socket_permissions, DaemonRuntime};
 
+/// Create shared app state with a DB connection.
+fn create_state(runtime: DaemonRuntime, event_bus: flowctl_scheduler::EventBus) -> Result<(AppState, tokio_util::sync::CancellationToken)> {
+    let db_path = runtime.paths.state_dir.parent()
+        .map(|flow_dir| flow_dir.join("flowctl.db"))
+        .context("cannot resolve db path")?;
+    let conn = flowctl_db::open(&db_path)
+        .with_context(|| format!("failed to open db: {}", db_path.display()))?;
+    let cancel = runtime.cancel.clone();
+    let state = Arc::new(DaemonState {
+        runtime,
+        event_bus,
+        db: std::sync::Mutex::new(conn),
+    });
+    Ok((state, cancel))
+}
+
 /// Build the Axum router with all daemon API routes.
 fn build_router(state: AppState) -> axum::Router {
     let cors = CorsLayer::new()
@@ -54,12 +70,7 @@ pub async fn serve(runtime: DaemonRuntime, event_bus: flowctl_scheduler::EventBu
 
     info!("daemon API listening on {}", socket_path.display());
 
-    let cancel = runtime.cancel.clone();
-
-    let state: AppState = Arc::new(DaemonState {
-        runtime,
-        event_bus,
-    });
+    let (state, cancel) = create_state(runtime, event_bus)?;
 
     let router = build_router(state);
 
@@ -87,12 +98,7 @@ pub async fn serve_tcp(
 
     info!("daemon API listening on http://{addr}");
 
-    let cancel = runtime.cancel.clone();
-
-    let state: AppState = Arc::new(DaemonState {
-        runtime,
-        event_bus,
-    });
+    let (state, cancel) = create_state(runtime, event_bus)?;
 
     let router = build_router(state);
 
@@ -119,6 +125,8 @@ mod tests {
         let flow_dir = tmp.path().join(".flow");
         let paths = DaemonPaths::new(&flow_dir);
         paths.ensure_state_dir().unwrap();
+        // Create DB so create_state() works.
+        let _ = flowctl_db::open(&flow_dir);
         let runtime = DaemonRuntime::new(paths);
         let (event_bus, _critical_rx) = flowctl_scheduler::EventBus::with_default_capacity();
         (tmp, runtime, event_bus)

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

@@ -5,9 +5,9 @@
 //!
 //! # Architecture
 //!
-//! - **Markdown is canonical, SQLite is cache.** The `flowctl reindex`
-//!   command can fully rebuild the indexed tables from Markdown frontmatter.
-//!   Runtime-only data (locks, heartbeats, events, metrics) is not recoverable.
+//! - **SQLite is the single source of truth.** All reads and writes go through
+//!   the repository layer. Markdown files are an export format (`flowctl export`).
+//!   `flowctl import` (reindex) rebuilds the DB from Markdown for migration.
 //!
 //! - **PRAGMAs are per-connection**, not in migration files. WAL mode,
 //!   busy_timeout, and foreign_keys are set on every connection open.
@@ -22,7 +22,8 @@ pub mod metrics;
 pub mod migration;
 pub mod pool;
 pub mod repo;
-pub mod sync;
+#[allow(dead_code)]
+mod sync; // Legacy dual-write module, kept for backward compatibility but not re-exported.
 
 pub use error::DbError;
 pub use pool::{cleanup, open, open_memory, resolve_db_path, resolve_state_dir};
@@ -31,6 +32,5 @@ pub use migration::{migrate_runtime_state, needs_reindex, has_legacy_state, Migr
 pub use repo::{EpicRepo, EvidenceRepo, EventRepo, EventRow, FileLockRepo, PhaseProgressRepo, RuntimeRepo, TaskRepo};
 pub use events::{EventLog, TokenRecord};
 pub use metrics::StatsQuery;
-pub use sync::{write_epic, write_task, write_task_with_legacy, check_staleness, refresh_if_stale, retry_pending, SyncStatus};
 
 pub use flowctl_core;

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

@@ -1,4 +1,4 @@
-//! Bidirectional Markdown-SQLite sync.
+//! Legacy bidirectional Markdown-SQLite sync (deprecated).
 //!
 //! **Invariant**: SQLite is updated first (in a transaction), then Markdown
 //! frontmatter is written. If the Markdown write fails after SQLite commit,