feat(flowctl): add declarative Changes pattern, bidirectional dep index, and --dry-run support

Introduce three foundational improvements inspired by IWE reference:

1. Changes pattern: All mutation operations (task/epic create, reset, skip,
   split, dep add/rm) now return a declarative Changes struct instead of
   writing directly. ChangesApplier persists to JSON + libSQL with auto
   audit logging to events table.

2. Bidirectional dependency index: task_reverse_deps table with SQL triggers
   on task_deps INSERT/DELETE. O(1) reverse lookups replace O(N×M) BFS scan.
   DepRepo gains list_dependents() and list_all_dependents() methods.

3. --dry-run CLI flag: Global flag on all mutation commands. Computes changes
   and outputs JSON preview without persisting. 2 integration tests verify
   no side effects.

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

Author: z23cc

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

@@ -334,6 +334,21 @@ impl FileLockRepo {
     }
 }
 
+// ── Event repository ──────────────────────────────────────────────
+
+pub struct EventRepoSync(libsql::Connection);
+
+impl EventRepoSync {
+    pub fn new(conn: &Connection) -> Self {
+        Self(conn.inner())
+    }
+
+    /// Return the inner async EventRepo for use with ChangesApplier.
+    pub fn as_async(&self) -> flowctl_db::EventRepo {
+        flowctl_db::EventRepo::new(self.0.clone())
+    }
+}
+
 // ── Phase progress repository ──────────────────────────────────────
 
 pub struct PhaseProgressRepo(libsql::Connection);

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

@@ -1,11 +1,14 @@
 //! Dependency management commands: add, remove.
 
+use std::path::Path;
+
 use chrono::Utc;
 use clap::Subcommand;
 use serde_json::json;
 
 use crate::output::{error_exit, json_output};
 
+use flowctl_core::changes::{Changes, Mutation};
 use flowctl_core::id::{epic_id_from_task, is_task_id};
 
 use super::helpers::get_flow_dir;
@@ -36,16 +39,20 @@ fn ensure_flow_exists() -> std::path::PathBuf {
     flow_dir
 }
 
