Merge branch 'fn-25-complete-json-migration': service + daemon fully migrated to json_store

Author: z23cc

flowctl/crates/flowctl-cli/tests/parity_test.rs

@@ -367,29 +367,17 @@ fn setup_task(prefix: &str) -> (tempfile::TempDir, String) {
 
 /// Read task status from the DB directly via async libSQL.
 #[allow(dead_code)]
-fn db_task_status(work_dir: &Path, task_id: &str) -> String {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .unwrap();
-    rt.block_on(async {
-        let db = flowctl_db::open_async(work_dir).await.expect("open db");
-        let conn = db.connect().expect("connect");
-        let repo = flowctl_db::TaskRepo::new(conn);
-        let task = repo.get(task_id).await.expect("get task");
-        task.status.to_string()
-    })
+fn json_task_status(work_dir: &Path, task_id: &str) -> String {
+    let flow_dir = work_dir.join(".flow");
+    let task = flowctl_core::json_store::task_read(&flow_dir, task_id).expect("read task");
+    task.status.to_string()
 }
 
-// Removed: rusqlite parity tests (fn-19 migration complete). The service
-// layer is now async libSQL end-to-end; the original parity placeholders
-// have been deleted.
-
 #[test]
 fn parity_service_round_trip() {
     // Smoke test: create an epic+task via the CLI, then read it back via
-    // the async libsql repo. Mirrors what the old parity tests checked.
+    // json_store. Verifies CLI writes JSON files correctly.
     let (dir, task_id) = setup_task("parity-rt");
-    let status = db_task_status(dir.path(), &task_id);
+    let status = json_task_status(dir.path(), &task_id);
     assert_eq!(status, "todo", "newly created task should be todo");
 }

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

@@ -43,11 +43,19 @@ pub struct TaskState {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub claimed_at: Option<DateTime<Utc>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub completed_at: Option<DateTime<Utc>>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub evidence: Option<Evidence>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub blocked_reason: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub duration_seconds: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub baseline_rev: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub final_rev: Option<String>,
+    #[serde(default)]
+    pub retry_count: u32,
     #[serde(default = "Utc::now")]
     pub updated_at: DateTime<Utc>,
 }
@@ -58,9 +66,13 @@ impl Default for TaskState {
             status: Status::Todo,
             assignee: None,
             claimed_at: None,
+            completed_at: None,
             evidence: None,
             blocked_reason: None,
             duration_seconds: None,
+            baseline_rev: None,
+            final_rev: None,
+            retry_count: 0,
             updated_at: Utc::now(),
         }
     }

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

@@ -64,7 +64,7 @@ pub async fn start_task_handler(
 ) -> Result<Json<serde_json::Value>, AppError> {
     let task_id = body.task_id.clone();
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = StartTaskRequest {
         task_id,
         force: false,
@@ -98,7 +98,7 @@ pub async fn start_task_rest_handler(
     let force = body.force.unwrap_or(false);
     let actor = body.actor.unwrap_or_else(|| "daemon".to_string());
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = StartTaskRequest { task_id, force, actor };
 
     match flowctl_service::lifecycle::start_task(Some(&conn), &flow_dir, req).await {
@@ -127,7 +127,7 @@ pub async fn done_task_rest_handler(
 ) -> Result<Json<serde_json::Value>, AppError> {
     let body = body.unwrap_or_default();
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = DoneTaskRequest {
         task_id,
         summary: body.summary,
@@ -165,7 +165,7 @@ pub async fn block_task_rest_handler(
 ) -> Result<Json<serde_json::Value>, AppError> {
     let reason = body.reason;
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = BlockTaskRequest { task_id, reason };
 
     match flowctl_service::lifecycle::block_task(Some(&conn), &flow_dir, req).await {
@@ -195,7 +195,7 @@ pub async fn restart_task_rest_handler(
     let body = body.unwrap_or_default();
     let force = body.force.unwrap_or(true);
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = RestartTaskRequest {
         task_id,
         dry_run: false,
@@ -263,7 +263,7 @@ pub async fn done_task_handler(
     let task_id = body.task_id.clone();
     let summary = body.summary.clone();
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = DoneTaskRequest {
         task_id,
         summary,
@@ -333,7 +333,7 @@ pub async fn block_task_handler(
     let task_id = body.task_id.clone();
     let reason = body.reason.clone();
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = BlockTaskRequest { task_id, reason };
 
     match flowctl_service::lifecycle::block_task(Some(&conn), &flow_dir, req).await {
@@ -362,7 +362,7 @@ pub async fn restart_task_handler(
 ) -> Result<Json<serde_json::Value>, AppError> {
     let task_id = body.task_id.clone();
     let conn = state.db.clone();
-    let flow_dir = flow_dir();
+    let flow_dir = state.flow_dir.clone();
     let req = RestartTaskRequest {
         task_id,
         dry_run: false,

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

@@ -378,18 +378,29 @@ mod tests {
 
     #[tokio::test]
     async fn start_task_validates_transition() {
-        // Setup: create epic + task in todo state, then start it (should succeed),
-        // then try to start again from in_progress (should fail with CONFLICT).
         let (_tmp, runtime, event_bus) = test_setup();
         let (state, _cancel) = create_state(runtime, event_bus).await.unwrap();
-        state.db.execute(
-            "INSERT INTO epics (id, title, status, file_path, created_at, updated_at) VALUES ('fn-1', 'E', 'open', 'e.md', '2025-01-01T00:00:00Z', '2025-01-01T00:00:00Z')",
-            (),
-        ).await.unwrap();
-        state.db.execute(
-            "INSERT INTO tasks (id, epic_id, title, status, domain, file_path, created_at, updated_at) VALUES ('fn-1.1', 'fn-1', 'T', 'todo', 'general', 't.md', '2025-01-01T00:00:00Z', '2025-01-01T00:00:00Z')",
-            (),
-        ).await.unwrap();
+        // Create epic + task via json_store
+        use chrono::Utc;
+        let epic = flowctl_core::types::Epic {
+            schema_version: 1, id: "fn-1".into(), title: "E".into(),
+            status: flowctl_core::types::EpicStatus::Open,
+            branch_name: None, plan_review: flowctl_core::types::ReviewStatus::Unknown,
+            completion_review: flowctl_core::types::ReviewStatus::Unknown,
+            depends_on_epics: vec![], default_impl: None, default_review: None,
+            default_sync: None, auto_execute_pending: None, auto_execute_set_at: None,
+            archived: false, file_path: None, created_at: Utc::now(), updated_at: Utc::now(),
+        };
+        flowctl_core::json_store::epic_write(&state.flow_dir, &epic).unwrap();
+        let task = flowctl_core::types::Task {
+            schema_version: 1, id: "fn-1.1".into(), epic: "fn-1".into(), title: "T".into(),
+            status: flowctl_core::Status::Todo, priority: None,
+            domain: flowctl_core::types::Domain::General, depends_on: vec![],
+            files: vec![], r#impl: None, review: None, sync: None, file_path: None,
+            created_at: Utc::now(), updated_at: Utc::now(),
+        };
+        flowctl_core::json_store::task_write_definition(&state.flow_dir, &task).unwrap();
+        flowctl_core::json_store::state_write(&state.flow_dir, "fn-1.1", &flowctl_core::json_store::TaskState::default()).unwrap();
         let app = build_router(state.clone());
 
         // Start: todo → in_progress (should succeed)
@@ -429,14 +440,26 @@ mod tests {
     async fn done_task_rejects_from_todo() {
         let (_tmp, runtime, event_bus) = test_setup();
         let (state, _cancel) = create_state(runtime, event_bus).await.unwrap();
-        state.db.execute(
-            "INSERT INTO epics (id, title, status, file_path, created_at, updated_at) VALUES ('fn-2', 'E', 'open', 'e.md', '2025-01-01T00:00:00Z', '2025-01-01T00:00:00Z')",
-            (),
-        ).await.unwrap();
-        state.db.execute(
-            "INSERT INTO tasks (id, epic_id, title, status, domain, file_path, created_at, updated_at) VALUES ('fn-2.1', 'fn-2', 'T', 'todo', 'general', 't.md', '2025-01-01T00:00:00Z', '2025-01-01T00:00:00Z')",
-            (),
-        ).await.unwrap();
+        use chrono::Utc;
+        let epic = flowctl_core::types::Epic {
+            schema_version: 1, id: "fn-2".into(), title: "E".into(),
+            status: flowctl_core::types::EpicStatus::Open,
+            branch_name: None, plan_review: flowctl_core::types::ReviewStatus::Unknown,
+            completion_review: flowctl_core::types::ReviewStatus::Unknown,
+            depends_on_epics: vec![], default_impl: None, default_review: None,
+            default_sync: None, auto_execute_pending: None, auto_execute_set_at: None,
+            archived: false, file_path: None, created_at: Utc::now(), updated_at: Utc::now(),
+        };
+        flowctl_core::json_store::epic_write(&state.flow_dir, &epic).unwrap();
+        let task = flowctl_core::types::Task {
+            schema_version: 1, id: "fn-2.1".into(), epic: "fn-2".into(), title: "T".into(),
+            status: flowctl_core::Status::Todo, priority: None,
+            domain: flowctl_core::types::Domain::General, depends_on: vec![],
+            files: vec![], r#impl: None, review: None, sync: None, file_path: None,
+            created_at: Utc::now(), updated_at: Utc::now(),
+        };
+        flowctl_core::json_store::task_write_definition(&state.flow_dir, &task).unwrap();
+        flowctl_core::json_store::state_write(&state.flow_dir, "fn-2.1", &flowctl_core::json_store::TaskState::default()).unwrap();
         let app = build_router(state);
 
         // done from todo → should be rejected