feat: workflow optimization — safety fixes, review consistency, observability

- Plan-sync safety: skip in_progress task specs to prevent worker drift
- File lock conflict detection in Phase 3c½ before worker spawn
- Unify MAX_REVIEW_ITERATIONS=3 across all review skills (fix steps.md
  contradiction: was >=2 AND max 5, now consistently max 3)
- Add explicit circuit breaker to plan-review and epic-review SKILL.md
- Brainstorm/plan dedup: Step 0.5 skips when requirements doc exists
- Formal Wave Model definition in phases.md
- Scout decision tree simplified to 3 profiles (quick/standard/deep)
- New `flowctl log decision` command for workflow traceability
- New `flowctl status --progress` with Unicode progress bar
- New `flowctl doctor --workflow` checks (backend config, tools, locks)

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

Author: z23cc

agents/plan-sync.md

@@ -181,3 +181,4 @@ Updated tasks (cross-epic):  # Only if CROSS_EPIC enabled and found
 - **Preserve intent** - Update references, not requirements
 - **Minimal changes** - Only fix stale references, don't rewrite specs
 - **Skip if no drift** - Return quickly if implementation matches spec
+- **Never edit in_progress tasks** - Before editing any task spec, check status via `$FLOWCTL show <task-id> --json`. If status is `in_progress`, skip the edit and log: `"Skipping <task-id>: task is in_progress (worker may be executing)"`. This prevents spec drift while workers are actively implementing.

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

@@ -64,7 +64,7 @@ mod exchange;
 // ── Re-exports (preserves public API) ──────────────────────────────
 
 pub use init::{cmd_init, cmd_detect};
-pub use status::{cmd_status, cmd_doctor, cmd_validate};
+pub use status::{cmd_status, cmd_doctor, cmd_progress, cmd_validate};
 pub use review::{cmd_review_backend, cmd_parse_findings};
 pub use config::{cmd_config, cmd_state_path, ConfigCmd};
 pub use guard::{cmd_guard, cmd_worker_prompt};

flowctl/crates/flowctl-cli/src/commands/admin/status.rs

@@ -10,9 +10,10 @@ use serde_json::json;
 use crate::output::{error_exit, json_output, pretty_output};
 
 use flowctl_core::types::{
-    CONFIG_FILE, EPICS_DIR, MEMORY_DIR, META_FILE, REVIEWS_DIR, SCHEMA_VERSION,
+    CONFIG_FILE, EPICS_DIR, EpicStatus, MEMORY_DIR, META_FILE, REVIEWS_DIR, SCHEMA_VERSION,
     SPECS_DIR, TASKS_DIR,
 };
+use flowctl_core::state_machine::Status;
 
 use super::get_flow_dir;
 
