feat: add dispatch preflight/init validation step (duplicate-work check + workflow-store smoke test)

crates/flare-workflow/src/lib.rs

@@ -37,6 +37,6 @@ pub use json::{
 };
 pub use retry::{Backoff, Retryable, apply_jitter};
 pub use sqlite_store::{SqliteStore, SqliteStoreError};
-pub use store::{InMemoryStore, StateStore};
+pub use store::{InMemoryStore, StateStore, smoke_test};
 pub use types::*;
 pub use variables::{capture_output, expand_variables};

crates/flare-workflow/src/sqlite_store.rs

@@ -697,7 +697,7 @@ mod tests {
     use super::*;
     use crate::types::{JournalEntry, StepId, StepState, WorkflowId, WorkflowRunId};
 
-    #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+    #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
     struct TestData {
         value: i32,
     }
@@ -865,4 +865,81 @@ mod tests {
         let err = store.load(original.run_id).await.unwrap_err();
         assert!(matches!(err, WorkflowError::NotFound(_)));
     }
+
+    /// Boot-time gate (item #164): `crate::store::smoke_test` must pass
+    /// against the real on-disk `SqliteStore` schema, not just an in-memory
+    /// double, since it's what actually caught nothing for the ~33h
+    /// `delete_state`'s wrong column name (item #576) went unnoticed.
+    #[tokio::test]
+    async fn store_smoke_test_passes_against_the_real_sqlite_schema() {
+        let store = SqliteStore::<TestData>::open_memory().unwrap();
+        crate::store::smoke_test(&store).await.unwrap();
+    }
+
+    /// Item #164's other acceptance criterion, driven through the real
+    /// rusqlite code path rather than a trait double: `write_state` only
+    /// ever inserts into `workflow_runs` (keyed on `id`) for a smoke test's
+    /// empty-steps/empty-vars `WorkflowState`, so `save` succeeds here and
+    /// the schema defect — `journal` missing the `run_id` column
+    /// `delete_state` unconditionally deletes by, the exact shape of #576's
+    /// regression — is only reached by `delete_state`'s real SQL, not
+    /// silently swallowed as a save failure.
+    #[tokio::test]
+    async fn smoke_test_fails_against_a_schema_missing_the_id_run_id_distinction() {
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            "CREATE TABLE workflow_runs (
+                id TEXT PRIMARY KEY,
+                workflow_id TEXT NOT NULL,
+                status TEXT NOT NULL,
+                current_step TEXT,
+                input TEXT,
+                output TEXT,
+                error TEXT,
+                state_json TEXT NOT NULL,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+             );
+             CREATE TABLE journal (
+                bad_run_id TEXT NOT NULL,
+                seq INTEGER NOT NULL,
+                entry_type TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                PRIMARY KEY (bad_run_id, seq)
+             );
+             CREATE TABLE step_state (
+                run_id TEXT NOT NULL,
+                step_id TEXT NOT NULL,
+                status TEXT NOT NULL,
+                attempt INTEGER,
+                last_error TEXT,
+                started_at TEXT,
+                completed_at TEXT,
+                duration_ms INTEGER,
+                input_tokens INTEGER,
+                output_tokens INTEGER,
+                PRIMARY KEY (run_id, step_id)
+             );
+             CREATE TABLE run_vars (
+                run_id TEXT NOT NULL,
+                key TEXT NOT NULL,
+                value TEXT,
+                PRIMARY KEY (run_id, key)
+             );",
+        )
+        .unwrap();
+
+        let store = SqliteStore::<TestData> {
+            conn: Arc::new(Mutex::new(conn)),
+            update_lock: Arc::new(tokio::sync::Mutex::new(())),
+            _marker: PhantomData,
+        };
+
+        let err = crate::store::smoke_test(&store).await.unwrap_err();
+        assert!(matches!(err, WorkflowError::Store(_)), "got {err:?}");
+        assert!(
+            err.to_string().contains("journal"),
+            "expected the failure to come from delete_state's journal delete, got: {err}"
+        );
+    }
 }

crates/flare-workflow/src/store.rs

@@ -66,6 +66,36 @@ pub trait StateStore<D: WorkflowData>: Send + Sync + Clone {
     async fn workflow_metrics(&self, filter: MetricsFilter) -> WorkflowResult<WorkflowMetrics>;
 }
 