-pub fn dispatch(cmd: &DepCmd, json: bool) {
+pub fn dispatch(cmd: &DepCmd, json: bool, dry_run: bool) {
     match cmd {
-        DepCmd::Add { task, depends_on } => cmd_dep_add(json, task, depends_on),
-        DepCmd::Rm { task, depends_on } => cmd_dep_rm(json, task, depends_on),
+        DepCmd::Add { task, depends_on } => cmd_dep_add(json, task, depends_on, dry_run),
+        DepCmd::Rm { task, depends_on } => cmd_dep_rm(json, task, depends_on, dry_run),
     }
 }
 
-fn cmd_dep_add(json: bool, task_id: &str, depends_on: &str) {
-    let flow_dir = ensure_flow_exists();
-
+/// Pure compute: build Changes to add a dependency.
+/// Returns (updated task, whether dep was actually added).
+fn compute_dep_add(
+    flow_dir: &Path,
+    task_id: &str,
+    depends_on: &str,
+) -> (flowctl_core::types::Task, bool, Changes) {
     if !is_task_id(task_id) {
         error_exit(&format!(
             "Invalid task ID: {}. Expected format: fn-N.M or fn-N-slug.M",
@@ -59,7 +66,6 @@ fn cmd_dep_add(json: bool, task_id: &str, depends_on: &str) {
         ));
     }
 
-    // Validate same epic
     let task_epic = epic_id_from_task(task_id)
         .unwrap_or_else(|_| error_exit(&format!("Cannot parse epic from task ID: {}", task_id)));
     let dep_epic = epic_id_from_task(depends_on)
@@ -71,15 +77,34 @@ fn cmd_dep_add(json: bool, task_id: &str, depends_on: &str) {
         ));
     }
 
-    let mut task = flowctl_core::json_store::task_read(&flow_dir, task_id)
+    let mut task = flowctl_core::json_store::task_read(flow_dir, task_id)
         .unwrap_or_else(|_| error_exit(&format!("Task not found: {}", task_id)));
 
-    if !task.depends_on.contains(&depends_on.to_string()) {
+    let added = if !task.depends_on.contains(&depends_on.to_string()) {
         task.depends_on.push(depends_on.to_string());
         task.updated_at = Utc::now();
-        if let Err(e) = flowctl_core::json_store::task_write_definition(&flow_dir, &task) {
-            error_exit(&format!("Failed to write task: {e}"));
-        }
+        true
+    } else {
+        false
+    };
+
+    let changes = if added {
+        Changes::new().with(Mutation::UpdateTask { task: task.clone() })
+    } else {
+        Changes::new()
+    };
+
+    (task, added, changes)
+}
+
+fn cmd_dep_add(json: bool, task_id: &str, depends_on: &str, dry_run: bool) {
+    let flow_dir = ensure_flow_exists();
+
+    let (task, _added, changes) = compute_dep_add(&flow_dir, task_id, depends_on);
+
+    crate::commands::helpers::maybe_apply_changes(&flow_dir, &changes, dry_run);
+    if dry_run {
+        return;
     }
 
     if json {
@@ -93,26 +118,51 @@ fn cmd_dep_add(json: bool, task_id: &str, depends_on: &str) {
     }
 }
 
-fn cmd_dep_rm(json: bool, task_id: &str, depends_on: &str) {
-    let flow_dir = ensure_flow_exists();
-
+/// Pure compute: build Changes to remove a dependency.
+/// Returns (updated task, whether dep was actually removed).
+fn compute_dep_rm(
+    flow_dir: &Path,
+    task_id: &str,
+    depends_on: &str,
+) -> (flowctl_core::types::Task, bool, Changes) {
     if !is_task_id(task_id) {
         error_exit(&format!("Invalid task ID: {}", task_id));
     }
     if !is_task_id(depends_on) {
         error_exit(&format!("Invalid dependency ID: {}", depends_on));
     }
 
-    let mut task = flowctl_core::json_store::task_read(&flow_dir, task_id)
+    let mut task = flowctl_core::json_store::task_read(flow_dir, task_id)
         .unwrap_or_else(|_| error_exit(&format!("Task not found: {}", task_id)));
 
-    if let Some(pos) = task.depends_on.iter().position(|d| d == depends_on) {
+    let removed = if let Some(pos) = task.depends_on.iter().position(|d| d == depends_on) {
         task.depends_on.remove(pos);
         task.updated_at = Utc::now();
-        if let Err(e) = flowctl_core::json_store::task_write_definition(&flow_dir, &task) {
-            error_exit(&format!("Failed to write task: {e}"));
-        }
+        true
+    } else {
+        false
+    };
+
+    let changes = if removed {
+        Changes::new().with(Mutation::UpdateTask { task: task.clone() })
+    } else {
+        Changes::new()
+    };
+
+    (task, removed, changes)
+}
+
+fn cmd_dep_rm(json: bool, task_id: &str, depends_on: &str, dry_run: bool) {
+    let flow_dir = ensure_flow_exists();
+
+    let (task, removed, changes) = compute_dep_rm(&flow_dir, task_id, depends_on);
+
+    crate::commands::helpers::maybe_apply_changes(&flow_dir, &changes, dry_run);
+    if dry_run {
+        return;
+    }
 
+    if removed {
         if json {
             json_output(json!({
                 "task": task_id,

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

@@ -1,33 +1,38 @@
 //! Epic CRUD commands: create, plan, review, completion, branch, title, backend, auto_exec.
 
 use std::fs;
+use std::path::Path;
 
 use chrono::Utc;
 use serde_json::json;
 
 use crate::output::{error_exit, json_output};
 
+use flowctl_core::changes::{Changes, Mutation};
 use flowctl_core::id::{generate_epic_suffix, parse_id, slugify};
 use flowctl_core::types::{
-    Epic, EpicStatus, ReviewStatus, Document, FLOW_DIR, SPECS_DIR,
+    Epic, EpicStatus, ReviewStatus, FLOW_DIR, SPECS_DIR,
 };
 
 use super::helpers::{
     create_epic_spec_body, ensure_flow_exists, ensure_meta_exists, find_max_epic_number,
     load_epic, read_file_or_stdin, save_epic, validate_epic_id,
 };
-pub fn cmd_create(title: &str, branch: &Option<String>, json_mode: bool) {
-    let flow_dir = ensure_flow_exists();
-    ensure_meta_exists(&flow_dir);
 
-    // DB-based ID allocation
+/// Pure compute: build Changes for epic creation.
+/// Returns (epic_id, spec_body, changes).
+fn compute_epic_create(
+    flow_dir: &Path,
+    title: &str,
+    branch: &Option<String>,
+) -> (String, String, Changes) {
     let max_epic = find_max_epic_number();
     let epic_num = max_epic + 1;
     let slug = slugify(title, 40);
     let suffix = slug.unwrap_or_else(|| generate_epic_suffix(3));
     let epic_id = format!("fn-{epic_num}-{suffix}");
 
-    // Collision check: only check spec file (no more epic MD)
+    // Collision check
     let spec_path = flow_dir.join(SPECS_DIR).join(format!("{epic_id}.md"));
     if spec_path.exists() {
         error_exit(&format!(
@@ -59,15 +64,31 @@ pub fn cmd_create(title: &str, branch: &Option<String>, json_mode: bool) {
         updated_at: now,
     };
 
-    // Write to DB (sole source of truth)
     let body = create_epic_spec_body(&epic_id, title);
-    let doc = Document {
-        frontmatter: epic,
-        body: body.clone(),
-    };
-    save_epic(&doc);
 
-    // Write spec file (body-only Markdown in specs/)
+    let changes = Changes::new()
+        .with(Mutation::CreateEpic { epic })
+        .with(Mutation::SetEpicSpec {
+            epic_id: epic_id.clone(),
+            content: body.clone(),
+        });
+
+    (epic_id, body, changes)
+}
+
+pub fn cmd_create(title: &str, branch: &Option<String>, json_mode: bool, dry_run: bool) {
+    let flow_dir = ensure_flow_exists();
+    ensure_meta_exists(&flow_dir);
+
+    let (epic_id, body, changes) = compute_epic_create(&flow_dir, title, branch);
+
+    crate::commands::helpers::maybe_apply_changes(&flow_dir, &changes, dry_run);
+    if dry_run {
+        return;
+    }
+
+    // Also write the spec Markdown file (body-only in specs/)
+    let spec_path = flow_dir.join(SPECS_DIR).join(format!("{epic_id}.md"));
     if let Some(parent) = spec_path.parent() {
         let _ = fs::create_dir_all(parent);
     }

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

@@ -15,9 +15,9 @@ pub use history::{cmd_diff, cmd_replay};
 pub use types::EpicCmd;
 
 /// Dispatch an epic subcommand.
-pub fn dispatch(cmd: &EpicCmd, json: bool) {
+pub fn dispatch(cmd: &EpicCmd, json: bool, dry_run: bool) {
     match cmd {
-        EpicCmd::Create { title, branch } => crud::cmd_create(title, branch, json),
+        EpicCmd::Create { title, branch } => crud::cmd_create(title, branch, json, dry_run),
         EpicCmd::Plan { id, file } => crud::cmd_set_plan(id, file, json),
         EpicCmd::Review { id, status } => crud::cmd_set_plan_review_status(id, status, json),
         EpicCmd::Completion { id, status } => {

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

@@ -143,6 +143,68 @@ fn copy_dir_recursive(src: &Path, dst: &Path) -> Result<(), std::io::Error> {
     Ok(())
 }
 
+/// Apply a `Changes` batch via the service-layer `ChangesApplier`.
+///
+/// This opens the libSQL event DB, spins a short-lived tokio runtime, and
+/// applies all mutations (JSON store writes + event logging) in order.
+/// Returns the number of mutations applied. Calls `error_exit` on failure.
+pub fn apply_changes(flow_dir: &Path, changes: &flowctl_core::changes::Changes) -> usize {
+    use crate::output::error_exit;
+    use flowctl_service::changes::ChangesApplier;
+
+    if changes.is_empty() {
+        return 0;
+    }
+
+    let actor = resolve_actor();
+
+    // Open DB for event logging (best-effort: if DB unavailable, fall back to JSON-only)
+    let cwd = env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap_or_else(|e| error_exit(&format!("tokio runtime: {e}")));
+
+    let applied = rt.block_on(async {
+        let db = flowctl_db::open_async(&cwd).await
+            .unwrap_or_else(|e| error_exit(&format!("Failed to open DB: {e}")));
+        let conn = db.connect()
+            .unwrap_or_else(|e| error_exit(&format!("Failed to connect to DB: {e}")));
+        let event_repo = flowctl_db::EventRepo::new(conn);
+
+        let applier = ChangesApplier::new(flow_dir, &event_repo)
+            .with_actor(&actor);
+
+        let result = applier.apply(changes).await
+            .unwrap_or_else(|e| error_exit(&format!("Failed to apply changes: {e}")));
+        // Leak the DB handle to keep it alive (same pattern as db_shim)
+        std::mem::forget(db);
+        result.applied
+    });
+
+    applied
+}
+
+/// Handle dry-run or real apply of a `Changes` batch.
+///
+/// When `dry_run` is true, prints the changes as a JSON preview and returns 0
+/// without touching storage. Otherwise delegates to `apply_changes`.
+pub fn maybe_apply_changes(
+    flow_dir: &Path,
+    changes: &flowctl_core::changes::Changes,
+    dry_run: bool,
+) -> usize {
+    if dry_run {
+        let preview = serde_json::json!({
+            "dry_run": true,
+            "changes": changes,
+        });
+        println!("{}", serde_json::to_string(&preview).unwrap());
+        return 0;
+    }
+    apply_changes(flow_dir, changes)
+}
+
 /// Resolve current actor: FLOW_ACTOR env > git config user.email > git config user.name > $USER > "unknown"
 pub fn resolve_actor() -> String {
     if let Ok(actor) = env::var("FLOW_ACTOR") {