From 853dee620cde835e2d0c960f37bd15794763c297 Mon Sep 17 00:00:00 2001
From: Wes <wesbillman@users.noreply.github.com>
Date: Wed, 5 Aug 2026 11:49:39 -0600
Subject: [PATCH] desktop: enumerate pending PMA attempts

Snapshot the latest pending retained aggregate per agent within the captured owner scope so the flush driver can drain immutable retry inputs without crossing identities or replaying superseded generations.

Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>
---
 .../src-tauri/src/managed_agents/retention.rs | 90 ++++++++++++++++---
 1 file changed, 78 insertions(+), 12 deletions(-)

diff --git a/desktop/src-tauri/src/managed_agents/retention.rs b/desktop/src-tauri/src/managed_agents/retention.rs
index 6868f50a2f..c518f68623 100644
--- a/desktop/src-tauri/src/managed_agents/retention.rs
+++ b/desktop/src-tauri/src/managed_agents/retention.rs
@@ -230,23 +230,65 @@ pub fn get_retained_managed_agent_aggregate(
           ORDER BY generation DESC
           LIMIT 1",
         params![owner_pubkey, agent_pubkey],
-        |row| {
-            Ok(RetainedManagedAgentAggregate {
-                owner_pubkey: row.get(0)?,
-                agent_pubkey: row.get(1)?,
-                generation: row.get::<_, i64>(2)? as u64,
-                private_event_id: row.get(3)?,
-                state: row.get(4)?,
-                request_json: row.get(5)?,
-                pending_sync: row.get::<_, i32>(6)? != 0,
-                last_error: row.get(7)?,
-            })
-        },
+        aggregate_from_row,
     )
     .optional()
     .map_err(|e| format!("failed to get retained managed-agent aggregate: {e}"))
 }
 
+/// Snapshot every pending aggregate attempt for one captured owner scope.
+///
+/// At most one generation per agent may be pending during normal operation,
+/// but selecting the latest pending generation defensively avoids replaying a
+/// superseded row after a crash between generation advance and acknowledgement.
+pub fn get_pending_managed_agent_aggregates(
+    conn: &Connection,
+    owner_pubkey: &str,
+) -> Result<Vec<RetainedManagedAgentAggregate>, String> {
+    let mut stmt = conn
+        .prepare(
+            "SELECT a.owner_pubkey, a.agent_pubkey, a.generation,
+                    a.private_event_id, a.state, a.request_json,
+                    a.pending_sync, a.last_error
+               FROM managed_agent_aggregates a
+              WHERE a.owner_pubkey = ?1 AND a.pending_sync = 1
+                AND a.generation = (
+                    SELECT MAX(latest.generation)
+                      FROM managed_agent_aggregates latest
+                     WHERE latest.owner_pubkey = a.owner_pubkey
+                       AND latest.agent_pubkey = a.agent_pubkey
+                )
+              ORDER BY a.agent_pubkey",
+        )
+        .map_err(|e| format!("failed to prepare pending managed-agent aggregates: {e}"))?;
+    let rows = stmt
+        .query_map(params![owner_pubkey], aggregate_from_row)
+        .map_err(|e| format!("failed to query pending managed-agent aggregates: {e}"))?;
+    rows.collect::<Result<Vec<_>, _>>()
+        .map_err(|e| format!("failed to read pending managed-agent aggregate: {e}"))
+}
+
+fn aggregate_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RetainedManagedAgentAggregate> {
+    let generation = row.get::<_, i64>(2)?;
+    let generation = u64::try_from(generation).map_err(|error| {
+        rusqlite::Error::FromSqlConversionFailure(
+            2,
+            rusqlite::types::Type::Integer,
+            Box::new(error),
+        )
+    })?;
+    Ok(RetainedManagedAgentAggregate {
+        owner_pubkey: row.get(0)?,
+        agent_pubkey: row.get(1)?,
+        generation,
+        private_event_id: row.get(3)?,
+        state: row.get(4)?,
+        request_json: row.get(5)?,
+        pending_sync: row.get::<_, i32>(6)? != 0,
+        last_error: row.get(7)?,
+    })
+}
+
 /// Mark one exact retained aggregate request as confirmed by relay read-back.
 #[allow(dead_code)] // Consumed by the PMA aggregate submit/retry driver in the next slice.
 pub fn mark_managed_agent_aggregate_synced(
@@ -811,6 +853,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn pending_aggregate_snapshot_is_owner_scoped_and_latest_only() {
+        let mut conn = test_db();
+        retain_managed_agent_aggregate(
+            &mut conn,
+            &aggregate(1, "event-a1", "active", r#"{"generation":1}"#),
+        )
+        .unwrap();
+        retain_managed_agent_aggregate(
+            &mut conn,
+            &aggregate(2, "event-a2", "deleted", r#"{"generation":2}"#),
+        )
+        .unwrap();
+        let mut other = aggregate(1, "event-b1", "active", r#"{"generation":1}"#);
+        other.owner_pubkey = "other-owner".into();
+        other.agent_pubkey = "other-agent".into();
+        retain_managed_agent_aggregate(&mut conn, &other).unwrap();
+
+        let pending = get_pending_managed_agent_aggregates(&conn, "owner").unwrap();
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].generation, 2);
+        assert_eq!(pending[0].private_event_id, "event-a2");
+    }
+
     #[test]
     fn aggregate_compare_and_clear_cannot_ack_a_newer_attempt() {
         let mut conn = test_db();