feat(db): cleanup residuals + update tests — complete DB-sole-truth migration

- Remove EPICS_DIR/TASKS_DIR from CRUD paths (only in init/detect/exchange)
- Remove frontmatter imports from epic/task command modules
- Fix query.rs scan fallback removal
- Update TOML test expectations (next_json, validate_json)
- Zero try_open_db references remaining

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

Author: z23cc

flowctl/crates/flowctl-cli/src/commands/admin/status.rs

@@ -16,10 +16,12 @@ use flowctl_core::types::{
 
 use super::get_flow_dir;
 
-/// Try to open a DB connection. Returns None if DB doesn't exist or can't be opened.
-fn try_open_db() -> Option<crate::commands::db_shim::Connection> {
-    let cwd = env::current_dir().ok()?;
-    crate::commands::db_shim::open(&cwd).ok()
+/// Open DB connection (hard error on failure, DB is sole source of truth).
+fn require_db() -> crate::commands::db_shim::Connection {
+    crate::commands::db_shim::require_db()
+        .unwrap_or_else(|e| {
+            crate::output::error_exit(&format!("Cannot open database: {e}"));
+        })
 }
 
 // ── Status command ──────────────────────────────────────────────────
@@ -221,9 +223,9 @@ fn is_daemon_heartbeat_alive(flow_dir: &Path) -> bool {
 fn find_interrupted_epics(_flow_dir: &Path) -> Vec<serde_json::Value> {
     let mut interrupted = Vec::new();
 
-    let conn = match try_open_db() {
-        Some(c) => c,
-        None => return interrupted,
+    let conn = match crate::commands::db_shim::require_db() {
+        Ok(c) => c,
+        Err(_) => return interrupted,
     };
 
     let epic_repo = crate::commands::db_shim::EpicRepo::new(&conn);
@@ -339,20 +341,18 @@ pub(super) fn validate_epic(flow_dir: &Path, epic_id: &str) -> (Vec<String>, Vec
     let mut tasks: std::collections::HashMap<String, flowctl_core::types::Task> =
         std::collections::HashMap::new();
 
-    if let Some(conn) = try_open_db() {
-        let task_repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        if let Ok(task_list) = task_repo.list_by_epic(epic_id) {
-            for task in task_list {
-                tasks.insert(task.id.clone(), task);
-            }
+    let conn = require_db();
+    let task_repo = crate::commands::db_shim::TaskRepo::new(&conn);
+    if let Ok(task_list) = task_repo.list_by_epic(epic_id) {
+        for task in task_list {
+            tasks.insert(task.id.clone(), task);
         }
     }
 
     // Validate each task
     for (task_id, task) in &tasks {
         // Validate task body has required headings (read from DB)
-        if let Some(conn) = try_open_db() {
-            let task_repo = crate::commands::db_shim::TaskRepo::new(&conn);
+        {
             match task_repo.get_with_body(task_id) {
                 Ok((_t, body)) => {
                     if body.is_empty() {
@@ -430,13 +430,13 @@ pub fn cmd_validate(json_mode: bool, epic: Option<String>, all: bool) {
         // Validate all epics
         let root_errors = validate_flow_root(&flow_dir);
 
-        let mut epic_ids: Vec<String> = Vec::new();
-        if let Some(conn) = try_open_db() {
-            let epic_repo = crate::commands::db_shim::EpicRepo::new(&conn);
-            if let Ok(epics) = epic_repo.list(None) {
-                epic_ids = epics.into_iter().map(|e| e.id).collect();
-            }
-        }
+        let conn = require_db();
+        let epic_repo = crate::commands::db_shim::EpicRepo::new(&conn);
+        let mut epic_ids: Vec<String> = epic_repo.list(None)
+            .unwrap_or_default()
+            .into_iter()
+            .map(|e| e.id)
+            .collect();
         epic_ids.sort();
 
         let mut all_errors: Vec<String> = root_errors.clone();
@@ -552,7 +552,8 @@ pub fn cmd_doctor(json_mode: bool) {
     let root_errors = validate_flow_root(&flow_dir);
     let mut validate_errors = root_errors.clone();
 
-    if let Some(conn) = try_open_db() {
+    {
+        let conn = require_db();
         let epic_repo = crate::commands::db_shim::EpicRepo::new(&conn);
         if let Ok(epics) = epic_repo.list(None) {
             for epic in &epics {

flowctl/crates/flowctl-cli/src/commands/db_shim.rs

@@ -62,8 +62,8 @@ pub fn open(working_dir: &Path) -> Result<Connection, DbError> {
 }
 
 /// Open DB connection with hard error on failure (DB must be available).
-/// This is the preferred entry point — replaces `try_open_db()` which
-/// returned `Option<Connection>` and silently degraded to MD fallback.
+/// This is the preferred entry point — all CLI code should use this
+/// (DB is the sole source of truth, no fallback path).
 pub fn require_db() -> Result<Connection, DbError> {
     let cwd = std::env::current_dir()
         .map_err(|e| DbError::StateDir(format!("cannot get current dir: {e}")))?;

flowctl/crates/flowctl-cli/src/commands/epic.rs

@@ -12,10 +12,9 @@ use serde_json::json;
 
 use crate::output::{error_exit, json_output};
 
-use flowctl_core::frontmatter;
 use flowctl_core::id::{generate_epic_suffix, is_epic_id, parse_id, slugify};
 use flowctl_core::types::{
-    Epic, EpicStatus, ReviewStatus, ARCHIVE_DIR, FLOW_DIR, META_FILE,
+    Document, Epic, EpicStatus, ReviewStatus, ARCHIVE_DIR, FLOW_DIR, META_FILE,
     REVIEWS_DIR, SPECS_DIR,
 };
 
@@ -169,18 +168,18 @@ fn validate_epic_id(id: &str) {
 }
 
 /// Load epic document from DB (sole source of truth).
-fn load_epic(id: &str) -> frontmatter::Document<Epic> {
+fn load_epic(id: &str) -> Document<Epic> {
     let conn = require_db()
         .unwrap_or_else(|e| error_exit(&format!("DB required: {e}")));
     let repo = crate::commands::db_shim::EpicRepo::new(&conn);
     match repo.get_with_body(id) {
-        Ok((epic, body)) => frontmatter::Document { frontmatter: epic, body },
+        Ok((epic, body)) => Document { frontmatter: epic, body },
         Err(_) => error_exit(&format!("Epic {id} not found")),
     }
 }
 
 /// Write an epic document to DB (sole source of truth, no MD export).
-fn save_epic(doc: &frontmatter::Document<Epic>) {
+fn save_epic(doc: &Document<Epic>) {
     let conn = require_db()
         .unwrap_or_else(|e| error_exit(&format!("DB required: {e}")));
     let repo = crate::commands::db_shim::EpicRepo::new(&conn);
@@ -297,7 +296,7 @@ fn cmd_create(title: &str, branch: &Option<String>, json_mode: bool) {
 
     // Write to DB (sole source of truth)
     let body = create_epic_spec_body(&epic_id, title);
-    let doc = frontmatter::Document {
+    let doc = Document {
         frontmatter: epic,
         body: body.clone(),
     };

flowctl/crates/flowctl-cli/src/commands/query.rs

@@ -2,7 +2,6 @@
 //!
 //! Reads from SQLite as the sole source of truth.
 
-use std::env;
 use std::fs;
 use std::path::PathBuf;
 
@@ -28,12 +27,6 @@ fn ensure_flow_exists() -> PathBuf {
     flow_dir
 }
 
-/// Try to open a DB connection. Returns None if DB doesn't exist or can't be opened.
-fn try_open_db() -> Option<crate::commands::db_shim::Connection> {
-    let cwd = env::current_dir().ok()?;
-    crate::commands::db_shim::open(&cwd).ok()
-}
-
 /// Serialize an Epic to the JSON format matching Python output.
 fn epic_to_json(epic: &Epic) -> serde_json::Value {
     let spec_path = format!(".flow/specs/{}.md", epic.id);
@@ -65,15 +58,14 @@ fn task_to_json(task: &Task) -> serde_json::Value {
     let mut claimed_at: serde_json::Value = json!(null);
     let claim_note: serde_json::Value = json!("");
 
-    if let Some(conn) = try_open_db() {
-        let runtime_repo = crate::commands::db_shim::RuntimeRepo::new(&conn);
-        if let Ok(Some(state)) = runtime_repo.get(&task.id) {
-            if let Some(a) = &state.assignee {
-                assignee = json!(a);
-            }
-            if let Some(ca) = &state.claimed_at {
-                claimed_at = json!(ca.to_rfc3339());
-            }
+    let conn = require_db();
+    let runtime_repo = crate::commands::db_shim::RuntimeRepo::new(&conn);
+    if let Ok(Some(state)) = runtime_repo.get(&task.id) {
+        if let Some(a) = &state.assignee {
+            assignee = json!(a);
+        }
+        if let Some(ca) = &state.claimed_at {
+            claimed_at = json!(ca.to_rfc3339());
         }
     }
 
@@ -132,38 +124,30 @@ fn require_db() -> crate::commands::db_shim::Connection {
 
 /// Get a single epic by ID from DB.
 fn get_epic(_flow_dir: &PathBuf, id: &str) -> Option<Epic> {
-    let conn = try_open_db()?;
+    let conn = require_db();
     let repo = crate::commands::db_shim::EpicRepo::new(&conn);
     repo.get(id).ok()
 }
 
 /// Get a single task by ID from DB.
 fn get_task(_flow_dir: &PathBuf, id: &str) -> Option<Task> {
-    let conn = try_open_db()?;
+    let conn = require_db();
     let repo = crate::commands::db_shim::TaskRepo::new(&conn);
     repo.get(id).ok()
 }
 
 /// Get all tasks for an epic from DB.
 fn get_epic_tasks(_flow_dir: &PathBuf, epic_id: &str) -> Vec<Task> {
-    if let Some(conn) = try_open_db() {
-        let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        if let Ok(tasks) = repo.list_by_epic(epic_id) {
-            return tasks;
-        }
-    }
-    Vec::new()
+    let conn = require_db();
+    let repo = crate::commands::db_shim::TaskRepo::new(&conn);
+    repo.list_by_epic(epic_id).unwrap_or_default()
 }
 
 /// Get all epics from DB.
 fn get_all_epics(_flow_dir: &PathBuf) -> Vec<Epic> {
-    if let Some(conn) = try_open_db() {
-        let repo = crate::commands::db_shim::EpicRepo::new(&conn);
-        if let Ok(epics) = repo.list(None) {
-            return epics;
-        }
-    }
-    Vec::new()
+    let conn = require_db();
+    let repo = crate::commands::db_shim::EpicRepo::new(&conn);
+    repo.list(None).unwrap_or_default()
 }
 
 /// Get all tasks, optionally filtered, from DB.
@@ -173,28 +157,21 @@ fn get_all_tasks(
     status_filter: Option<&str>,
     domain_filter: Option<&str>,
 ) -> Vec<Task> {
-    if let Some(conn) = try_open_db() {
-        let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        match epic_filter {
-            Some(epic_id) => {
-                if let Ok(mut tasks) = repo.list_by_epic(epic_id) {
-                    if let Some(status) = status_filter {
-                        tasks.retain(|t| t.status.to_string() == status);
-                    }
-                    if let Some(domain) = domain_filter {
-                        tasks.retain(|t| t.domain.to_string() == domain);
-                    }
-                    return tasks;
-                }
+    let conn = require_db();
+    let repo = crate::commands::db_shim::TaskRepo::new(&conn);
+    match epic_filter {
+        Some(epic_id) => {
+            let mut tasks = repo.list_by_epic(epic_id).unwrap_or_default();
+            if let Some(status) = status_filter {
+                tasks.retain(|t| t.status.to_string() == status);
             }
-            None => {
-                if let Ok(tasks) = repo.list_all(status_filter, domain_filter) {
-                    return tasks;
-                }
+            if let Some(domain) = domain_filter {
+                tasks.retain(|t| t.domain.to_string() == domain);
             }
+            tasks
         }
+        None => repo.list_all(status_filter, domain_filter).unwrap_or_default(),
     }
-    Vec::new()
 }
 
 // ── Show command ────────────────────────────────────────────────────
@@ -550,24 +527,22 @@ pub fn cmd_files(json_mode: bool, epic: String) {
     let mut ownership: std::collections::BTreeMap<String, Vec<String>> =
         std::collections::BTreeMap::new();
 
-    let conn_opt = try_open_db();
+    let conn = require_db();
     for task in &tasks {
         let mut task_files: Vec<String> = task.files.clone();
 
         // Fallback: parse **Files:** from task body in DB if no structured files
         if task_files.is_empty() {
-            if let Some(ref conn) = conn_opt {
-                let repo = crate::commands::db_shim::TaskRepo::new(conn);
-                if let Ok((_t, body)) = repo.get_with_body(&task.id) {
-                    for line in body.lines() {
-                        if let Some(rest) = line.strip_prefix("**Files:**") {
-                            task_files = rest
-                                .split(',')
-                                .map(|f| f.trim().trim_matches('`').to_string())
-                                .filter(|f| !f.is_empty())
-                                .collect();
-                            break;
-                        }
+            let repo = crate::commands::db_shim::TaskRepo::new(&conn);
+            if let Ok((_t, body)) = repo.get_with_body(&task.id) {
+                for line in body.lines() {
+                    if let Some(rest) = line.strip_prefix("**Files:**") {
+                        task_files = rest
+                            .split(',')
+                            .map(|f| f.trim().trim_matches('`').to_string())
+                            .filter(|f| !f.is_empty())
+                            .collect();
+                        break;
                     }
                 }
             }

flowctl/crates/flowctl-cli/src/commands/task/create.rs

@@ -130,7 +130,7 @@ pub(super) fn cmd_task_create(
     let body = create_task_spec(&task_id, title, acceptance.as_deref());
 
     // Write to DB via write_task_doc (DB-only, no MD export)
-    let doc = flowctl_core::frontmatter::Document {
+    let doc = flowctl_core::types::Document {
         frontmatter: task,
         body,
     };

flowctl/crates/flowctl-cli/src/commands/task/mod.rs

@@ -13,10 +13,9 @@ use regex::Regex;
 
 use crate::output::error_exit;
 
-use flowctl_core::frontmatter;
 use flowctl_core::id::epic_id_from_task;
 use flowctl_core::types::{
-    Domain, Epic, Task,
+    Document, Domain, Epic, Task,
 };
 
 #[derive(Subcommand, Debug)]
@@ -194,20 +193,20 @@ fn load_epic_md(_flow_dir: &Path, epic_id: &str) -> Option<Epic> {
 }
 
 /// Load task's full document (frontmatter + body) from DB (no MD fallback).
-fn load_task_doc(_flow_dir: &Path, task_id: &str) -> frontmatter::Document<Task> {
+fn load_task_doc(_flow_dir: &Path, task_id: &str) -> Document<Task> {
     let conn = require_db()
         .unwrap_or_else(|e| error_exit(&format!("DB required: {e}")));
     let repo = crate::commands::db_shim::TaskRepo::new(&conn);
     let (task, body) = repo.get_with_body(task_id)
         .unwrap_or_else(|_| error_exit(&format!("Task {} not found", task_id)));
-    frontmatter::Document {
+    Document {
         frontmatter: task,
         body,
     }
 }
 
 /// Write a task document to DB only (no MD export).
-fn write_task_doc(_flow_dir: &Path, task_id: &str, doc: &frontmatter::Document<Task>) {
+fn write_task_doc(_flow_dir: &Path, task_id: &str, doc: &Document<Task>) {
     let conn = require_db()
         .unwrap_or_else(|e| error_exit(&format!("DB required for write: {e}")));
     let repo = crate::commands::db_shim::TaskRepo::new(&conn);

flowctl/crates/flowctl-cli/src/commands/task/mutate.rs

@@ -217,7 +217,7 @@ pub(super) fn cmd_task_split(json_mode: bool, task_id: &str, titles: &str, chain
         };
 
         let body = create_task_spec(&sub_id, sub_title, None);
-        let sub_doc = flowctl_core::frontmatter::Document {
+        let sub_doc = flowctl_core::types::Document {
             frontmatter: sub_task,
             body,
         };