fix(flowctl): harden state writes with atomic rename and error propagation

[fn-110-workflow-reliability-and-performance.3]

- state_write() now uses write-to-temp + fs::rename() for POSIX atomic
  writes, preventing partial/corrupt state files on crash
- All 7 `let _ = state_write(...)` sites in lifecycle.rs now either
  propagate errors via `?` (done_task, block_task, restart_task,
  handle_task_failure) or log warnings (propagate_upstream_failure loop)
- Added test verifying atomic write leaves no temp file and fully
  replaces prior content

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

Author: z23cc

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

@@ -340,11 +340,18 @@ pub fn state_read(flow_dir: &Path, task_id: &str) -> Result<TaskState> {
 }
 
 /// Write task runtime state to `.state/tasks/<id>.state.json`.
+///
+/// Uses write-to-temp + `fs::rename()` for atomic writes on POSIX,
+/// preventing partial/corrupt state files on crash or power loss.
 pub fn state_write(flow_dir: &Path, task_id: &str, state: &TaskState) -> Result<()> {
     ensure_dir(&state_tasks_dir(flow_dir))?;
     let path = task_state_path(flow_dir, task_id);
     let content = serde_json::to_string_pretty(state)?;
-    fs::write(&path, content)?;
+
+    // Write to a temporary file in the same directory, then atomic rename
+    let tmp_path = path.with_extension("state.json.tmp");
+    fs::write(&tmp_path, &content)?;
+    fs::rename(&tmp_path, &path)?;
     Ok(())
 }
 
@@ -628,6 +635,39 @@ mod tests {
         assert!(read_back[1].resolved);
     }
 
+    #[test]
+    fn test_state_write_atomic_no_corrupt() {
+        // Verify that state_write uses atomic rename: if the original file
+        // exists, a second write should fully replace it (no partial content).
+        let tmp = TempDir::new().unwrap();
+        let flow_dir = tmp.path();
+
+        // Write initial state
+        let state1 = TaskState {
+            status: Status::InProgress,
+            assignee: Some("worker-1".to_string()),
+            ..Default::default()
+        };
+        state_write(flow_dir, "fn-1-test.1", &state1).unwrap();
+
+        // Write a second state that overwrites the first
+        let state2 = TaskState {
+            status: Status::Done,
+            assignee: Some("worker-2".to_string()),
+            ..Default::default()
+        };
+        state_write(flow_dir, "fn-1-test.1", &state2).unwrap();
+
+        // Read back and verify the file contains only the second write
+        let read_back = state_read(flow_dir, "fn-1-test.1").unwrap();
+        assert_eq!(read_back.status, Status::Done);
+        assert_eq!(read_back.assignee.as_deref(), Some("worker-2"));
+
+        // Verify no leftover .tmp file exists
+        let tmp_path = task_state_path(flow_dir, "fn-1-test.1").with_extension("state.json.tmp");
+        assert!(!tmp_path.exists(), "temporary file should be cleaned up by rename");
+    }
+
     #[test]
     fn test_not_found_errors() {
         let tmp = TempDir::new().unwrap();

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

@@ -235,13 +235,17 @@ async fn propagate_upstream_failure(
         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);
+            if let Err(e) = flowctl_core::json_store::state_write(flow_dir, tid, &state) {
+                eprintln!("warning: failed to write upstream_failed state for {tid}: {e}");
+            }
         } else {
             let state = flowctl_core::json_store::TaskState {
                 status: Status::UpstreamFailed,
                 ..Default::default()
             };
-            let _ = flowctl_core::json_store::state_write(flow_dir, tid, &state);
+            if let Err(e) = flowctl_core::json_store::state_write(flow_dir, tid, &state) {
+                eprintln!("warning: failed to write upstream_failed state for {tid}: {e}");
+            }
         }
 
         affected.push(tid.clone());
@@ -256,7 +260,7 @@ async fn handle_task_failure(
     flow_dir: &Path,
     task_id: &str,
     runtime: &Option<RuntimeState>,
-) -> (Status, Vec<String>) {
+) -> std::io::Result<(Status, Vec<String>)> {
     let max_retries = get_max_retries(flow_dir);
     let current_retry_count = runtime.as_ref().map(|r| r.retry_count).unwrap_or(0);
 
@@ -276,18 +280,20 @@ async fn handle_task_failure(
             retry_count: new_retry_count,
             updated_at: Utc::now(),
         };
-        let _ = flowctl_core::json_store::state_write(flow_dir, task_id, &task_state);
+        flowctl_core::json_store::state_write(flow_dir, task_id, &task_state)
+            .map_err(|e| std::io::Error::other(format!("failed to write retry state for {task_id}: {e}")))?;
 
-        (Status::UpForRetry, Vec::new())
+        Ok((Status::UpForRetry, Vec::new()))
     } else {
         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);
+        flowctl_core::json_store::state_write(flow_dir, task_id, &task_state)
+            .map_err(|e| std::io::Error::other(format!("failed to write failed state for {task_id}: {e}")))?;
 
         let affected = propagate_upstream_failure(conn, flow_dir, task_id).await;
-        (Status::Failed, affected)
+        Ok((Status::Failed, affected))
     }
 }
 
@@ -575,7 +581,8 @@ pub async fn done_task(
             retry_count: runtime.as_ref().map(|r| r.retry_count).unwrap_or(0),
             updated_at: now,
         };
-        let _ = flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state);
+        flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state)
+            .map_err(|e| ServiceError::IoError(std::io::Error::other(e.to_string())))?;
     }
 
     // Archive review receipt if present
@@ -649,7 +656,8 @@ pub async fn block_task(
             retry_count: existing.as_ref().map(|r| r.retry_count).unwrap_or(0),
             updated_at: Utc::now(),
         };
-        let _ = flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state);
+        flowctl_core::json_store::state_write(flow_dir, &req.task_id, &task_state)
+            .map_err(|e| ServiceError::IoError(std::io::Error::other(e.to_string())))?;
     }
 
     Ok(BlockTaskResponse {
@@ -681,7 +689,8 @@ pub async fn fail_task(
     let reason_text = req.reason.unwrap_or_else(|| "Task failed".to_string());
 
     let (final_status, upstream_failed_ids) =
-        handle_task_failure(conn, flow_dir, &req.task_id, &runtime).await;
+        handle_task_failure(conn, flow_dir, &req.task_id, &runtime).await
+            .map_err(ServiceError::IoError)?;
 
     let max_retries = get_max_retries(flow_dir);
     let retry_count = if final_status == Status::UpForRetry {
@@ -789,7 +798,8 @@ pub async fn restart_task(
     for tid in &to_reset {
         // Reset to blank JSON state
         let blank = flowctl_core::json_store::TaskState::default();
-        let _ = flowctl_core::json_store::state_write(flow_dir, tid, &blank);
+        flowctl_core::json_store::state_write(flow_dir, tid, &blank)
+            .map_err(|e| ServiceError::IoError(std::io::Error::other(e.to_string())))?;
 
         reset_ids.push(tid.clone());
     }