+/// Cheap end-to-end proof that `store`'s actual schema works: write a
+/// throwaway state, delete it, and confirm a subsequent load reports it
+/// gone. Exists because `SqliteStore::delete_state` silently used the wrong
+/// column name on three of its four tables for ~33h before anyone noticed
+/// (agentflare item #164/#576) — every claimed item just retried and failed
+/// individually instead of surfacing anything actionable. Callers (daemon
+/// boot) should refuse to dispatch entirely on failure rather than let that
+/// repeat.
+pub async fn smoke_test<D, S>(store: &S) -> WorkflowResult<()>
+where
+    D: WorkflowData + Default,
+    S: StateStore<D>,
+{
+    let run_id = WorkflowRunId::new();
+    let state = crate::types::WorkflowState::new(
+        run_id,
+        crate::types::WorkflowId::new("agentflare-store-smoke-test"),
+        D::default(),
+    );
+    store.save(state).await?;
+    store.delete(run_id).await?;
+    match store.load(run_id).await {
+        Err(WorkflowError::NotFound(_)) => Ok(()),
+        Ok(_) => Err(WorkflowError::Store(
+            "smoke test: state still loadable after delete()".into(),
+        )),
+        Err(e) => Err(e),
+    }
+}
+
 /// In-memory state storage for workflow instances.
 #[derive(Clone)]
 pub struct InMemoryStore<D: WorkflowData> {
@@ -305,3 +335,89 @@ impl<D: WorkflowData> StateStore<D> for InMemoryStore<D> {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+    struct TestData;
+
+    impl WorkflowData for TestData {
+        fn workflow_type() -> &'static str {
+            "smoke-test-data"
+        }
+    }
+
+    #[tokio::test]
+    async fn smoke_test_passes_against_a_working_store() {
+        let store = InMemoryStore::<TestData>::new();
+        smoke_test(&store).await.unwrap();
+    }
+
+    /// A store double whose `delete` is a no-op -- the exact shape of the
+    /// bug `SqliteStore::delete_state` had (item #576): `save`/`load` work
+    /// fine, but `delete` silently fails to remove the row, so a
+    /// caller-observable `load()` after `delete()` still succeeds.
+    #[derive(Clone, Default)]
+    struct BrokenDeleteStore<D: WorkflowData> {
+        inner: InMemoryStore<D>,
+    }
+
+    #[async_trait]
+    impl<D: WorkflowData> StateStore<D> for BrokenDeleteStore<D> {
+        async fn save(&self, state: WorkflowState<D>) -> WorkflowResult<()> {
+            self.inner.save(state).await
+        }
+        async fn load(&self, run_id: WorkflowRunId) -> WorkflowResult<WorkflowState<D>> {
+            self.inner.load(run_id).await
+        }
+        async fn update<F>(&self, run_id: WorkflowRunId, f: F) -> WorkflowResult<()>
+        where
+            F: FnOnce(&mut WorkflowState<D>) + Send,
+        {
+            self.inner.update(run_id, f).await
+        }
+        async fn delete(&self, _run_id: WorkflowRunId) -> WorkflowResult<()> {
+            Ok(())
+        }
+        async fn list_active(&self) -> WorkflowResult<Vec<WorkflowState<D>>> {
+            self.inner.list_active().await
+        }
+        async fn list_all(&self) -> WorkflowResult<Vec<WorkflowState<D>>> {
+            self.inner.list_all().await
+        }
+        async fn is_cancelled(&self, run_id: WorkflowRunId) -> WorkflowResult<bool> {
+            self.inner.is_cancelled(run_id).await
+        }
+        async fn cleanup_old_workflows(&self, ttl: Duration) -> usize {
+            self.inner.cleanup_old_workflows(ttl).await
+        }
+        async fn get_context(&self, run_id: WorkflowRunId) -> WorkflowResult<WorkflowContext<D>> {
+            self.inner.get_context(run_id).await
+        }
+        async fn cleanup_if_terminal(&self, run_id: WorkflowRunId) -> bool {
+            self.inner.cleanup_if_terminal(run_id).await
+        }
+        async fn append_journal(
+            &self,
+            run_id: WorkflowRunId,
+            entry: JournalEntry,
+        ) -> WorkflowResult<u64> {
+            self.inner.append_journal(run_id, entry).await
+        }
+        async fn journal(&self, run_id: WorkflowRunId) -> WorkflowResult<Vec<JournalEntry>> {
+            self.inner.journal(run_id).await
+        }
+        async fn workflow_metrics(&self, filter: MetricsFilter) -> WorkflowResult<WorkflowMetrics> {
+            self.inner.workflow_metrics(filter).await
+        }
+    }
+
+    #[tokio::test]
+    async fn smoke_test_fails_when_delete_does_not_remove_state() {
+        let store = BrokenDeleteStore::<TestData>::default();
+        let err = smoke_test(&store).await.unwrap_err();
+        assert!(matches!(err, WorkflowError::Store(_)));
+    }
+}

