Proposing the same write again withdraws the older held copy

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 Written by hand. Each release says what changed for someone running Rowbit, not every commit.
 
+## Unreleased
+
+### Fixed
+
+- When an agent proposes the same statement on the same database again, its older held copy is withdrawn, with a `Superseded` note naming the new change. Its review page links the new change and won't approve the old one. A held change left over from an earlier run could be approved in its place: it applied, while the change the agent was watching stayed held, and approving that one too failed on drift.
+
 ## 0.2.2 (2026-11-07)
 
 Set up on your database: `rowbit setup` checks your database's URL and Rowbit's grants, writes rules that hold every write, and connects your agents; TLS to the database; a container image and an MCP Registry listing.

crates/rowbit-engine/src/lib.rs

@@ -359,6 +359,13 @@ impl Engine {
             };
             store.transition_notifying(id, &event, &actor, None, &notify)?;
         }
+        if decision != Decision::Block || shadow {
+            // Only tidies up: the new change is recorded and decided, so a
+            // failure here must not tell the agent its proposal failed.
+            if let Err(e) = self.supersede(agent, database, &classified.canonical, id) {
+                eprintln!("rowbit: change {id}: couldn't withdraw older held copies: {e}");
+            }
+        }
         if decision == Decision::Allow || shadow {
             self.apply(id).await?;
         }
@@ -571,6 +578,37 @@ impl Engine {
         Ok(self.store().verify()?)
     }
 
+    /// Withdraws `agent`'s older held changes that are the same statement
+    /// on the same database as `newer`, so a person never approves a stale
+    /// copy of what the agent just proposed again. Each one gets a
+    /// `Superseded` note naming `newer`, in the same transaction as the
+    /// withdraw. A copy a person decided on in the meantime is left alone.
+    ///
+    /// Matches on the agent's name, not the connection: two projects
+    /// connected as the same agent replace each other's held copies.
+    fn supersede(
+        &self,
+        agent: &str,
+        database: &str,
+        canonical: &str,
+        newer: ChangeId,
+    ) -> Result<(), EngineError> {
+        let mut store = self.store();
+        for id in store.list(State::Held)? {
+            if id == newer {
+                continue;
+            }
+            let old = store.get(id)?;
+            let same = old.agent == agent
+                && old.database == database
+                && classify(&old.sql).is_ok_and(|c| c.canonical == canonical);
+            if same {
+                store.withdraw_superseded(id, agent, &format!("{SUPERSEDED_PREFIX}{newer}"))?;
+            }
+        }
+        Ok(())
+    }
+
     /// Changes waiting for a person, oldest first.
     pub fn held(&self) -> Result<Vec<ChangeId>, EngineError> {
         Ok(self.store().list(State::Held)?)
@@ -695,8 +733,8 @@ impl Engine {
             Ok(_) | Err(StoreError::NotFound(_)) => return Err(EngineError::NotFound(id)),
             Err(e) => return Err(e.into()),
         };
-        let reason = store
-            .events(id)?
+        let history = store.events(id)?;
+        let reason = history
             .iter()
             .rev()
             .find(|e| e.from.as_deref() != Some(e.to.as_str()) || e.event == "Proposed")
@@ -706,7 +744,8 @@ impl Engine {
                     "PolicyHeld" | "PolicyBlocked" | "Reject" | "ApplyFailed"
                 )
             })