@@ -491,7 +492,81 @@ pub fn cmd_validate(json_mode: bool, epic: Option<String>, all: bool) {
 
 // ── Doctor command ─────────────────────────────────────────────────
 
-pub fn cmd_doctor(json_mode: bool) {
+// ── Progress command ──────────────────────────────────────────────
+
+pub fn cmd_progress(json_mode: bool, epic_id: Option<String>) {
+    let flow_dir = get_flow_dir();
+    if !flow_dir.exists() {
+        error_exit(".flow/ does not exist. Run 'flowctl init' first.");
+    }
+
+    // Find the epic — either from flag or auto-detect first open epic
+    let epic = if let Some(id) = epic_id {
+        id
+    } else {
+        match flowctl_core::json_store::epic_list(&flow_dir) {
+            Ok(epics) => {
+                epics
+                    .iter()
+                    .find(|e| e.status == EpicStatus::Open)
+                    .map(|e| e.id.clone())
+                    .unwrap_or_else(|| {
+                        error_exit("No open epic found. Pass --epic <id>.");
+                    })
+            }
+            Err(_) => error_exit("Cannot read epics."),
+        }
+    };
+
+    // Load tasks for this epic
+    let tasks = match flowctl_core::json_store::task_list_by_epic(&flow_dir, &epic) {
+        Ok(t) => t,
+        Err(_) => {
+            error_exit(&format!("Cannot load tasks for epic {epic}"));
+        }
+    };
+
+    let total = tasks.len();
+    let done = tasks.iter().filter(|t| t.status == Status::Done || t.status == Status::Skipped).count();
+    let in_progress: Vec<&str> = tasks.iter().filter(|t| t.status == Status::InProgress).map(|t| t.id.as_str()).collect();
+    let blocked = tasks.iter().filter(|t| t.status == Status::Blocked).count();
+    let todo = tasks.iter().filter(|t| t.status == Status::Todo).count();
+    let failed = tasks.iter().filter(|t| t.status == Status::Failed || t.status == Status::UpstreamFailed).count();
+
+    // Estimate wave: count how many distinct "rounds" have completed
+    // Simple heuristic: wave = number of tasks that are done/in_progress groups
+    // Wave estimation reserved for future use
+    let percent = if total > 0 { (done * 100) / total } else { 0 };
+
+    if json_mode {
+        json_output(json!({
+            "epic": epic,
+            "tasks_total": total,
+            "tasks_done": done,
+            "tasks_in_progress": in_progress,
+            "tasks_blocked": blocked,
+            "tasks_todo": todo,
+            "tasks_failed": failed,
+            "percent": percent,
+        }));
+    } else {
+        // Progress bar using Unicode blocks
+        let bar_width = 30;
+        let filled = (percent * bar_width) / 100;
+        let empty = bar_width - filled;
+        let bar: String = "\u{2588}".repeat(filled) + &"\u{2591}".repeat(empty);
+
+        println!("Epic: {epic}");
+        println!("Tasks: {done}/{total} done, {} in progress, {} todo, {} blocked, {} failed",
+            in_progress.len(), todo, blocked, failed);
+        if !in_progress.is_empty() {
+            println!("Active: [{}]", in_progress.join(", "));
+        }
+        println!("{bar} {percent}%");
+    }
+}
+
+pub fn cmd_doctor(json_mode: bool, workflow: bool) {
     let flow_dir = get_flow_dir();
     if !flow_dir.exists() {
         error_exit(".flow/ does not exist. Run 'flowctl init' first.");
@@ -604,6 +679,85 @@ pub fn cmd_doctor(json_mode: bool) {
         }
     }
 
+    // Check 5-8: Workflow checks (only when --workflow flag is passed)
+    if workflow {
+        // Check 5: review backend configured
+        let _backend_out = Command::new("git")
+            .args(["rev-parse", "--show-toplevel"])
+            .output()
+            .ok()
+            .and_then(|o| {
+                if o.status.success() {
+                    Some(String::from_utf8_lossy(&o.stdout).trim().to_string())
+                } else {
+                    None
+                }
+            });
+        let config_path = flow_dir.join(CONFIG_FILE);
+        let review_backend = if config_path.exists() {
+            fs::read_to_string(&config_path)
+                .ok()
+                .and_then(|raw| serde_json::from_str::<serde_json::Value>(&raw).ok())
+                .and_then(|v| v["review"]["backend"].as_str().map(String::from))
+        } else {
+            None
+        };
+        match review_backend.as_deref() {
+            Some("rp") | Some("codex") | Some("none") => {
+                checks.push(json!({"name": "review_backend", "status": "pass", "message": format!("Review backend configured: {}", review_backend.as_ref().unwrap())}));
+            }
+            _ => {
+                checks.push(json!({"name": "review_backend", "status": "warn", "message": "Review backend not configured. Run /flow-code:setup or set review.backend in .flow/config.json"}));
+            }
+        }
+
+        // Check 6: configured backend tool available
+        if let Some(ref backend) = review_backend {
+            let tool = match backend.as_str() {
+                "rp" => Some("rp-cli"),
+                "codex" => Some("codex"),
+                _ => None,
+            };
+            if let Some(tool_name) = tool {
+                match Command::new("which").arg(tool_name).output() {
+                    Ok(o) if o.status.success() => {
+                        checks.push(json!({"name": "tool_available", "status": "pass", "message": format!("{} found on PATH", tool_name)}));
+                    }
+                    _ => {
+                        checks.push(json!({"name": "tool_available", "status": "fail", "message": format!("{} not found on PATH (required by review.backend={})", tool_name, backend)}));
+                    }
+                }
+            }
+        }
+
+        // Check 7: stale file locks (count via SQL)
+        let cwd = env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
+        if let Ok(conn) = crate::commands::db_shim::open(&cwd) {
+            let lock_count = crate::commands::db_shim::block_on_pub(async {
+                let mut rows = conn.inner_conn()
+                    .query("SELECT COUNT(*) FROM file_locks", ())
+                    .await
+                    .map_err(flowctl_db::DbError::LibSql)?;
+                if let Some(row) = rows.next().await.map_err(flowctl_db::DbError::LibSql)? {
+                    Ok::<i64, flowctl_db::DbError>(row.get::<i64>(0).unwrap_or(0))
+                } else {
+                    Ok(0)
+                }
+            });
+            match lock_count {
+                Ok(n) if n > 0 => {
+                    checks.push(json!({"name": "stale_locks", "status": "warn", "message": format!("{} file lock(s) active — verify with 'flowctl lock-check'", n)}));
+                }
+                Ok(_) => {
+                    checks.push(json!({"name": "stale_locks", "status": "pass", "message": "No active file locks"}));
+                }
+                Err(_) => {
+                    checks.push(json!({"name": "stale_locks", "status": "warn", "message": "Could not query file locks"}));
+                }
+            }
+        }
+    }
+
     // Build summary
     let mut summary = json!({"pass": 0, "warn": 0, "fail": 0});
     for c in &checks {

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

@@ -46,6 +46,11 @@ fn block_on<F: std::future::Future>(fut: F) -> F::Output {
         .block_on(fut)
 }
 
+/// Public block_on for command modules that call async repos directly.
+pub fn block_on_pub<F: std::future::Future>(fut: F) -> F::Output {
+    block_on(fut)
+}
+
 // ── Pool functions ──────────────────────────────────────────────────
 
 pub fn resolve_state_dir(working_dir: &Path) -> Result<PathBuf, DbError> {

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

@@ -0,0 +1,174 @@
+//! Decision logging commands: log decision, log decisions.
+//!
+//! Records workflow auto-decisions in the events table (event_type = "decision")
+//! for post-hoc traceability. Skills call `flowctl log decision` at each auto-
+//! decision point; `flowctl log decisions` queries the stored decisions.
+
+use clap::Subcommand;
+use serde_json::json;
+
+use crate::output::{error_exit, json_output, pretty_output};
+use super::db_shim;
+use super::helpers::get_flow_dir;
+
+#[derive(Subcommand, Debug)]
+pub enum LogCmd {
+    /// Record a workflow decision.
+    Decision {
+        /// Decision key (e.g., "review_backend", "branch_strategy").
+        #[arg(long)]
+        key: String,
+        /// Decision value (e.g., "rp-mcp", "worktree").
+        #[arg(long)]
+        value: String,
+        /// Why this decision was made.
+        #[arg(long)]
+        reason: String,
+        /// Epic ID (optional, for scoping).
+        #[arg(long)]
+        epic: Option<String>,
+        /// Task ID (optional, for scoping).
+        #[arg(long)]
+        task: Option<String>,
+    },
+    /// Query stored decisions.
+    Decisions {
+        /// Filter by epic ID.
+        #[arg(long)]
+        epic: Option<String>,
+        /// Maximum number of results (default 20).
+        #[arg(long, default_value = "20")]
+        limit: usize,
+    },
+}
+
+pub fn dispatch(cmd: &LogCmd, json_mode: bool) {
+    match cmd {
+        LogCmd::Decision {
+            key,
+            value,
+            reason,
+            epic,
+            task,
+        } => cmd_log_decision(json_mode, key, value, reason, epic.as_deref(), task.as_deref()),
+        LogCmd::Decisions { epic, limit } => cmd_log_decisions(json_mode, epic.as_deref(), *limit),
+    }
+}
+
+fn cmd_log_decision(
+    json_mode: bool,
+    key: &str,
+    value: &str,
+    reason: &str,
+    epic_id: Option<&str>,
+    task_id: Option<&str>,
+) {
+    let flow_dir = get_flow_dir();
+    let conn = db_shim::open(&flow_dir).unwrap_or_else(|e| {
+        error_exit(&format!("Cannot open DB: {e}"));
+    });
+
+    let payload = json!({
+        "key": key,
+        "value": value,
+        "reason": reason,
+    })
+    .to_string();
+
+    let epic = epic_id.unwrap_or("_global");
+
+    let repo = flowctl_db::repo::EventRepo::new(conn.inner_conn());
+    let id = db_shim::block_on_pub(async {
+        repo.insert(epic, task_id, "decision", None, Some(&payload), None)
+            .await
+    })
+    .unwrap_or_else(|e| {
+        error_exit(&format!("Failed to log decision: {e}"));
+    });
+
+    if json_mode {
+        json_output(json!({
+            "id": id,
+            "event_type": "decision",
+            "key": key,
+            "value": value,
+            "reason": reason,
+            "epic_id": epic,
+            "task_id": task_id,
+        }));
+    } else {
+        pretty_output("log", &format!("Decision logged: {key}={value} (reason: {reason})"));
+    }
+}
+
+fn cmd_log_decisions(json_mode: bool, epic_id: Option<&str>, limit: usize) {
+    let flow_dir = get_flow_dir();
+    let conn = db_shim::open(&flow_dir).unwrap_or_else(|e| {
+        error_exit(&format!("Cannot open DB: {e}"));
+    });
+
+    let repo = flowctl_db::repo::EventRepo::new(conn.inner_conn());
+    let events = if let Some(epic) = epic_id {
+        db_shim::block_on_pub(async { repo.list_by_epic(epic, limit * 2).await })
+            .unwrap_or_else(|e| {
+                error_exit(&format!("Failed to query decisions: {e}"));
+            })
+            .into_iter()
+            .filter(|e| e.event_type == "decision")
+            .take(limit)
+            .collect::<Vec<_>>()
+    } else {
+        db_shim::block_on_pub(async { repo.list_by_type("decision", limit).await })
+            .unwrap_or_else(|e| {
+                error_exit(&format!("Failed to query decisions: {e}"));
+            })
+    };
+
+    if json_mode {
+        let items: Vec<_> = events
+            .iter()
+            .map(|e| {
+                let payload: serde_json::Value = e
+                    .payload
+                    .as_deref()
+                    .and_then(|p| serde_json::from_str(p).ok())
+                    .unwrap_or(json!(null));
+                json!({
+                    "id": e.id,
+                    "timestamp": e.timestamp,
+                    "epic_id": e.epic_id,
+                    "task_id": e.task_id,
+                    "key": payload["key"],
+                    "value": payload["value"],
+                    "reason": payload["reason"],
+                })
+            })
+            .collect();
+        json_output(json!({ "decisions": items, "count": items.len() }));
+    } else {
+        if events.is_empty() {
+            pretty_output("log", "No decisions recorded.");
+            return;
+        }
+        pretty_output("log", &format!("Decisions ({}):", events.len()));
+        for e in &events {
+            let payload: serde_json::Value = e
+                .payload
+                .as_deref()
+                .and_then(|p| serde_json::from_str(p).ok())
+                .unwrap_or(json!(null));
+            pretty_output(
+                "log",
+                &format!(
+                    "  [{}] {}={} — {} (epic: {}, task: {})",
+                    &e.timestamp[..19],
+                    payload["key"].as_str().unwrap_or("?"),
+                    payload["value"].as_str().unwrap_or("?"),
+                    payload["reason"].as_str().unwrap_or("?"),
+                    e.epic_id,
+                    e.task_id.as_deref().unwrap_or("-"),
+                ),
+            );
+        }
+    }
+}

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

@@ -10,6 +10,7 @@ pub mod dep;
 pub mod epic;
 pub mod gap;
 pub mod hook;
+pub mod log;
 pub mod memory;
 pub mod outputs;
 pub mod query;