src/cli/daemon.rs

@@ -27,6 +27,15 @@ pub enum DaemonSubcommand {
         #[arg(short, long)]
         follow: bool,
     },
+    /// Internal: run the workflow-store boot smoke test standalone (save,
+    /// delete, confirm gone) and exit 0/1. `dashboard::server::run` runs the
+    /// same check in-process on every daemon start; this hidden verb lets
+    /// `dev_install::run` (item #164) run it against a freshly built binary
+    /// right after a binary swap -- the exact trigger (item #576) that let a
+    /// broken `delete_state` go unnoticed for ~33h. Not meant to be run by
+    /// hand.
+    #[command(hide = true, name = "workflow-store-smoke-test")]
+    WorkflowStoreSmokeTest,
 }
 
 impl DaemonArgs {
@@ -39,6 +48,7 @@ impl DaemonArgs {
             DaemonSubcommand::Enable => cmd_enable(),
             DaemonSubcommand::Disable => cmd_disable(),
             DaemonSubcommand::Logs { follow } => cmd_logs(follow),
+            DaemonSubcommand::WorkflowStoreSmokeTest => cmd_workflow_store_smoke_test(),
         }
     }
 }
@@ -135,3 +145,17 @@ fn cmd_disable() {
         }
     }
 }
+
+fn cmd_workflow_store_smoke_test() {
+    let store = crate::work_item_pipeline::engine().state_store();
+    let result = tokio::runtime::Runtime::new()
+        .unwrap()
+        .block_on(flare_workflow::smoke_test(store));
+    match result {
+        Ok(()) => crate::ui::success("workflow store smoke test passed"),
+        Err(e) => {
+            crate::ui::error(&format!("workflow store smoke test failed: {e}"));
+            std::process::exit(1);
+        }
+    }
+}

src/cli/work.rs

@@ -542,6 +542,7 @@ pub(crate) fn notify(recipient: &str, body: &str, item_id: &str) {
     }
 }
 