-            .and_then(|e| e.detail.clone());
+            .and_then(|e| e.detail.clone())
+            .or_else(|| superseded_by(&history).map(|n| format!("{SUPERSEDED_PREFIX}{n}")));
         let (table, rows) = match &change.review {
             Some(Review::Rows(p)) => (Some(p.table.qualified_name()), Some(p.rows.len())),
             Some(Review::Ddl { .. }) | None => (None, None),
@@ -921,6 +960,23 @@ pub const DEFAULT_REVIEW_ADDRESS: &str = "http://127.0.0.1:8080";
 
 /// A change's page on the review UI at `address`, by way of sign-in if
 /// the browser isn't signed in.
+/// How a `Superseded` note's detail starts; the newer change's id follows.
+const SUPERSEDED_PREFIX: &str = "proposed again as #";
+
+/// The change that replaced this one, from its `Superseded` note.
+pub fn superseded_by(history: &[AuditEvent]) -> Option<ChangeId> {
+    history
+        .iter()
+        .find(|e| e.event == "Superseded")
+        .and_then(|e| {
+            e.detail
+                .as_deref()?
+                .strip_prefix(SUPERSEDED_PREFIX)?
+                .parse()
+                .ok()
+        })
+}
+
 pub fn change_link(address: &str, id: ChangeId) -> String {
     format!("{}/changes/{id}", address.trim_end_matches('/'))
 }

crates/rowbit-engine/tests/propose.rs

@@ -354,3 +354,63 @@ async fn rejected_and_drifted_changes_write_nothing(var: &str, table: &str) {
     assert_eq!(refunded, [4]);
     conn.disconnect().await.unwrap();
 }
+
+// An agent that proposes the same statement again replaces its older held
+// copy, so the person can't approve a stale one by mistake.
+on_both!(proposing_again_withdraws_the_older_held_copy);
+async fn proposing_again_withdraws_the_older_held_copy(var: &str, table: &str) {
+    let Some((engine, t)) = setup(var, &format!("{table}_w")).await else {
+        return;
+    };
+    let sql = format!("UPDATE {t} SET status = 'refunded' WHERE status = 'approved'");
+    let first = propose(&engine, "refund-bot", &sql).await;
+    assert_eq!(first.state, State::Held);
+    let other = propose(
+        &engine,
+        "refund-bot",
+        &format!("UPDATE {t} SET status = 'refunded' WHERE id = 3"),
+    )
+    .await;
+    assert_eq!(other.state, State::Held, "{other:?}");
+
+    // Same statement, spelled differently.
+    let second = propose(
+        &engine,
+        "refund-bot",
+        &format!("update {t}  set status = 'refunded' where status = 'approved'"),
+    )
+    .await;
+    assert_eq!(second.state, State::Held);
+    assert_eq!(engine.held().unwrap(), [other.id, second.id]);
+
+    let old = engine.status("refund-bot", first.id).unwrap();
+    assert_eq!(old.state, State::Rejected);
+    assert_eq!(old.review_url, None);
+    // The agent that looks at the old copy learns why it was withdrawn.
+    let because = format!("proposed again as #{}", second.id);
+    assert_eq!(old.reason.as_deref(), Some(because.as_str()));
+    let history = engine.history(first.id).unwrap();
+    let events: Vec<_> = history
+        .iter()
+        .rev()
+        .take(2)
+        .map(|e| e.event.as_str())
+        .collect();
+    assert_eq!(
+        events,
+        ["Superseded", "Withdraw"],
+        "noted after the withdraw"
+    );
+    assert_eq!(
+        history.last().unwrap().detail.as_deref(),
+        Some(because.as_str())
+    );
+    assert_eq!(rowbit_engine::superseded_by(&history), Some(second.id));
+    assert!(engine.verify().unwrap().is_ok());
+
+    let hash = engine.change(first.id).unwrap().review_hash.unwrap();
+    assert!(engine.approve("philip", first.id, &hash).await.is_err());
+    let hash = engine.change(second.id).unwrap().review_hash.unwrap();
+    let change = engine.approve("philip", second.id, &hash).await.unwrap();
+    assert_eq!(change.state, State::Applied);
+}

crates/rowbit-mcp/tests/attacks.rs

@@ -208,7 +208,9 @@ async fn self_approval(var: &str, table: &str) {
     )
     .await;
     assert!(text.contains("\"held\""), "{text}");
-    assert_eq!(w.engine.held().unwrap().len(), 2);
+    // The same statement again replaces the first copy; neither is approved.
+    let again: Value = serde_json::from_str(&text).unwrap();
+    assert_eq!(w.engine.held().unwrap(), [again["id"].as_i64().unwrap()]);
     w.untouched().await;
 }
 

crates/rowbit-store/src/lib.rs

@@ -456,7 +456,33 @@ impl Store {
         if matches!(event, Event::ApplySucceeded) {
             return Err(StoreError::ApplyNeedsRecord);
         }
-        self.transition_with(id, event, actor, reviewed, None, notify)
+        self.transition_with(id, event, actor, reviewed, None, notify, None)
+    }
+
+    /// Withdraws a held change for `agent`, which proposed it, and logs a
+    /// `Superseded` note with `detail` in the same transaction, so the note
+    /// is never there without the withdraw. Returns false, and logs
+    /// nothing, if the change is no longer held; only the proposer may.
+    pub fn withdraw_superseded(
+        &mut self,
+        id: ChangeId,
+        agent: &str,
+        detail: &str,
+    ) -> Result<bool, StoreError> {
+        let actor = Actor::Agent(agent.to_owned());
+        match self.transition_with(
+            id,
+            &Event::Withdraw,
+            &actor,
+            None,
+            None,
+            &[],
+            Some(("Superseded", detail)),
+        ) {
+            Ok(_) => Ok(true),
+            Err(StoreError::Transition(TransitionError::InvalidTransition { .. })) => Ok(false),
+            Err(e) => Err(e),
+        }
     }
 
     /// Records a successful apply (`ApplySucceeded`, by the system) together
