diff --git a/hive-c0re/src/forge/config_pr_poll.rs b/hive-c0re/src/forge/config_pr_poll.rs index b74b587a..ba233b44 100644 --- a/hive-c0re/src/forge/config_pr_poll.rs +++ b/hive-c0re/src/forge/config_pr_poll.rs @@ -93,32 +93,12 @@ pub async fn poll_open_config_prs(core_token: &str, coord: &Arc) -> }; open_prs.insert((agent.to_owned(), pr_number)); - // Skip if a pending approval already exists for this PR. - match coord - .approvals - .has_pending_merge_config_pr(agent, pr_number) - { - Ok(true) => { - tracing::debug!( - %agent, %pr_number, - "config-pr poll: approval already pending, skipping" - ); - continue; - } - Ok(false) => {} - Err(e) => { - tracing::warn!( - %agent, %pr_number, error = ?e, - "config-pr poll: DB check failed, skipping" - ); - continue; - } - } - - tracing::info!( - %agent, %pr_number, - "config-pr poll: queuing missed MergeConfigPr approval" - ); + // `submit_merge_config_pr` is idempotent + PR-drift aware: it + // no-ops when an approval pinned to this PR's *current* head is + // already pending, and cancels+re-queues when the head has drifted. + // So the poll can call it unconditionally — it backstops both a + // missed `opened` webhook (no approval yet) and a missed + // `synchronize` (stale approval whose head moved). let description = format!("PR #{pr_number} on {CONFIG_ORG}/{agent} (poll fallback)"); if let Err(e) = crate::socket_server::submit_merge_config_pr( coord, @@ -131,7 +111,7 @@ pub async fn poll_open_config_prs(core_token: &str, coord: &Arc) -> { tracing::warn!( %agent, %pr_number, error = ?e, - "config-pr poll: failed to queue MergeConfigPr approval" + "config-pr poll: failed to reconcile MergeConfigPr approval" ); } } diff --git a/hive-c0re/src/socket_server/config_approvals.rs b/hive-c0re/src/socket_server/config_approvals.rs index 37bdc2e7..6a3ea8f3 100644 --- a/hive-c0re/src/socket_server/config_approvals.rs +++ b/hive-c0re/src/socket_server/config_approvals.rs @@ -157,6 +157,31 @@ pub(crate) async fn submit_merge_config_pr( let sha = crate::forge::pr_head_sha(&repo, pr_number) .await .map_err(|e| anyhow::anyhow!("fetch PR head sha for {agent} PR #{pr_number}: {e}"))?; + // Both the webhook (`synchronize`) and the poll fallback call this on + // every PR update. If an approval for this PR is already pending, reconcile + // it against the live head sha rather than blindly queuing another: + // - same sha → the PR hasn't moved, so this is a duplicate signal — no-op. + // - drifted sha → the reviewed head is stale. Don't mutate the + // pending row in place (that races a concurrent approve); cancel it and + // fall through to queue a FRESH approval pinned to the new head. + if let Some((old_id, old_sha)) = coord.approvals.pending_merge_config_pr(agent, pr_number)? { + if old_sha.as_deref() == Some(sha.as_str()) { + return Ok(old_id); + } + let cancelled = coord + .approvals + .mark_cancelled(old_id, "config PR updated — superseded by a fresh approval") + .map_err(|e| anyhow::anyhow!("cancel superseded merge_config_pr approval: {e:#}"))?; + coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { + id: old_id, + agent, + approval_kind: "merge_config_pr", + sha_short: old_sha.map(|s| s[..s.len().min(12)].to_owned()), + status: "cancelled", + note: Some("PR head moved; superseded by a fresh approval".to_owned()), + description: cancelled.description, + }); + } let id = coord .approvals .submit_kind( diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs index 1f511e35..da67397a 100644 --- a/hive-c0re/src/stores/approvals.rs +++ b/hive-c0re/src/stores/approvals.rs @@ -136,19 +136,29 @@ impl Approvals { Ok(()) } - /// Return `true` when there is already a `pending` `merge_config_pr` - /// approval for `(agent, pr_number)`. Used by the polling fallback to - /// skip re-submitting approvals that were already queued by the webhook. - pub fn has_pending_merge_config_pr(&self, agent: &str, pr_number: u64) -> Result { + /// Return the `(id, fetched_sha)` of the pending `merge_config_pr` + /// approval for `(agent, pr_number)`, if one exists. Drives + /// `submit_merge_config_pr`'s idempotency + PR-drift handling: same + /// `fetched_sha` → no new request (the webhook + poll both call submit, + /// so re-submits of an unchanged PR must be no-ops); a drifted head → + /// cancel this stale row and queue a fresh approval. + pub fn pending_merge_config_pr( + &self, + agent: &str, + pr_number: u64, + ) -> Result)>> { let conn = self.conn.lock().unwrap(); - let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM approvals \ - WHERE agent = ?1 AND kind = 'merge_config_pr' \ - AND commit_ref = ?2 AND status = 'pending'", - params![agent, pr_number.to_string()], - |row| row.get(0), - )?; - Ok(count > 0) + let row = conn + .query_row( + "SELECT id, fetched_sha FROM approvals \ + WHERE agent = ?1 AND kind = 'merge_config_pr' \ + AND commit_ref = ?2 AND status = 'pending' \ + ORDER BY id DESC LIMIT 1", + params![agent, pr_number.to_string()], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + Ok(row) } /// Last `limit` resolved approvals (approved / denied / failed),