+include!("work_duplicate_pr.rs");
 impl WorkArgs {
     pub fn run(self) {
         if let Some(agent) = agent_detector::agent_name() {
@@ -567,20 +568,15 @@ impl WorkArgs {
     }
 }
 
-/// `execute_work`'s result: the process exit code (0 = success), plus — set
-/// only when the failure was classified as rate-limit shaped — a hint for
-/// how long the job queue should wait before retrying this item, and
-/// `fatal` for a structural setup failure (see its doc comment below).
-/// `WorkItemExecutor` converts this into `agentflare_jobs::JobFailure`.
+/// `execute_work`'s result: exit code, an optional rate-limit retry-after
+/// hint, and `fatal` for a structural setup failure (see below). Converted
+/// into `agentflare_jobs::JobFailure` by `WorkItemExecutor`.
 pub(crate) struct WorkOutcome {
     pub exit_code: i32,
     pub retry_after_secs: Option<u64>,
-    /// Set for a failure that happened while establishing the working
-    /// environment (e.g. "claim succeeded but no worktree was created") as
-    /// opposed to a failure during the agent run itself. The underlying
-    /// cause of a structural setup failure doesn't change between attempts
-    /// — see `agentflare_jobs::JobFailure::fatal`, which this maps onto so
-    /// the job queue fails it straight to terminal instead of retrying.
+    /// Set for a failure establishing the working environment (vs. during
+    /// the agent run) — maps onto `JobFailure::fatal` to fail straight to
+    /// terminal instead of retrying, since the cause won't change on retry.
     pub fatal: bool,
 }
 
@@ -618,13 +614,10 @@ const DEFAULT_COOLDOWN_PROFILE: &str = "__default__";
 /// interactive path's rate-limit rotation.
 const RATE_LIMIT_COOLDOWN_MINUTES: u32 = 30;
 
-/// Classifies a headless run's failure message the same way the interactive
-/// `agentflare run` path does (`auth_runner::is_rate_limited`) and, if it
-/// looks rate-limit shaped, records a cooldown so `auth_db::is_cooling_down`
-/// (checked by the discovery tick before dispatching the next item for this
-/// agent, and by `auth rotate`) sees it too. Returns the seconds until that
-/// cooldown clears, for the caller to pass through as the job queue's
-/// retry-after delay.
+/// Classifies a headless failure the same way `auth_runner::is_rate_limited`
+/// does for the interactive path and, if rate-limit shaped, records a
+/// cooldown so `auth_db::is_cooling_down` (discovery tick, `auth rotate`)
+/// sees it too. Returns seconds until clear, as the job queue's retry delay.
 fn classify_and_cooldown(agent: &str, failure_message: &str) -> Option<u64> {
     if !crate::auth_runner::is_rate_limited(failure_message) {
         return None;
@@ -645,27 +638,22 @@ fn classify_and_cooldown(agent: &str, failure_message: &str) -> Option<u64> {
 
 /// Claims `args.target`, runs the resolved agent on it, and reports the
 /// outcome back onto the item — the whole body of `agentflare work`.
-/// Progress lines that used to go straight to stdout now go through `log`
-/// instead, so this same logic can run in-process inside the daemon
-/// (`WorkItemExecutor`, called from `agentflare_jobs::WorkerPool`) with its
-/// progress captured into that job's own log file — the exact same file the
-/// dashboard already tails for subprocess-dispatched jobs — rather than only
-/// working when there's a real subprocess's stdout to capture.
+/// Progress goes through `log` instead of straight to stdout, so this same
+/// logic can run in-process inside the daemon (`WorkItemExecutor`, called
+/// from `agentflare_jobs::WorkerPool`), captured into that job's own log
+/// file — the same file the dashboard already tails for subprocess jobs.
 ///
-/// `args.repo_root`, when set (daemon dispatch — see `WorkItemExecutor`),
-/// scopes project/worktree resolution to the claimed item's own project
-/// directory instead of this process's cwd (item #63) — a human running
-/// `agentflare work` directly leaves it unset and keeps the prior
-/// cwd-resolved behavior.
+/// `args.repo_root`, when set (daemon dispatch), scopes project/worktree
+/// resolution to the claimed item's own project directory instead of this
+/// process's cwd (item #63); a human running `agentflare work` directly
+/// leaves it unset and keeps the prior cwd-resolved behavior.
 pub(crate) fn execute_work(args: WorkArgs, log: &mut dyn std::io::Write) -> WorkOutcome {
     execute_work_impl(args, log, crate::work_item_pipeline::run_or_resume)
 }
 
 /// Test seam: same as [`execute_work`], with the `sdd_loop`→`finalize`
 /// pipeline runner injected — mirrors `work_item_pipeline`'s own
-/// `_with_sender` pattern, one level up (this file doesn't touch
-/// `flare_workflow::json::SendMessage` directly, only whatever runs the
-/// whole pipeline for a given item).
+/// `_with_sender` pattern, one level up.
 #[allow(clippy::too_many_arguments)]
 fn execute_work_impl(
     args: WorkArgs,
@@ -753,10 +741,9 @@ fn execute_work_impl(
     let status = claim["status"].as_str().unwrap_or("unknown");
     if status != "acquired" {
         // "held" (a live claim by another owner) and "blocked" (an unaccepted
-        // handoff — see `ClaimOutcome::BlockedByAssignee`) are different
-        // shapes: "blocked" has no `owner`/`age_secs` at all, so formatting
-        // it as if it were "held" printed the nonsensical "held by ? (0s)"
-        // instead of the actionable reason the claim response already carries.
+        // handoff, `ClaimOutcome::BlockedByAssignee`) differ: "blocked" has
+        // no `owner`/`age_secs`, so formatting it as "held" printed the
+        // nonsensical "held by ? (0s)" instead of the real reason.
         let msg = match status {
             "blocked" => claim["reason"]
                 .as_str()
@@ -828,6 +815,17 @@ fn execute_work_impl(
         }
     };
 
+    if let Some(outcome) = duplicate_pr_short_circuit(
+        &mcp,
+        &item_detail,
+        wpath,
+        args.notify.as_deref(),
+        &mut claim_guard,
+        log,
+    ) {
+        return outcome;
+    }
+
     // --- Resolve agent (explicit flag, else route from the item) ---
     // Only pay for detecting installed agents / loading the router config
     // when there's no explicit --agent to short-circuit resolve_agent with.
@@ -2097,4 +2095,6 @@ rotate = true
         // exercised cross-module from `cli::work`.
         let _ = AgentflareMcp::item_update;
     }
+
+    include!("work_duplicate_pr_tests.rs");
 }