@@ -469,9 +495,13 @@ impl Store {
             None,
             Some(applied.to_canonical_json()),
             &[],
+            None,
         )
     }
 
+    // One private entry point for every transition; the callers above name
+    // what each argument is for.
+    #[allow(clippy::too_many_arguments)]
     fn transition_with(
         &mut self,
         id: ChangeId,
@@ -480,6 +510,7 @@ impl Store {
         reviewed: Option<&str>,
         applied: Option<Vec<u8>>,
         notify: &[NewDelivery],
+        note: Option<(&str, &str)>,
     ) -> Result<State, StoreError> {
         let tx = self
             .conn
@@ -536,6 +567,19 @@ impl Store {
             },
             at,
         )?;
+        if let Some((event, detail)) = note {
+            append(
+                &tx,
+                id,
+                Some(to),
+                to,
+                event,
+                Some(detail),
+                &actor.to_string(),
+                None,
+                at,
+            )?;
+        }
         for d in notify {
             tx.execute(
                 "INSERT INTO deliveries (change_id, target, delivery, payload, state, next_at_ms, created_at_ms)

crates/rowbit-store/src/tests.rs

@@ -641,3 +641,54 @@ fn settings_are_listed_by_prefix_and_deleted() {
     // Characters LIKE would treat as wildcards are literal.
     assert!(store.settings_with_prefix("agent_").unwrap().is_empty());
 }
+
+#[test]
+fn a_superseded_copy_is_withdrawn_and_noted_together() {
+    let mut store = Store::open_in_memory().unwrap();
+    let (id, _) = previewed(&mut store);
+    store
+        .transition(
+            id,
+            &Event::PolicyHeld { reason: "r".into() },
+            &Actor::Policy("default".into()),
+            None,
+        )
+        .unwrap();
+    // Only the proposer's name withdraws it.
+    assert!(
+        store
+            .withdraw_superseded(id, "other-bot", "proposed again as #9")
+            .is_err()
+    );
+    assert_eq!(store.get(id).unwrap().state, State::Held);
+    assert!(
+        store
+            .events(id)
+            .unwrap()
+            .iter()
+            .all(|e| e.event != "Superseded")
+    );
+
+    assert!(
+        store
+            .withdraw_superseded(id, "refund-bot", "proposed again as #9")
+            .unwrap()
+    );
+    assert_eq!(store.get(id).unwrap().state, State::Rejected);
+    let events = store.events(id).unwrap();
+    let last: Vec<_> = events
+        .iter()
+        .rev()
+        .take(2)
+        .map(|e| e.event.as_str())
+        .collect();
+    assert_eq!(last, ["Superseded", "Withdraw"]);
+    // No longer held: nothing changes, nothing is noted.
+    assert!(
+        !store
+            .withdraw_superseded(id, "refund-bot", "proposed again as #10")
+            .unwrap()
+    );
+    assert_eq!(store.events(id).unwrap().len(), events.len());
+    assert!(store.verify().unwrap().is_ok());
+}

crates/rowbit-web/src/api.rs

@@ -184,6 +184,11 @@ fn summary(app: &App, change: &Change) -> Value {
             )
         })
         .and_then(|e| e.detail.clone());
+    let superseded_by = rowbit_engine::superseded_by(&history);
+    let reason = reason.or_else(|| {
+        superseded_by
+            .map(|n| format!("Replaced by #{n}: the agent proposed the same change again."))
+    });
     let shadowed = history.iter().find(|e| e.event == "PolicyShadowed");
     json!({
         "id": change.id,
@@ -199,6 +204,8 @@ fn summary(app: &App, change: &Change) -> Value {
         "kind": kind,
         "rows": rows,
         "reason": reason.map(|r| visible(&r)),
+        // The change that replaced this one, when the agent proposed it again.
+        "superseded_by": superseded_by,
         "decided_by": decided.map(|e| visible(&e.actor)),
         "created_at_ms": change.created_at_ms,
     })
@@ -364,6 +371,14 @@ pub(crate) async fn approve(
         Ok(c) => c,
         Err(e) => return error(StatusCode::NOT_FOUND, &e.to_string()),
     };
+    if change.state != ChangeState::Held
+        && let Some(n) = rowbit_engine::superseded_by(&app.engine.history(id).unwrap_or_default())
+    {
+        return error(
+            StatusCode::CONFLICT,
+            &format!("#{id} was replaced by #{n}, the agent's newer copy; review #{n} instead"),
+        );
+    }
     if let (_, _, Some(total)) = table_and_kind(&change)
         && total > CONFIRM_ROWS_ABOVE
         && body.rows != Some(total)