diff --git a/crates/flare-workflow/src/lib.rs b/crates/flare-workflow/src/lib.rs index e97c3d39a..cb50e4795 100644 --- a/crates/flare-workflow/src/lib.rs +++ b/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}; diff --git a/crates/flare-workflow/src/sqlite_store.rs b/crates/flare-workflow/src/sqlite_store.rs index 17ccee11e..6bdb7018b 100644 --- a/crates/flare-workflow/src/sqlite_store.rs +++ b/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::::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:: { + 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}" + ); + } } diff --git a/crates/flare-workflow/src/store.rs b/crates/flare-workflow/src/store.rs index dffc71326..b370bf8d7 100644 --- a/crates/flare-workflow/src/store.rs +++ b/crates/flare-workflow/src/store.rs @@ -66,6 +66,36 @@ pub trait StateStore: Send + Sync + Clone { async fn workflow_metrics(&self, filter: MetricsFilter) -> WorkflowResult; } +/// 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(store: &S) -> WorkflowResult<()> +where + D: WorkflowData + Default, + S: StateStore, +{ + 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 { @@ -305,3 +335,89 @@ impl StateStore for InMemoryStore { }) } } + +#[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::::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 { + inner: InMemoryStore, + } + + #[async_trait] + impl StateStore for BrokenDeleteStore { + async fn save(&self, state: WorkflowState) -> WorkflowResult<()> { + self.inner.save(state).await + } + async fn load(&self, run_id: WorkflowRunId) -> WorkflowResult> { + self.inner.load(run_id).await + } + async fn update(&self, run_id: WorkflowRunId, f: F) -> WorkflowResult<()> + where + F: FnOnce(&mut WorkflowState) + Send, + { + self.inner.update(run_id, f).await + } + async fn delete(&self, _run_id: WorkflowRunId) -> WorkflowResult<()> { + Ok(()) + } + async fn list_active(&self) -> WorkflowResult>> { + self.inner.list_active().await + } + async fn list_all(&self) -> WorkflowResult>> { + self.inner.list_all().await + } + async fn is_cancelled(&self, run_id: WorkflowRunId) -> WorkflowResult { + 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> { + 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 { + self.inner.append_journal(run_id, entry).await + } + async fn journal(&self, run_id: WorkflowRunId) -> WorkflowResult> { + self.inner.journal(run_id).await + } + async fn workflow_metrics(&self, filter: MetricsFilter) -> WorkflowResult { + self.inner.workflow_metrics(filter).await + } + } + + #[tokio::test] + async fn smoke_test_fails_when_delete_does_not_remove_state() { + let store = BrokenDeleteStore::::default(); + let err = smoke_test(&store).await.unwrap_err(); + assert!(matches!(err, WorkflowError::Store(_))); + } +} diff --git a/src/cli/daemon.rs b/src/cli/daemon.rs index 3a0a43029..76b12f3ff 100644 --- a/src/cli/daemon.rs +++ b/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); + } + } +} diff --git a/src/cli/work.rs b/src/cli/work.rs index 2c56bd1be..f3f883dc9 100644 --- a/src/cli/work.rs +++ b/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, - /// 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 { 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 { /// 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"); } diff --git a/src/cli/work_duplicate_pr.rs b/src/cli/work_duplicate_pr.rs new file mode 100644 index 000000000..34281eacc --- /dev/null +++ b/src/cli/work_duplicate_pr.rs @@ -0,0 +1,161 @@ +/// Duplicate-work pre-check (item #164): searches GitHub for a PR that +/// already carries `item`'s `for item #` marker (the footer +/// `worktree::pr_footer` stamps onto every PR agentflare opens). Unlike +/// `worktree::is_pr_merged`, this doesn't depend on the item's own tracked +/// branch name still matching — it's what catches a PR that merged while +/// the item's tracked state fell out of sync (items #122/#156: the state +/// promotion never ran, so a routine redispatch nearly re-did already- +/// merged work). +/// +/// Best-effort: no resolvable remote, no GitHub credentials, or a search +/// API error all just return `None` so dispatch proceeds normally — same +/// fail-open contract `worktree::is_pr_merged`/`relabel_pr_completed` +/// already use. Prefers a merged match over a still-open one (sorted first) +/// so `handle_duplicate_pr` can self-heal instead of merely flagging for +/// review, on the rare chance both exist. +fn find_duplicate_pr( + item: &agentflare_backend::item::Item, + repo_root: &std::path::Path, +) -> Option { + let repo = crate::github::RepoId::resolve_from_remote(repo_root)?; + let client = crate::github::Client::new().ok()?; + let prs = crate::github::pulls::find_by_item_marker(&client, &repo, item.sequence_id).ok()?; + pick_duplicate_pr(prs) +} + +/// Picks which match to act on when `find_by_item_marker` returns more than +/// one PR for the same item — a merged one wins so `handle_duplicate_pr` +/// self-heals instead of merely flagging for review. A closed-but-unmerged +/// PR (a previously abandoned attempt, e.g. an item found superseded and +/// closed) is dropped entirely rather than treated as an open duplicate — +/// otherwise it would permanently block a legitimate redispatch. Split out +/// from `find_duplicate_pr` so this selection logic is unit-testable +/// without a mock GitHub server. +fn pick_duplicate_pr( + mut prs: Vec, +) -> Option { + prs.retain(|pr| pr.merged_at.is_some() || pr.state == "open"); + prs.sort_by_key(|pr| pr.merged_at.is_none()); + prs.into_iter().next() +} + +/// Handles a duplicate PR `find_duplicate_pr` already found, instead of +/// letting `execute_work_impl` dispatch an agent to redo (or re-review) +/// work that's already landed or already pending. Both branches are +/// terminal successes, not failures — matching how `work_item_pipeline`'s +/// `finalize` step treats its own "hold"/"needs human review" diversions. +/// +/// - Merged: self-heal — mark the item completed, clean up its worktree, +/// relabel the PR `agentflare:completed`, and release the claim, the same +/// shape `item_check_merge`'s promoted path already uses. If marking +/// completed doesn't confirm success, this returns a retryable failure +/// instead of reporting success, leaving `claim_guard` armed so its +/// `Drop` releases the claim for a future attempt. +/// - Still open: skip dispatch and flag for a human instead of racing a +/// second PR for the same item. +/// +/// Either way, `claim_guard` is only disarmed once `item_release` actually +/// confirms success — a failed release leaves it armed so `Drop`'s +/// best-effort retry is still the backstop, matching `release_and_comment`'s +/// own contract. +#[allow(clippy::too_many_arguments)] +fn handle_duplicate_pr( + mcp: &AgentflareMcp, + item_id: &str, + item: &agentflare_backend::item::Item, + worktree_path: &std::path::Path, + pr: &crate::github::models::PullRequest, + notify_recipient: Option<&str>, + claim_guard: &mut ClaimGuard, + log: &mut dyn std::io::Write, +) -> WorkOutcome { + let body = if pr.merged_at.is_some() { + let owner = crate::claims::owner_id(); + let marked = mcp + .with_backend_db(|conn| agentflare_backend::item::mark_completed(conn, item_id, &owner)) + .ok() + .and_then(Result::ok) + .unwrap_or(false); + if !marked { + let msg = format!( + "duplicate work: PR #{} already merged for {item_id}, but mark_completed \ + didn't confirm success -- leaving claim armed for retry", + pr.number + ); + let _ = writeln!(log, "{msg}"); + crate::ui::error(&msg); + return 1.into(); + } + crate::worktree::cleanup_worktree(item, worktree_path); + crate::worktree::relabel_pr_completed(item, worktree_path); + let _ = writeln!( + log, + "duplicate work: PR #{} already merged for {item_id} -- auto-completed", + pr.number + ); + format!( + "## agentflare work — duplicate work detected\n\nPR #{} ({}) already merged for \ + this item; auto-completing instead of redispatching.", + pr.number, pr.html_url + ) + } else { + let _ = writeln!( + log, + "duplicate work: open PR #{} already covers {item_id} -- skipping dispatch", + pr.number + ); + format!( + "## agentflare work — needs human review\n\nFound an open PR #{} ({}) already \ + covering this item; skipping dispatch to avoid opening a duplicate.", + pr.number, pr.html_url + ) + }; + let _ = mcp.comment_impl(CommentRequest { + action: "create".into(), + item_id: Some(item_id.into()), + body: Some(body.clone()), + ..Default::default() + }); + match mcp.item_release(ItemRequest { + action: "release".into(), + id: Some(item_id.into()), + ..Default::default() + }) { + Ok(_) => claim_guard.disarm(), + Err(e) => { + let _ = writeln!( + log, + "duplicate work: item_release failed for {item_id}: {e} -- \ + leaving claim armed so ClaimGuard's Drop retries it" + ); + } + } + if let Some(recipient) = notify_recipient { + notify(recipient, &body, item_id); + } + 0.into() +} + +/// Thin wrapper so `execute_work_impl`'s call site is a one-line short +/// circuit instead of repeating `find_duplicate_pr`/`handle_duplicate_pr`'s +/// full argument lists inline. +fn duplicate_pr_short_circuit( + mcp: &AgentflareMcp, + item: &agentflare_backend::item::Item, + worktree_path: &std::path::Path, + notify_recipient: Option<&str>, + claim_guard: &mut ClaimGuard, + log: &mut dyn std::io::Write, +) -> Option { + let pr = find_duplicate_pr(item, worktree_path)?; + Some(handle_duplicate_pr( + mcp, + &item.id, + item, + worktree_path, + &pr, + notify_recipient, + claim_guard, + log, + )) +} diff --git a/src/cli/work_duplicate_pr_tests.rs b/src/cli/work_duplicate_pr_tests.rs new file mode 100644 index 000000000..5449749fe --- /dev/null +++ b/src/cli/work_duplicate_pr_tests.rs @@ -0,0 +1,191 @@ + fn duplicate_pr(number: u64, merged: bool) -> crate::github::models::PullRequest { + let merged_at = if merged { + r#","merged_at":"2026-08-01T00:00:00Z""# + } else { + "" + }; + serde_json::from_str(&format!( + r#"{{"number":{number},"html_url":"https://github.com/o/r/pull/{number}","state":"closed","title":"t"{merged_at}}}"# + )) + .unwrap() + } + + /// A genuinely open (unmerged, unclosed) PR, unlike [`duplicate_pr`]'s + /// unmerged case, which is closed. + fn open_duplicate_pr(number: u64) -> crate::github::models::PullRequest { + serde_json::from_str(&format!( + r#"{{"number":{number},"html_url":"https://github.com/o/r/pull/{number}","state":"open","title":"t"}}"# + )) + .unwrap() + } + + #[test] + fn pick_duplicate_pr_prefers_a_merged_match_over_an_open_one() { + let picked = + pick_duplicate_pr(vec![open_duplicate_pr(1), duplicate_pr(2, true)]).unwrap(); + assert_eq!(picked.number, 2); + } + + #[test] + fn pick_duplicate_pr_returns_none_for_an_empty_list() { + assert!(pick_duplicate_pr(vec![]).is_none()); + } + + /// A closed-but-unmerged PR (a previously abandoned attempt, e.g. one + /// found superseded and closed rather than merged) must not be treated + /// as an open duplicate — that would permanently block a legitimate + /// redispatch. `duplicate_pr(_, false)` builds exactly this shape: + /// `state: "closed"`, `merged_at: None`. + #[test] + fn pick_duplicate_pr_ignores_a_closed_unmerged_pr() { + assert!(pick_duplicate_pr(vec![duplicate_pr(3, false)]).is_none()); + } + + #[test] + fn pick_duplicate_pr_selects_a_genuinely_open_pr_when_no_merged_match_exists() { + let picked = pick_duplicate_pr(vec![duplicate_pr(4, false), open_duplicate_pr(5)]).unwrap(); + assert_eq!(picked.number, 5); + } + + /// Item #164's acceptance criterion: a merged duplicate PR self-heals + /// the item to "completed" instead of letting a redispatch re-do + /// already-merged work (the near-miss from items #122/#156). + #[test] + fn handle_duplicate_pr_auto_completes_on_a_merged_match() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + let mcp = AgentflareMcp::for_project_dir(repo_root.clone()); + let item = mcp.with_backend_db(|conn| seeded_item(&mcp, conn)).unwrap(); + let claim = mcp + .item_claim(ItemRequest { + action: "claim".into(), + id: Some(item.id.clone()), + ..Default::default() + }) + .unwrap(); + assert_eq!( + serde_json::from_str::(&claim).unwrap()["status"], + "acquired" + ); + + let mut guard = ClaimGuard::new(&mcp, &item.id); + let mut log = Vec::new(); + let outcome = handle_duplicate_pr( + &mcp, + &item.id, + &item, + &repo_root, + &duplicate_pr(42, true), + None, + &mut guard, + &mut log, + ); + assert_eq!(outcome.exit_code, 0); + + let refreshed = mcp + .with_backend_db(|conn| agentflare_backend::item::get(conn, &item.id)) + .unwrap() + .unwrap(); + let state = mcp + .with_backend_db(|conn| agentflare_backend::state::get(conn, &refreshed.state_id)) + .unwrap() + .unwrap(); + assert_eq!(state.group_name, "completed"); + + let claim_after = mcp + .item_claim(ItemRequest { + action: "claim".into(), + id: Some(item.id.clone()), + ..Default::default() + }) + .unwrap(); + assert_eq!( + serde_json::from_str::(&claim_after).unwrap()["status"], + "acquired", + "claim must be released so a re-claim succeeds" + ); + + let comments = mcp + .with_backend_db(|conn| agentflare_backend::comment::list_by_item(conn, &item.id)) + .unwrap() + .unwrap(); + assert!( + comments + .iter() + .any(|c| c.body.contains("duplicate work detected") && c.body.contains("#42")), + "expected a duplicate-work comment naming the merged PR, got: {comments:?}" + ); + }); + } + + /// The other half of item #164's MVP: an open (unmerged) duplicate skips + /// dispatch and flags for a human instead of self-completing or opening + /// a second PR. + #[test] + fn handle_duplicate_pr_flags_for_review_on_an_open_match() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + let mcp = AgentflareMcp::for_project_dir(repo_root.clone()); + let item = mcp.with_backend_db(|conn| seeded_item(&mcp, conn)).unwrap(); + let claim = mcp + .item_claim(ItemRequest { + action: "claim".into(), + id: Some(item.id.clone()), + ..Default::default() + }) + .unwrap(); + assert_eq!( + serde_json::from_str::(&claim).unwrap()["status"], + "acquired" + ); + // Capture state post-claim (claiming itself moves the item into + // the "started" group) so the assertion below isolates what + // `handle_duplicate_pr`'s open-PR branch does, not what + // claiming already did. + let state_id_after_claim = mcp + .with_backend_db(|conn| agentflare_backend::item::get(conn, &item.id)) + .unwrap() + .unwrap() + .state_id; + + let mut guard = ClaimGuard::new(&mcp, &item.id); + let mut log = Vec::new(); + let outcome = handle_duplicate_pr( + &mcp, + &item.id, + &item, + &repo_root, + &open_duplicate_pr(7), + None, + &mut guard, + &mut log, + ); + assert_eq!(outcome.exit_code, 0); + + let refreshed = mcp + .with_backend_db(|conn| agentflare_backend::item::get(conn, &item.id)) + .unwrap() + .unwrap(); + assert_eq!( + refreshed.state_id, state_id_after_claim, + "an open duplicate must not change the item's state" + ); + + let comments = mcp + .with_backend_db(|conn| agentflare_backend::comment::list_by_item(conn, &item.id)) + .unwrap() + .unwrap(); + assert!( + comments + .iter() + .any(|c| c.body.contains("needs human review") && c.body.contains("#7")), + "expected a needs-human-review comment naming the open PR, got: {comments:?}" + ); + }); + } diff --git a/src/dashboard/server.rs b/src/dashboard/server.rs index 2930561a4..576919b63 100644 --- a/src/dashboard/server.rs +++ b/src/dashboard/server.rs @@ -730,62 +730,104 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) { .expect("failed to open job queue"); // Before anything below can dequeue a single job (see its own doc comment). super::orphan_reconcile::reconcile_orphaned_jobs(&queue); - // Same reasoning as `reconcile_orphaned_jobs` above, for the work-item - // `flare-workflow` pipeline's own crash story: register a definition - // under `work_item_pipeline::WORKFLOW_ID` so `engine().recover()` (a - // whole-store sweep — see its doc comment on why it's only ever called - // here, once, never per-dispatch) has something to resume non-terminal - // runs against, then run it, before `WorkerPool::start` below can - // dispatch a fresh job for one of those same items. - // - // This definition carries no item/agent-specific identity itself — - // `build_work_item_pipeline`'s `sdd_loop`/`finalize` steps read the real - // item id, agent, and prompts from each resumed run's own persisted - // `WorkItemData` (`ctx.data`) at execution time instead of from whatever - // was passed at registration, so a genuine crash-resume reproduces the - // crashed run's actual target item and prompts, not a placeholder. - // `--timeout`/`--idle-timeout`/extra CLI args from the original dispatch - // aren't persisted, so a resumed run falls back to the defaults here — - // an accepted trade-off, not an identity gap. - { - let boot_mcp = std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()); - let boot_definition = crate::work_item_pipeline::build_work_item_pipeline( - boot_mcp, - std::time::Duration::from_secs(crate::cli::work::DEFAULT_TIMEOUT_SECS), - std::time::Duration::from_secs(crate::cli::work::DEFAULT_IDLE_TIMEOUT_SECS), - Vec::new(), - ); - if let Err(e) = crate::work_item_pipeline::engine().register_workflow(boot_definition) { - crate::ui::error(&format!( - "failed to register work-item pipeline definition at boot: {e}" - )); - } else if let Err(e) = crate::work_item_pipeline::engine().recover().await { + // Kept alive for `run()` (no graceful shutdown path, see `daemon::stop_daemon`). + // init_global() starts the sampler `supervisor` dispatch consults (item #435). + // Independent of the workflow store, so it runs regardless of the smoke + // test below. + agentflare_resource_gate::init_global(); + // Boot-time gate (item #164): a cheap save/delete/load round-trip + // against the real on-disk workflow store, proving its schema actually + // works before anything -- including workflow recovery just below -- + // touches it. Without this, a broken store (e.g. `delete()` silently + // no-op'ing on a column-name mismatch, item #576) doesn't fail loudly + // -- every claimed item just retries and fails individually into a + // stuck state, which is exactly what ran undetected for ~33h. On + // failure, skip registration/recovery, the worker pool, and the + // supervisor's discovery/review ticks entirely (no claims, no + // dispatch) while leaving the dashboard itself up so the alert below + // is actually visible. + match flare_workflow::smoke_test(crate::work_item_pipeline::engine().state_store()).await { + Ok(()) => { + // Same reasoning as `reconcile_orphaned_jobs` above, for the + // work-item `flare-workflow` pipeline's own crash story: + // register a definition under `work_item_pipeline::WORKFLOW_ID` + // so `engine().recover()` (a whole-store sweep — see its doc + // comment on why it's only ever called here, once, never + // per-dispatch) has something to resume non-terminal runs + // against, then run it, before `WorkerPool::start` below can + // dispatch a fresh job for one of those same items. + // + // This definition carries no item/agent-specific identity + // itself — `build_work_item_pipeline`'s `sdd_loop`/`finalize` + // steps read the real item id, agent, and prompts from each + // resumed run's own persisted `WorkItemData` (`ctx.data`) at + // execution time instead of from whatever was passed at + // registration, so a genuine crash-resume reproduces the + // crashed run's actual target item and prompts, not a + // placeholder. `--timeout`/`--idle-timeout`/extra CLI args + // from the original dispatch aren't persisted, so a resumed + // run falls back to the defaults here — an accepted + // trade-off, not an identity gap. + let boot_mcp = std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()); + let boot_definition = crate::work_item_pipeline::build_work_item_pipeline( + boot_mcp, + std::time::Duration::from_secs(crate::cli::work::DEFAULT_TIMEOUT_SECS), + std::time::Duration::from_secs(crate::cli::work::DEFAULT_IDLE_TIMEOUT_SECS), + Vec::new(), + ); + // A registration/recovery failure means the pipeline's own + // resume state is untrustworthy, so dispatch stays disabled + // here too -- same "fail closed, dashboard stays up" contract + // as the smoke-test gate this block is already nested inside. + let pipeline_ready = + match crate::work_item_pipeline::engine().register_workflow(boot_definition) { + Ok(()) => match crate::work_item_pipeline::engine().recover().await { + Ok(_) => true, + Err(e) => { + crate::ui::error(&format!( + "failed to recover in-flight work-item pipeline runs at boot: {e}" + )); + false + } + }, + Err(e) => { + crate::ui::error(&format!( + "failed to register work-item pipeline definition at boot: {e}" + )); + false + } + }; + + if pipeline_ready { + let mut worker_pool = agentflare_jobs::WorkerPool::new(queue.clone()) + .with_executor(std::sync::Arc::new(crate::cli::work::WorkItemExecutor)) + .with_terminal_failure_hook(std::sync::Arc::new(|_job_id, job| { + super::orphan_reconcile::handle_terminal_job_failure(job); + })); + worker_pool.start(work_max_concurrency()); + spawn_supervisor_discovery( + queue.clone(), + std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()), + SUPERVISOR_DISCOVERY_INTERVAL, + ); + spawn_supervisor_review_sweep( + queue.clone(), + std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()), + SUPERVISOR_REVIEW_SWEEP_INTERVAL, + ); + } + } + Err(e) => { crate::ui::error(&format!( - "failed to recover in-flight work-item pipeline runs at boot: {e}" + "workflow store failed its boot-time smoke test: {e} -- refusing to \ + start work dispatch (registration, recovery, worker pool, supervisor \ + ticks) until this is fixed. The dashboard will still run so you can \ + see this alert." )); } } - // Kept alive for `run()` (no graceful shutdown path, see `daemon::stop_daemon`). - // init_global() starts the sampler `supervisor` dispatch consults (item #435). - agentflare_resource_gate::init_global(); - let mut worker_pool = agentflare_jobs::WorkerPool::new(queue.clone()) - .with_executor(std::sync::Arc::new(crate::cli::work::WorkItemExecutor)) - .with_terminal_failure_hook(std::sync::Arc::new(|_job_id, job| { - super::orphan_reconcile::handle_terminal_job_failure(job); - })); - worker_pool.start(work_max_concurrency()); spawn_binary_staleness_watchdog(crate::daemon::BinarySnapshot::capture()); spawn_job_cleanup(queue.clone()); - spawn_supervisor_discovery( - queue.clone(), - std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()), - SUPERVISOR_DISCOVERY_INTERVAL, - ); - spawn_supervisor_review_sweep( - queue.clone(), - std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()), - SUPERVISOR_REVIEW_SWEEP_INTERVAL, - ); let listener = tokio::net::TcpListener::bind((host, port)) .await diff --git a/src/dev_install/mod.rs b/src/dev_install/mod.rs index 51d923ada..f4a507ffd 100644 --- a/src/dev_install/mod.rs +++ b/src/dev_install/mod.rs @@ -48,6 +48,20 @@ pub fn run(release: bool, dry_run: bool) { std::process::exit(1); } + // Same reasoning, one layer deeper (item #164): `--version` proves the + // binary starts, not that its storage layer actually works. A binary + // swap is the exact trigger that let a broken `SqliteStore::delete_state` + // (item #576) go undetected for ~33h -- this process is still running + // the *old* code, so only a subprocess of `built` itself can prove the + // *new* code's save/delete/load round-trip works before anything + // overwrites the installed binary. + if let Err(e) = verify_workflow_store(&built) { + crate::ui::error(&format!( + "built binary failed its workflow-store smoke test: {e}" + )); + std::process::exit(1); + } + let target = match std::env::current_exe() { Ok(p) => p, Err(e) => { @@ -177,6 +191,38 @@ fn verify_runs(binary: &Path) -> Result<(), String> { } } +/// Run ` daemon workflow-store-smoke-test` (the hidden verb behind +/// [`crate::cli::daemon::DaemonSubcommand::WorkflowStoreSmokeTest`]) and +/// confirm it exits successfully within [`VERIFY_TIMEOUT`] — same +/// spawn/try_wait/kill pattern as [`verify_runs`], so a hung smoke test +/// can't block `dev-install` indefinitely. +fn verify_workflow_store(binary: &Path) -> Result<(), String> { + let mut child = Command::new(binary) + .args(["daemon", "workflow-store-smoke-test"]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .map_err(|e| format!("failed to spawn workflow-store-smoke-test: {e}"))?; + + let deadline = Instant::now() + VERIFY_TIMEOUT; + loop { + match child.try_wait() { + Ok(Some(status)) if status.success() => return Ok(()), + Ok(Some(status)) => { + return Err(format!("workflow-store-smoke-test exited with {status}")); + } + Ok(None) => { + if Instant::now() >= deadline { + let _ = child.kill(); + return Err("workflow-store-smoke-test timed out".to_string()); + } + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => return Err(format!("waiting on workflow-store-smoke-test: {e}")), + } + } +} + /// `dev-install` replaces the *one* shared `agentflare` binary and, via its /// migrations, mutates the *one* shared `~/.agentflare/*.db` files. If the /// current branch adds a schema migration, installing it here bakes that @@ -258,4 +304,14 @@ mod tests { let missing = std::env::temp_dir().join("agentflare-nonexistent-binary-xyz"); assert!(verify_runs(&missing).is_err()); } + + #[test] + fn verify_workflow_store_errors_for_a_missing_binary() { + // Same guard as `verify_runs_errors_for_a_missing_binary`, for the + // workflow-store smoke-test subprocess -- the happy path (a real + // built binary passing its store round-trip) is exercised by the + // real `dev-install` flow, not unit tests. + let missing = std::env::temp_dir().join("agentflare-nonexistent-binary-xyz"); + assert!(verify_workflow_store(&missing).is_err()); + } } diff --git a/src/github/pulls.rs b/src/github/pulls.rs index 4620b3099..55198d0f5 100644 --- a/src/github/pulls.rs +++ b/src/github/pulls.rs @@ -1,6 +1,15 @@ use crate::github::models::{PullRequest, Review, ReviewComment}; use crate::github::{Client, GitHubError, RepoId}; +/// Extractor for the Search API's `{"items": [...], "total_count": N}` +/// envelope, mirroring `actions::workflow_runs`/`check_runs`. +fn search_items(page: &serde_json::Value) -> Vec { + page.get("items") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default() +} + fn create_body(title: &str, head: &str, base: &str, body: Option<&str>) -> serde_json::Value { let mut v = serde_json::json!({ "title": title, "head": head, "base": base }); if let Some(b) = body { @@ -51,6 +60,36 @@ pub fn find_existing( .find(|pr| pr.head.as_ref().is_some_and(|h| h.git_ref == branch))) } +/// Finds every PR (open, merged, or closed) whose body carries the +/// `for item #` marker `pr_footer` stamps onto every PR +/// agentflare opens (see `push_and_open_pr`) -- the pre-dispatch +/// duplicate-work check (item #164). Unlike `find_existing`, this doesn't +/// depend on the item's own tracked branch name, so it still finds a PR +/// that merged while the item's tracked state fell out of sync (items +/// #122/#156: the state-side promotion never ran, so a routine redispatch +/// nearly re-did already-merged work). Uses GitHub's search API rather than +/// `list` + filter, mirroring the exact `gh pr list --search "\"for item +/// #N \" in:body"` query a human ran to catch that incident. +pub fn find_by_item_marker( + client: &Client, + repo: &RepoId, + sequence_id: i64, +) -> Result, GitHubError> { + let query = format!( + "repo:{}/{} type:pr \"for item #{sequence_id} \" in:body", + repo.owner, repo.repo + ); + let path = format!("/search/issues?q={}", crate::github::encode_query(&query)); + let items = client.get_paginated(&path, search_items)?; + let numbers: Vec = items + .as_array() + .into_iter() + .flatten() + .filter_map(|item| item["number"].as_u64()) + .collect(); + numbers.into_iter().map(|n| get(client, repo, n)).collect() +} + pub fn get(client: &Client, repo: &RepoId, number: u64) -> Result { let path = format!("/repos/{}/{}/pulls/{number}", repo.owner, repo.repo); let json = client.request("GET", &path, None)?; @@ -254,6 +293,40 @@ mod tests { ); } + #[test] + fn find_by_item_marker_searches_and_fetches_each_matching_pr() { + let server = MockServer::start(vec![ + MockResponse::json(200, r#"{"items":[{"number":42}]}"#), + MockResponse::json( + 200, + r#"{"number":42,"html_url":"u","state":"closed","title":"t","merged_at":"2026-08-01T00:00:00Z"}"#, + ), + ]); + let client = server.client(None); + let found = find_by_item_marker(&client, &repo(), 164).unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].number, 42); + assert!(found[0].merged_at.is_some()); + + let reqs = server.requests(); + assert_eq!(reqs[0].method, "GET"); + assert!(reqs[0].path.starts_with("/search/issues?q=")); + assert!(reqs[0].path.contains("repo%3Ao/r")); + assert!(reqs[0].path.contains("%22for%20item%20%23164%20%22")); + assert_eq!(reqs[1].path, "/repos/o/r/pulls/42"); + } + + #[test] + fn find_by_item_marker_returns_empty_when_search_finds_nothing() { + let server = MockServer::start(vec![MockResponse::json(200, r#"{"items":[]}"#)]); + let client = server.client(None); + assert!( + find_by_item_marker(&client, &repo(), 999) + .unwrap() + .is_empty() + ); + } + #[test] fn get_fetches_a_single_pull() { let server = MockServer::start(vec![MockResponse::json(