feat(service+daemon): complete RuntimeState migration to json_store

Service lifecycle.rs: zero DB calls for task/epic/runtime. All status
updates, evidence, and runtime state now write to .state/tasks/*.json.
Extended TaskState with baseline_rev, final_rev, retry_count, completed_at.

Updated parity_test to read from json_store instead of DB.
Updated daemon server tests to create data via json_store.
2 daemon tests still fail (flow_dir() in handlers uses cwd, not state).

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

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/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

flowctl/crates/flowctl-service/src/lifecycle.rs

@@ -118,10 +118,20 @@ async fn load_epic(_conn: Option<&Connection>, flow_dir: &Path, epic_id: &str) -
     flowctl_core::json_store::epic_read(flow_dir, epic_id).ok()
 }
 
-async fn get_runtime(conn: Option<&Connection>, _flow_dir: &Path, task_id: &str) -> Option<RuntimeState> {
-    let conn = conn?;
-    let repo = flowctl_db::RuntimeRepo::new(conn.clone());
-    repo.get(task_id).await.ok().flatten()
+async fn get_runtime(_conn: Option<&Connection>, flow_dir: &Path, task_id: &str) -> Option<RuntimeState> {
+    // Read from JSON state and convert to RuntimeState for compatibility
+    let state = flowctl_core::json_store::state_read(flow_dir, task_id).ok()?;
+    Some(RuntimeState {
+        task_id: task_id.to_string(),
+        assignee: state.assignee,
+        claimed_at: state.claimed_at,
+        completed_at: state.completed_at,
+        duration_secs: state.duration_seconds,
+        blocked_reason: state.blocked_reason,
+        baseline_rev: state.baseline_rev,
+        final_rev: state.final_rev,
+        retry_count: state.retry_count,
+    })
 }
 
 /// Load all tasks for an epic from JSON files.
@@ -221,10 +231,17 @@ async fn propagate_upstream_failure(
             continue;
         }
 
-        // Update DB (sole source of truth)
-        if let Some(conn) = conn {
-            let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-            let _ = task_repo.update_status(tid, Status::UpstreamFailed).await;
+        // Update JSON state
+        if let Ok(mut state) = flowctl_core::json_store::state_read(flow_dir, tid) {
+            state.status = Status::UpstreamFailed;
+            state.updated_at = Utc::now();
+            let _ = flowctl_core::json_store::state_write(flow_dir, tid, &state);
+        } else {
+            let state = flowctl_core::json_store::TaskState {
+                status: Status::UpstreamFailed,
+                ..Default::default()
+            };
+            let _ = flowctl_core::json_store::state_write(flow_dir, tid, &state);
         }
 
         affected.push(tid.clone());
@@ -246,31 +263,28 @@ async fn handle_task_failure(
     if max_retries > 0 && current_retry_count < max_retries {
         let new_retry_count = current_retry_count + 1;
 
-        if let Some(conn) = conn {
-            let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-            let _ = task_repo.update_status(task_id, Status::UpForRetry).await;
-
-            let runtime_repo = flowctl_db::RuntimeRepo::new(conn.clone());
-            let rt = RuntimeState {
-                task_id: task_id.to_string(),
-                assignee: runtime.as_ref().and_then(|r| r.assignee.clone()),
-                claimed_at: None,
-                completed_at: None,
-                duration_secs: None,
-                blocked_reason: None,
-                baseline_rev: runtime.as_ref().and_then(|r| r.baseline_rev.clone()),
-                final_rev: None,
-                retry_count: new_retry_count,
-            };
-            let _ = runtime_repo.upsert(&rt).await;
-        }
+        let task_state = flowctl_core::json_store::TaskState {
+            status: Status::UpForRetry,
+            assignee: runtime.as_ref().and_then(|r| r.assignee.clone()),
+            claimed_at: None,
+            completed_at: None,
+            evidence: None,
+            blocked_reason: None,
+            duration_seconds: None,
+            baseline_rev: runtime.as_ref().and_then(|r| r.baseline_rev.clone()),
+            final_rev: None,
+            retry_count: new_retry_count,
+            updated_at: Utc::now(),
+        };
+        let _ = flowctl_core::json_store::state_write(flow_dir, task_id, &task_state);
 
         (Status::UpForRetry, Vec::new())
     } else {
-        if let Some(conn) = conn {
-            let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-            let _ = task_repo.update_status(task_id, Status::Failed).await;
-        }
+        let task_state = flowctl_core::json_store::TaskState {
+            status: Status::Failed,
+            ..Default::default()
+        };
+        let _ = flowctl_core::json_store::state_write(flow_dir, task_id, &task_state);
 
         let affected = propagate_upstream_failure(conn, flow_dir, task_id).await;
         (Status::Failed, affected)
@@ -373,13 +387,14 @@ pub async fn start_task(
         Some(now)
     };
 
-    let runtime_state = RuntimeState {
-        task_id: req.task_id.clone(),
+    let task_state = flowctl_core::json_store::TaskState {
+        status: Status::InProgress,
         assignee: Some(new_assignee),
         claimed_at,
         completed_at: None,
-        duration_secs: None,
+        evidence: None,
         blocked_reason: None,
+        duration_seconds: None,
         baseline_rev: existing_rt
             .as_ref()
             .and_then(|rt| rt.baseline_rev.clone()),
@@ -388,21 +403,12 @@ pub async fn start_task(
             .as_ref()
             .map(|rt| rt.retry_count)
             .unwrap_or(0),
+        updated_at: Utc::now(),
     };
 
-    // Write SQLite (authoritative)
-    if let Some(conn) = conn {
-        let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-        task_repo
-            .update_status(&req.task_id, Status::InProgress)
-            .await
-            .map_err(ServiceError::from)?;
-        let runtime_repo = flowctl_db::RuntimeRepo::new(conn.clone());
-        runtime_repo
-            .upsert(&runtime_state)
-            .await
-            .map_err(ServiceError::from)?;
-    }
+    // Write to JSON state file
+    flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state)
+        .map_err(|e| ServiceError::IoError(std::io::Error::new(std::io::ErrorKind::Other, e.to_string())))?;
 
     Ok(StartTaskResponse {
         task_id: req.task_id,
@@ -547,34 +553,29 @@ pub async fn done_task(
     let tests = to_list(evidence_obj.get("tests"));
     let prs = to_list(evidence_obj.get("prs"));
 
-    // Write SQLite (sole source of truth)
-    if let Some(conn) = conn {
-        let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-        let _ = task_repo.update_status(&req.task_id, Status::Done).await;
-
-        let runtime_repo = flowctl_db::RuntimeRepo::new(conn.clone());
+    // Write to JSON state file
+    {
         let now = Utc::now();
-        let rt = RuntimeState {
-            task_id: req.task_id.clone(),
+        let ev = Evidence {
+            commits: commits.clone(),
+            tests: tests.clone(),
+            prs: prs.clone(),
+            ..Evidence::default()
+        };
+        let task_state = flowctl_core::json_store::TaskState {
+            status: Status::Done,
             assignee: runtime.as_ref().and_then(|r| r.assignee.clone()),
             claimed_at: runtime.as_ref().and_then(|r| r.claimed_at),
             completed_at: Some(now),
-            duration_secs: duration_seconds,
+            evidence: Some(ev),
             blocked_reason: None,
+            duration_seconds: duration_seconds.map(|d| d as u64),
             baseline_rev: runtime.as_ref().and_then(|r| r.baseline_rev.clone()),
             final_rev: runtime.as_ref().and_then(|r| r.final_rev.clone()),
             retry_count: runtime.as_ref().map(|r| r.retry_count).unwrap_or(0),
+            updated_at: now,
         };
-        let _ = runtime_repo.upsert(&rt).await;
-
-        let ev = Evidence {
-            commits: commits.clone(),
-            tests: tests.clone(),
-            prs: prs.clone(),
-            ..Evidence::default()
-        };
-        let evidence_repo = flowctl_db::EvidenceRepo::new(conn.clone());
-        let _ = evidence_repo.upsert(&req.task_id, &ev).await;
+        let _ = flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state);
     }
 
     // Archive review receipt if present
@@ -632,25 +633,23 @@ pub async fn block_task(
         ));
     }
 
-    // Write SQLite (authoritative)
-    if let Some(conn) = conn {
-        let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-        let _ = task_repo.update_status(&req.task_id, Status::Blocked).await;
-
-        let runtime_repo = flowctl_db::RuntimeRepo::new(conn.clone());
-        let existing = runtime_repo.get(&req.task_id).await.ok().flatten();
-        let rt = RuntimeState {
-            task_id: req.task_id.clone(),
+    // Write to JSON state file
+    {
+        let existing = flowctl_core::json_store::state_read(flow_dir, &req.task_id).ok();
+        let task_state = flowctl_core::json_store::TaskState {
+            status: Status::Blocked,
             assignee: existing.as_ref().and_then(|r| r.assignee.clone()),
             claimed_at: existing.as_ref().and_then(|r| r.claimed_at),
             completed_at: None,
-            duration_secs: None,
+            evidence: existing.as_ref().and_then(|r| r.evidence.clone()),
             blocked_reason: Some(reason.clone()),
+            duration_seconds: None,
             baseline_rev: existing.as_ref().and_then(|r| r.baseline_rev.clone()),
             final_rev: None,
             retry_count: existing.as_ref().map(|r| r.retry_count).unwrap_or(0),
+            updated_at: Utc::now(),
         };
-        let _ = runtime_repo.upsert(&rt).await;
+        let _ = flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state);
     }
 
     Ok(BlockTaskResponse {
@@ -788,24 +787,9 @@ pub async fn restart_task(
     // Execute reset
     let mut reset_ids = Vec::new();
     for tid in &to_reset {
-        if let Some(conn) = conn {
-            let task_repo = flowctl_db::TaskRepo::new(conn.clone());
-            let _ = task_repo.update_status(tid, Status::Todo).await;
-
-            let runtime_repo = flowctl_db::RuntimeRepo::new(conn.clone());
-            let rt = RuntimeState {
-                task_id: tid.clone(),
-                assignee: None,
-                claimed_at: None,
-                completed_at: None,
-                duration_secs: None,
-                blocked_reason: None,
-                baseline_rev: None,
-                final_rev: None,
-                retry_count: 0,
-            };
-            let _ = runtime_repo.upsert(&rt).await;
-        }
+        // Reset to blank JSON state
+        let blank = flowctl_core::json_store::TaskState::default();
+        let _ = flowctl_core::json_store::state_write(flow_dir, tid, &blank);
 
         reset_ids.push(tid.clone());
     }