feat(cli): migrate task module from DB to json_store

Replace all db_shim/TaskRepo/EpicRepo calls in task/mod.rs, create.rs,
mutate.rs, and query.rs with json_store file operations. Task helpers
(load_task_md, load_task_doc, write_task_doc, scan_max_task_id,
find_dependents) now read/write JSON files instead of SQLite.

Add epic_max_num() and task_list_all() to json_store module.

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

Author: z23cc

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

@@ -7,11 +7,11 @@ use crate::output::{error_exit, json_output};
 
 use flowctl_core::id::{epic_id_from_task, is_epic_id, is_task_id};
 use flowctl_core::state_machine::Status;
-use flowctl_core::types::{Domain, Task, FLOW_DIR, SPECS_DIR};
+use flowctl_core::types::{Domain, Task, FLOW_DIR};
 
 use super::{
     create_task_spec, ensure_flow_exists, parse_domain, read_file_or_stdin, scan_max_task_id,
-    require_db, write_task_doc,
+    write_task_doc,
 };
 
 #[allow(clippy::too_many_arguments)]
@@ -34,33 +34,21 @@ pub(super) fn cmd_task_create(
         ));
     }
 
-    // Verify epic exists (DB is authoritative)
-    if let Ok(conn) = require_db() {
-        let repo = crate::commands::db_shim::EpicRepo::new(&conn);
-        if repo.get(epic_id).is_err() {
-            error_exit(&format!("Epic {} not found", epic_id));
-        }
-    } else {
-        // Fallback: check spec file if DB unavailable
-        let epic_spec = flow_dir.join(SPECS_DIR).join(format!("{}.md", epic_id));
-        if !epic_spec.exists() {
-            error_exit(&format!("Epic {} not found", epic_id));
-        }
+    // Verify epic exists (JSON file)
+    if flowctl_core::json_store::epic_read(&flow_dir, epic_id).is_err() {
+        error_exit(&format!("Epic {} not found", epic_id));
     }
 
     // Scan-based ID allocation
     let task_num = scan_max_task_id(&flow_dir, epic_id) + 1;
     let task_id = format!("{}.{}", epic_id, task_num);
 
-    // Check no collision in DB
-    if let Ok(conn) = require_db() {
-        let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        if repo.get(&task_id).is_ok() {
-            error_exit(&format!(
-                "Refusing to overwrite existing task {}. Check for orphaned entries.",
-                task_id
-            ));
-        }
+    // Check no collision
+    if flowctl_core::json_store::task_read(&flow_dir, &task_id).is_ok() {
+        error_exit(&format!(
+            "Refusing to overwrite existing task {}. Check for orphaned entries.",
+            task_id
+        ));
     }
 
     // Parse dependencies
@@ -129,13 +117,19 @@ pub(super) fn cmd_task_create(
     // Create spec markdown body
     let body = create_task_spec(&task_id, title, acceptance.as_deref());
 
-    // Write to DB via write_task_doc (DB-only, no MD export)
+    // Write task definition + spec + initial state
     let doc = flowctl_core::types::Document {
         frontmatter: task,
         body,
     };
     write_task_doc(&flow_dir, &task_id, &doc);
 
+    // Write initial runtime state
+    let initial_state = flowctl_core::json_store::TaskState::default();
+    if let Err(e) = flowctl_core::json_store::state_write(&flow_dir, &task_id, &initial_state) {
+        error_exit(&format!("Failed to write initial state: {e}"));
+    }
+
     let spec_path_str = format!("{}/tasks/{}.md", FLOW_DIR, task_id);
     if json_mode {
         json_output(json!({

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

@@ -130,11 +130,6 @@ fn ensure_flow_exists() -> PathBuf {
     flow_dir
 }
 
-/// Open DB connection (hard error if unavailable).
-fn require_db() -> Result<crate::commands::db_shim::Connection, crate::commands::db_shim::DbError> {
-    crate::commands::db_shim::require_db()
-}
-
 /// Read file content, or read from stdin if path is "-".
 fn read_file_or_stdin(path: &str) -> String {
     if path == "-" {
@@ -149,12 +144,9 @@ fn read_file_or_stdin(path: &str) -> String {
     }
 }
 
-/// Get max task number for an epic from DB. Returns 0 if none exist.
-fn scan_max_task_id(_flow_dir: &Path, epic_id: &str) -> u32 {
-    let conn = require_db()
-        .unwrap_or_else(|e| error_exit(&format!("DB required for scan_max_task_id: {e}")));
-    let n = crate::commands::db_shim::max_task_num(&conn, epic_id).unwrap_or(0);
-    n as u32
+/// Get max task number for an epic from JSON files.
+fn scan_max_task_id(flow_dir: &Path, epic_id: &str) -> u32 {
+    flowctl_core::json_store::task_max_num(flow_dir, epic_id).unwrap_or(0)
 }
 
 /// Parse a domain string into a Domain enum.
@@ -179,42 +171,36 @@ fn create_task_spec(id: &str, title: &str, acceptance: Option<&str>) -> String {
     )
 }
 
-/// Load a task from DB (no MD fallback).
-fn load_task_md(_flow_dir: &Path, task_id: &str) -> Task {
-    let conn = require_db()
-        .unwrap_or_else(|e| error_exit(&format!("DB required: {e}")));
-    let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-    repo.get(task_id)
+/// Load a task from JSON files.
+fn load_task_md(flow_dir: &Path, task_id: &str) -> Task {
+    flowctl_core::json_store::task_read(flow_dir, task_id)
         .unwrap_or_else(|_| error_exit(&format!("Task {} not found", task_id)))
 }
 
-/// Load an epic from DB (no MD fallback).
-fn load_epic_md(_flow_dir: &Path, epic_id: &str) -> Option<Epic> {
-    let conn = require_db().ok()?;
-    let repo = crate::commands::db_shim::EpicRepo::new(&conn);
-    repo.get(epic_id).ok()
+/// Load an epic from JSON files.
+fn load_epic_md(flow_dir: &Path, epic_id: &str) -> Option<Epic> {
+    flowctl_core::json_store::epic_read(flow_dir, epic_id).ok()
 }
 
-/// Load task's full document (frontmatter + body) from DB (no MD fallback).
-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)
+/// Load task's full document (definition + spec body) from JSON files.
+fn load_task_doc(flow_dir: &Path, task_id: &str) -> Document<Task> {
+    let task = flowctl_core::json_store::task_read(flow_dir, task_id)
         .unwrap_or_else(|_| error_exit(&format!("Task {} not found", task_id)));
+    let body = flowctl_core::json_store::task_spec_read(flow_dir, task_id)
+        .unwrap_or_default();
     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: &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);
-    if let Err(e) = repo.upsert_with_body(&doc.frontmatter, &doc.body) {
-        error_exit(&format!("DB write failed for {task_id}: {e}"));
+/// Write a task document to JSON files (definition + spec body).
+fn write_task_doc(flow_dir: &Path, _task_id: &str, doc: &Document<Task>) {
+    if let Err(e) = flowctl_core::json_store::task_write_definition(flow_dir, &doc.frontmatter) {
+        error_exit(&format!("Failed to write task definition: {e}"));
+    }
+    if let Err(e) = flowctl_core::json_store::task_spec_write(flow_dir, &doc.frontmatter.id, &doc.body) {
+        error_exit(&format!("Failed to write task spec: {e}"));
     }
 }
 
@@ -264,21 +250,15 @@ fn patch_body_section(body: &str, section: &str, new_content: &str) -> String {
     result.join("\n")
 }
 
-/// Find tasks that depend on a given task (recursive BFS within same epic) via DB.
-fn find_dependents(_flow_dir: &Path, task_id: &str) -> Vec<String> {
+/// Find tasks that depend on a given task (recursive BFS within same epic) via JSON files.
+fn find_dependents(flow_dir: &Path, task_id: &str) -> Vec<String> {
     let epic_id = match epic_id_from_task(task_id) {
         Ok(id) => id,
         Err(_) => return vec![],
     };
 
-    let conn = match require_db() {
-        Ok(c) => c,
-        Err(_) => return vec![],
-    };
-
-    // Load all tasks in the epic from DB
-    let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-    let all_tasks: Vec<(String, Vec<String>)> = match repo.list_by_epic(&epic_id) {
+    // Load all tasks in the epic from JSON files
+    let all_tasks: Vec<(String, Vec<String>)> = match flowctl_core::json_store::task_list_by_epic(flow_dir, &epic_id) {
         Ok(tasks) => tasks.into_iter().map(|t| (t.id.clone(), t.depends_on.clone())).collect(),
         Err(_) => return vec![],
     };

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

@@ -11,7 +11,7 @@ use flowctl_core::types::{Task, FLOW_DIR};
 
 use super::{
     clear_evidence_in_body, create_task_spec, ensure_flow_exists, find_dependents, load_epic_md,
-    load_task_doc, scan_max_task_id, require_db, write_task_doc,
+    load_task_doc, scan_max_task_id, write_task_doc,
 };
 
 pub(super) fn cmd_task_reset(json_mode: bool, task_id: &str, cascade: bool) {
@@ -61,18 +61,9 @@ pub(super) fn cmd_task_reset(json_mode: bool, task_id: &str, cascade: bool) {
     doc.body = clear_evidence_in_body(&doc.body);
     write_task_doc(&flow_dir, task_id, &doc);
 
-    // Update DB
-    if let Ok(conn) = require_db() {
-        let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        let _ = repo.update_status(task_id, Status::Todo);
-        // Clear runtime state by upserting a blank state
-        let runtime_repo = crate::commands::db_shim::RuntimeRepo::new(&conn);
-        let blank = flowctl_core::types::RuntimeState {
-            task_id: task_id.to_string(),
-            ..Default::default()
-        };
-        let _ = runtime_repo.upsert(&blank);
-    }
+    // Reset runtime state
+    let blank_state = flowctl_core::json_store::TaskState::default();
+    let _ = flowctl_core::json_store::state_write(&flow_dir, task_id, &blank_state);
 
     let mut reset_ids = vec![task_id.to_string()];
 
@@ -91,16 +82,8 @@ pub(super) fn cmd_task_reset(json_mode: bool, task_id: &str, cascade: bool) {
             dep_doc.body = clear_evidence_in_body(&dep_doc.body);
             write_task_doc(&flow_dir, dep_id, &dep_doc);
 
-            if let Ok(conn) = require_db() {
-                let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-                let _ = repo.update_status(dep_id, Status::Todo);
-                let runtime_repo = crate::commands::db_shim::RuntimeRepo::new(&conn);
-                let blank = flowctl_core::types::RuntimeState {
-                    task_id: dep_id.to_string(),
-                    ..Default::default()
-                };
-                let _ = runtime_repo.upsert(&blank);
-            }
+            let blank_state = flowctl_core::json_store::TaskState::default();
+            let _ = flowctl_core::json_store::state_write(&flow_dir, dep_id, &blank_state);
             reset_ids.push(dep_id.clone());
         }
     }
@@ -232,33 +215,30 @@ pub(super) fn cmd_task_split(json_mode: bool, task_id: &str, titles: &str, chain
     orig_doc.frontmatter.updated_at = now;
     write_task_doc(&flow_dir, task_id, &orig_doc);
 
-    // Update tasks that depended on original to depend on last sub-task (via DB)
+    // Update tasks that depended on original to depend on last sub-task
     let last_sub = created.last().unwrap().clone();
-    if let Ok(conn) = require_db() {
-        let repo = crate::commands::db_shim::TaskRepo::new(&conn);
-        if let Ok(all_tasks) = repo.list_by_epic(&epic_id) {
-            for other_task in all_tasks {
-                let other_id = &other_task.id;
-                if other_id == task_id || created.contains(other_id) {
-                    continue;
-                }
-                if other_task.depends_on.contains(&task_id.to_string()) {
-                    let mut other_doc = load_task_doc(&flow_dir, other_id);
-                    other_doc.frontmatter.depends_on = other_doc
-                        .frontmatter
-                        .depends_on
-                        .iter()
-                        .map(|d| {
-                            if d == task_id {
-                                last_sub.clone()
-                            } else {
-                                d.clone()
-                            }
-                        })
-                        .collect();
-                    other_doc.frontmatter.updated_at = now;
-                    write_task_doc(&flow_dir, other_id, &other_doc);
-                }
+    if let Ok(all_tasks) = flowctl_core::json_store::task_list_by_epic(&flow_dir, &epic_id) {
+        for other_task in all_tasks {
+            let other_id = &other_task.id;
+            if other_id == task_id || created.contains(other_id) {
+                continue;
+            }
+            if other_task.depends_on.contains(&task_id.to_string()) {
+                let mut other_doc = load_task_doc(&flow_dir, other_id);
+                other_doc.frontmatter.depends_on = other_doc
+                    .frontmatter
+                    .depends_on
+                    .iter()
+                    .map(|d| {
+                        if d == task_id {
+                            last_sub.clone()
+                        } else {
+                            d.clone()
+                        }
+                    })
+                    .collect();
+                other_doc.frontmatter.updated_at = now;
+                write_task_doc(&flow_dir, other_id, &other_doc);
             }
         }
     }

flowctl/crates/flowctl-core/src/json_store.rs

@@ -336,6 +336,56 @@ pub fn state_write(flow_dir: &Path, task_id: &str, state: &TaskState) -> Result<
     Ok(())
 }
 
+/// Find the highest epic number by scanning `epics/fn-N-*.json`.
+pub fn epic_max_num(flow_dir: &Path) -> Result<u32> {
+    let dir = epics_dir(flow_dir);
+    if !dir.exists() {
+        return Ok(0);
+    }
+    let mut max = 0u32;
+    for entry in fs::read_dir(&dir)? {
+        let entry = entry?;
+        let name = entry.file_name();
+        let name = name.to_str().unwrap_or("");
+        if name.starts_with("fn-") && name.ends_with(".json") {
+            // Extract N from "fn-N-slug.json"
+            let without_prefix = name.trim_start_matches("fn-");
+            if let Some(num_str) = without_prefix.split('-').next() {
+                if let Ok(num) = num_str.parse::<u32>() {
+                    max = max.max(num);
+                }
+            }
+        }
+    }
+    Ok(max)
+}
+
+/// List all tasks across all epics.
+pub fn task_list_all(flow_dir: &Path) -> Result<Vec<Task>> {
+    let dir = tasks_dir(flow_dir);
+    if !dir.exists() {
+        return Ok(vec![]);
+    }
+    let mut tasks = Vec::new();
+    for entry in fs::read_dir(&dir)? {
+        let entry = entry?;
+        let path = entry.path();
+        let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
+        if name.ends_with(".json") {
+            let content = fs::read_to_string(&path)?;
+            if let Ok(mut task) = serde_json::from_str::<Task>(&content) {
+                if let Ok(state) = state_read(flow_dir, &task.id) {
+                    task.status = state.status;
+                    task.updated_at = state.updated_at;
+                }
+                tasks.push(task);
+            }
+        }
+    }
+    tasks.sort_by(|a, b| a.id.cmp(&b.id));
+    Ok(tasks)
+}
+
 // ── Gap operations ──────────────────────────────────────────────────
 
 /// Gap entry stored in `gaps/<epic-id>.json`.