perf(desktop): batch secret projection into one blob mutation with gen reuse

The seam wrote each secret field with its own mutate_blob call and minted
a fresh generation every save, so a metadata-only save churned the keyring
blob and left GC-eligible orphan generations behind. write_secrets_batched
reuses the live generation when a field's bytes are unchanged (no write,
no new UUID, no GC churn) and commits every changed field in a single
store_batch_verified mutation. Per-field cancel_gc_candidacy is dropped on
this path: under the txn lock GC cannot run, a fresh UUID gen was never
observed by a sweep, and a reused gen is a live ref sweeps skip, so no
candidate marker can exist at save time. Verification bypasses the cache
via verify_stored_raw so a backend that acks a write it did not persist is
still caught.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

desktop/src-tauri/src/managed_agents/secret_projection.rs

@@ -147,6 +147,26 @@ pub trait ProjectionStore {
     fn load_all(&self) -> Result<Option<HashMap<String, String>>, String>;
     fn store_batch(&self, entries: &HashMap<String, String>) -> Result<(), String>;
     fn remove_batch(&self, keys: &[&str]) -> Result<(), String>;
+
+    /// Commit `entries` in a single blob mutation, then confirm each key holds
+    /// its value with a durable, cache-bypassing read — the batched analogue of
+    /// [`write_and_verify`](Self::write_and_verify).
+    ///
+    /// The default verifies via [`load_key`](Self::load_key), which is exact for
+    /// stores with no cache/durable split (the test fakes). [`SecretStore`]
+    /// overrides it with a raw keychain read so a backend that acknowledges a
+    /// write it did not persist is still caught, exactly as the per-key path
+    /// does.
+    fn store_batch_verified(&self, entries: &HashMap<String, String>) -> Result<(), String> {
+        self.store_batch(entries)?;
+        for (key, value) in entries {
+            match self.load_key(key)? {
+                Some(ref stored) if stored == value => {}
+                _ => return Err(format!("keyring read-back verify failed for {key}")),
+            }
+        }
+        Ok(())
+    }
 }
 
 impl ProjectionStore for SecretStore {
@@ -177,6 +197,24 @@ impl ProjectionStore for SecretStore {
         }
         Ok(())
     }
+
+    /// Override the default so verification bypasses the in-process cache: after
+    /// `store_all` advances the cache to the written state, a cached `load`
+    /// would pass even when the OS keychain write silently failed.
+    /// [`verify_stored_raw`](SecretStore::verify_stored_raw) reads the raw blob
+    /// direct from the backend, proving the round-trip the same way the per-key
+    /// `write_and_verify` does.
+    fn store_batch_verified(&self, entries: &HashMap<String, String>) -> Result<(), String> {
+        self.store_all(entries)?;
+        for (key, value) in entries {
+            match self.verify_stored_raw(key, value) {
+                Ok(true) => {}
+                Ok(false) => return Err(format!("keyring read-back verify failed for {key}")),
+                Err(e) => return Err(format!("keyring read-back verify error for {key}: {e}")),
+            }
+        }
+        Ok(())
+    }
 }
 
 // ── Projection outcome ─────────────────────────────────────────────────────
@@ -223,6 +261,126 @@ pub fn write_secret<S: ProjectionStore>(
     }
 }
 
+// ── Batched write-with-reuse ───────────────────────────────────────────────
+
+/// One field's contribution to a batched secret save.
+///
+/// The coordinate builder is `&dyn Fn` so a single [`write_secrets_batched`]
+/// call can carry fields from different namespaces (env / auth_tag /
+/// provider_config) without monomorphizing over each closure type.
+pub struct FieldSave<'a> {
+    /// Builds the full blob coordinate for a given generation id.
+    pub coord_key_fn: &'a dyn Fn(&str) -> String,
+    /// The serialized inline value, or `None` when the field is empty/absent.
+    pub value: Option<&'a str>,
+    /// The generation currently referenced on disk, if any — the reuse anchor.
+    pub existing_ref: Option<&'a str>,
+    /// Human-readable context for diagnostics.
+    pub context: &'a str,
+}
+
+/// Persist several secret fields in a SINGLE blob mutation, reusing the live
+/// generation for any field whose bytes are unchanged.
+///
+/// Returns one [`WriteOutcome`] per input field, in order, so the seam applies
+/// the same per-field record mutation it always has:
+///
+/// - `value == None` → [`WriteOutcome::Nothing`] (no write).
+/// - `value` byte-equal to what `existing_ref` already stores →
+///   [`WriteOutcome::Persisted`] with the SAME generation and **no write**: no
+///   new UUID is minted, no blob mutation happens, and the generation never
+///   becomes GC-eligible, so a metadata-only save is free of churn.
+/// - `value` changed (or no prior ref, or the prior value is unreadable) → a
+///   fresh generation is staged and committed with every other changed field in
+///   ONE [`store_batch_verified`](ProjectionStore::store_batch_verified). On
+///   success each is [`WriteOutcome::Persisted`] with its new gen; if the single
+///   mutation fails, ALL staged fields become [`WriteOutcome::KeptInline`]
+///   together (the blob write is atomic — there is no torn partial state).
+///
+/// # Why the old generation is safe
+///
+/// Like [`write_secret`], this never deletes a prior generation: a changed field
+/// mints a NEW coordinate and leaves the old one untouched, so a subsequent
+/// failed JSON commit still finds the on-disk ref's generation live. Retirement
+/// stays with the two-cycle GC.
+///
+/// # Why no `cancel_gc_candidacy`
+///
+/// The per-field seam cancelled candidacy after each write. That is redundant on
+/// this path and is deliberately dropped so the save is exactly one mutation:
+/// every save holds the cross-process transaction lock, under which GC cannot
+/// run, and a freshly-minted UUID generation has never been observed by a sweep
+/// (so it carries no candidate marker), while a REUSED generation is a live JSON
+/// ref that `mark_gc_candidates` skips by construction. Neither a new nor a
+/// reused generation can hold a candidate marker at save time, so there is
+/// nothing to cancel. (The boot-migration path keeps its cancel: it is
+/// contract-frozen and its semantics are proven elsewhere.)
+pub fn write_secrets_batched<S: ProjectionStore>(
+    store: &S,
+    fields: &[FieldSave<'_>],
+) -> Vec<WriteOutcome> {
+    let mut outcomes: Vec<Option<WriteOutcome>> = Vec::with_capacity(fields.len());
+    let mut batch: HashMap<String, String> = HashMap::new();
+    // (field index, freshly-minted gen) for each field staged into `batch`,
+    // finalized after the one write resolves.
+    let mut pending: Vec<(usize, String)> = Vec::new();
+
+    for (idx, field) in fields.iter().enumerate() {
+        let Some(value) = field.value else {
+            outcomes.push(Some(WriteOutcome::Nothing));
+            continue;
+        };
+        // Gen-reuse: if the live ref already stores these exact bytes, keep the
+        // generation and write nothing. A load error or absent value falls
+        // through to a fresh write (which, on a real outage, fails closed to
+        // KeptInline) — never a silent reuse of an unverified generation.
+        if let Some(existing) = field.existing_ref {
+            if matches!(
+                store.load_key(&(field.coord_key_fn)(existing)),
+                Ok(Some(ref stored)) if stored == value
+            ) {
+                outcomes.push(Some(WriteOutcome::Persisted {
+                    gen: existing.to_string(),
+                }));
+                continue;
+            }
+        }
+        let gen = new_gen_id();
+        batch.insert((field.coord_key_fn)(&gen), value.to_string());
+        pending.push((idx, gen));
+        outcomes.push(None); // finalized after the batched write
+    }
+
+    if !batch.is_empty() {
+        match store.store_batch_verified(&batch) {
+            Ok(()) => {
+                for (idx, gen) in pending {
+                    outcomes[idx] = Some(WriteOutcome::Persisted { gen });
+                }
+            }
+            Err(e) => {
+                let contexts: Vec<&str> = pending
+                    .iter()
+                    .map(|(idx, _)| fields[*idx].context)
+                    .collect();
+                eprintln!(
+                    "buzz-desktop: batched keyring write failed ({e}); keeping \
+                     inline in 0o600 JSON (retry on next boot): {}",
+                    contexts.join(", ")
+                );
+                for (idx, _) in pending {
+                    outcomes[idx] = Some(WriteOutcome::KeptInline { reason: e.clone() });
+                }
+            }
+        }
+    }
+
+    outcomes
+        .into_iter()
+        .map(|o| o.expect("every field assigned an outcome"))
+        .collect()
+}
+
 // ── Load-with-availability ─────────────────────────────────────────────────
 
 /// The conflict-marker key for a projection coordinate: `conflict:<coord>`.
@@ -786,3 +944,7 @@ pub fn deserialize_provider_config(s: &str) -> Result<serde_json::Value, String>
 #[cfg(test)]
 #[path = "secret_projection_tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "secret_projection_batched_tests.rs"]
+mod batched_tests;

desktop/src-tauri/src/managed_agents/secret_projection_batched_tests.rs

@@ -0,0 +1,354 @@
+//! Coverage for [`write_secrets_batched`] — the metadata-save fast path that
+//! reuses a live generation when a field's bytes are unchanged and commits
+//! every changed field in one blob mutation.
+//!
+//! Split out of `secret_projection_tests.rs` so each test file stays under the
+//! desktop file-size ratchet; both share the `secret_projection` module via
+//! `use super::*`.
+
+use super::*;
+use std::cell::RefCell;
+
+/// A store that counts blob mutations (`store_batch`) and per-key writes
+/// (`write_and_verify`) so tests can assert the exact I/O a save performs.
+struct CountingStore {
+    data: RefCell<HashMap<String, String>>,
+    batch_mutations: std::cell::Cell<usize>,
+    single_writes: std::cell::Cell<usize>,
+}
+
+impl CountingStore {
+    fn new() -> Self {
+        Self {
+            data: RefCell::new(HashMap::new()),
+            batch_mutations: std::cell::Cell::new(0),
+            single_writes: std::cell::Cell::new(0),
+        }
+    }
+    fn with_entry(self, key: &str, value: &str) -> Self {
+        self.data
+            .borrow_mut()
+            .insert(key.to_string(), value.to_string());
+        self
+    }
+    fn get(&self, key: &str) -> Option<String> {
+        self.data.borrow().get(key).cloned()
+    }
+    fn keys(&self) -> Vec<String> {
+        self.data.borrow().keys().cloned().collect()
+    }
+}
+
+impl ProjectionStore for CountingStore {
+    fn write_and_verify(&self, key: &str, value: &str) -> Result<(), String> {
+        self.single_writes.set(self.single_writes.get() + 1);
+        self.data
+            .borrow_mut()
+            .insert(key.to_string(), value.to_string());
+        Ok(())
+    }
+    fn load_key(&self, key: &str) -> Result<Option<String>, String> {
+        Ok(self.data.borrow().get(key).cloned())
+    }
+    fn load_all(&self) -> Result<Option<HashMap<String, String>>, String> {
+        Ok(Some(self.data.borrow().clone()))
+    }
+    fn store_batch(&self, entries: &HashMap<String, String>) -> Result<(), String> {
+        self.batch_mutations.set(self.batch_mutations.get() + 1);
+        for (k, v) in entries {
+            self.data.borrow_mut().insert(k.clone(), v.clone());
+        }
+        Ok(())
+    }
+    fn remove_batch(&self, keys: &[&str]) -> Result<(), String> {
+        for k in keys {
+            self.data.borrow_mut().remove(*k);
+        }
+        Ok(())
+    }
+}
+
+/// A store whose blob write always fails — a denied keychain prompt or a
+/// transient outage — so the batched save must fail closed to KeptInline.
+struct FailingBatchStore;
+
+impl ProjectionStore for FailingBatchStore {
+    fn write_and_verify(&self, _key: &str, _value: &str) -> Result<(), String> {
+        Err("write failed".to_string())
+    }
+    fn load_key(&self, _key: &str) -> Result<Option<String>, String> {
+        Ok(None)
+    }
+    fn load_all(&self) -> Result<Option<HashMap<String, String>>, String> {
+        Ok(None)
+    }
+    fn store_batch(&self, _entries: &HashMap<String, String>) -> Result<(), String> {
+        Err("blob write failed".to_string())
+    }
+    fn remove_batch(&self, _keys: &[&str]) -> Result<(), String> {
+        Ok(())
+    }
+}
+
+#[test]
+fn test_write_secrets_batched_reuses_gen_when_bytes_unchanged() {
+    // A field whose live ref already stores these exact bytes must keep its
+    // generation and perform NO write — the metadata-only-save fast path.
+    let store = CountingStore::new().with_entry(&agent_env_key("abc", "gen1"), r#"{"K":"v"}"#);
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[FieldSave {
+            coord_key_fn: &env_key,
+            value: Some(r#"{"K":"v"}"#),
+            existing_ref: Some("gen1"),
+            context: "agent:abc env_vars",
+        }],
+    );
+    assert_eq!(
+        outcomes[0],
+        WriteOutcome::Persisted {
+            gen: "gen1".to_string()
+        },
+        "unchanged bytes must reuse the existing generation"
+    );
+    assert_eq!(
+        store.batch_mutations.get(),
+        0,
+        "no blob mutation for an unchanged field"
+    );
+    assert_eq!(store.keys().len(), 1, "no new generation key was created");
+}
+
+#[test]
+fn test_write_secrets_batched_new_gen_when_bytes_changed() {
+    let store = CountingStore::new().with_entry(&agent_env_key("abc", "gen1"), r#"{"K":"old"}"#);
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[FieldSave {
+            coord_key_fn: &env_key,
+            value: Some(r#"{"K":"new"}"#),
+            existing_ref: Some("gen1"),
+            context: "agent:abc env_vars",
+        }],
+    );
+    let new_gen = match &outcomes[0] {
+        WriteOutcome::Persisted { gen } => gen.clone(),
+        other => panic!("expected Persisted, got {other:?}"),
+    };
+    assert_ne!(
+        new_gen, "gen1",
+        "changed bytes must mint a fresh generation"
+    );
+    assert_eq!(store.batch_mutations.get(), 1);
+    assert_eq!(
+        store.get(&agent_env_key("abc", "gen1")),
+        Some(r#"{"K":"old"}"#.to_string()),
+        "the old generation must survive — a failed JSON commit still needs it"
+    );
+    assert_eq!(
+        store.get(&agent_env_key("abc", &new_gen)),
+        Some(r#"{"K":"new"}"#.to_string())
+    );
+}
+
+#[test]
+fn test_write_secrets_batched_single_mutation_for_multiple_changed_fields() {
+    // Three changed fields across three namespaces must commit in EXACTLY one
+    // blob mutation and zero per-key writes.
+    let store = CountingStore::new();
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let auth_key = |gen: &str| agent_auth_tag_key("abc", gen);
+    let pc_key = |gen: &str| agent_provider_config_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[
+            FieldSave {
+                coord_key_fn: &env_key,
+                value: Some(r#"{"K":"v"}"#),
+                existing_ref: None,
+                context: "agent:abc env_vars",
+            },
+            FieldSave {
+                coord_key_fn: &auth_key,
+                value: Some("auth-tag"),
+                existing_ref: None,
+                context: "agent:abc auth_tag",
+            },
+            FieldSave {
+                coord_key_fn: &pc_key,
+                value: Some(r#"{"host":"x"}"#),
+                existing_ref: None,
+                context: "agent:abc provider_config",
+            },
+        ],
+    );
+    assert!(
+        outcomes
+            .iter()
+            .all(|o| matches!(o, WriteOutcome::Persisted { .. })),
+        "all three fields persist"
+    );
+    assert_eq!(
+        store.batch_mutations.get(),
+        1,
+        "three changed fields must be ONE blob mutation, not three"
+    );
+    assert_eq!(
+        store.single_writes.get(),
+        0,
+        "the batched path must not fall back to per-key writes"
+    );
+    assert_eq!(store.keys().len(), 3, "one generation per field");
+}
+
+#[test]
+fn test_write_secrets_batched_mixed_reuse_and_change_is_one_mutation() {
+    // One unchanged field (reuse, no write) + one changed field (new gen):
+    // still exactly one mutation, and the unchanged field keeps its gen.
+    let store = CountingStore::new()
+        .with_entry(&agent_env_key("abc", "gen_env"), r#"{"E":"same"}"#)
+        .with_entry(&agent_auth_tag_key("abc", "gen_auth"), "old-auth");
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let auth_key = |gen: &str| agent_auth_tag_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[
+            FieldSave {
+                coord_key_fn: &env_key,
+                value: Some(r#"{"E":"same"}"#),
+                existing_ref: Some("gen_env"),
+                context: "agent:abc env_vars",
+            },
+            FieldSave {
+                coord_key_fn: &auth_key,
+                value: Some("new-auth"),
+                existing_ref: Some("gen_auth"),
+                context: "agent:abc auth_tag",
+            },
+        ],
+    );
+    assert_eq!(
+        outcomes[0],
+        WriteOutcome::Persisted {
+            gen: "gen_env".to_string()
+        },
+        "unchanged env reuses its gen"
+    );
+    let new_auth = match &outcomes[1] {
+        WriteOutcome::Persisted { gen } => gen.clone(),
+        other => panic!("expected Persisted, got {other:?}"),
+    };
+    assert_ne!(new_auth, "gen_auth");
+    assert_eq!(
+        store.batch_mutations.get(),
+        1,
+        "only the changed field triggers the single mutation"
+    );
+}
+
+#[test]
+fn test_write_secrets_batched_nothing_on_empty_value() {
+    let store = CountingStore::new();
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[FieldSave {
+            coord_key_fn: &env_key,
+            value: None,
+            existing_ref: None,
+            context: "agent:abc env_vars",
+        }],
+    );
+    assert_eq!(outcomes[0], WriteOutcome::Nothing);
+    assert_eq!(
+        store.batch_mutations.get(),
+        0,
+        "an empty field writes nothing"
+    );
+}
+
+#[test]
+fn test_write_secrets_batched_all_kept_inline_on_write_failure() {
+    // The single blob write is atomic: if it fails, every staged field becomes
+    // KeptInline together — no torn partial state where some fields persisted.
+    let store = FailingBatchStore;
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let auth_key = |gen: &str| agent_auth_tag_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[
+            FieldSave {
+                coord_key_fn: &env_key,
+                value: Some(r#"{"K":"v"}"#),
+                existing_ref: None,
+                context: "agent:abc env_vars",
+            },
+            FieldSave {
+                coord_key_fn: &auth_key,
+                value: Some("auth"),
+                existing_ref: None,
+                context: "agent:abc auth_tag",
+            },
+        ],
+    );
+    assert!(
+        outcomes
+            .iter()
+            .all(|o| matches!(o, WriteOutcome::KeptInline { .. })),
+        "a failed atomic write keeps ALL staged fields inline: {outcomes:?}"
+    );
+}
+
+#[test]
+fn test_write_secrets_batched_changes_when_prior_gen_unreadable() {
+    // existing_ref points at a gen whose value is absent (outage/deleted). Reuse
+    // must NOT fire on an unverified generation — fall through to a fresh write.
+    let store = CountingStore::new();
+    let env_key = |gen: &str| agent_env_key("abc", gen);
+    let outcomes = write_secrets_batched(
+        &store,
+        &[FieldSave {
+            coord_key_fn: &env_key,
+            value: Some(r#"{"K":"v"}"#),
+            existing_ref: Some("gen_missing"),
+            context: "agent:abc env_vars",
+        }],
+    );
+    match &outcomes[0] {
+        WriteOutcome::Persisted { gen } => assert_ne!(gen, "gen_missing"),
+        other => panic!("expected fresh Persisted, got {other:?}"),
+    }
+    assert_eq!(store.batch_mutations.get(), 1);
+}
+
+#[test]
+fn test_store_batch_verified_default_catches_silent_write_loss() {
+    // A store whose store_batch is a no-op (acknowledges without persisting)
+    // must be caught by the verify pass, not reported as success.
+    struct LyingStore;
+    impl ProjectionStore for LyingStore {
+        fn write_and_verify(&self, _k: &str, _v: &str) -> Result<(), String> {
+            Ok(())
+        }
+        fn load_key(&self, _k: &str) -> Result<Option<String>, String> {
+            Ok(None) // nothing was actually stored
+        }
+        fn load_all(&self) -> Result<Option<HashMap<String, String>>, String> {
+            Ok(None)
+        }
+        fn store_batch(&self, _entries: &HashMap<String, String>) -> Result<(), String> {
+            Ok(()) // lies: acknowledges but persists nothing
+        }
+        fn remove_batch(&self, _keys: &[&str]) -> Result<(), String> {
+            Ok(())
+        }
+    }
+    let mut entries = HashMap::new();
+    entries.insert("global:env:gen1".to_string(), "sk-secret".to_string());
+    assert!(
+        LyingStore.store_batch_verified(&entries).is_err(),
+        "verify must reject a write the backend did not persist"
+    );
+}

desktop/src-tauri/src/managed_agents/secret_seam.rs

@@ -9,7 +9,8 @@ use crate::managed_agents::{
     secret_projection::{
         agent_auth_tag_key, agent_env_key, agent_provider_config_key, cancel_gc_candidacy,
         definition_env_key, deserialize_env_map, deserialize_provider_config, load_secret,
-        serialize_env_map, serialize_provider_config, write_secret, ProjectionStore, WriteOutcome,
+        serialize_env_map, serialize_provider_config, write_secret, write_secrets_batched,
+        FieldSave, ProjectionStore, WriteOutcome,
     },
     BackendKind, ManagedAgentRecord,
 };
@@ -105,21 +106,70 @@ pub(crate) fn strip_and_persist_agent_secrets_with<S: ProjectionStore>(
 ) {
     let unavailable = record.secrets_unavailable;
     let pubkey = record.pubkey.clone();
+
+    // Serialize each field's inline value once. `None` means "field empty" and
+    // maps to WriteOutcome::Nothing (no write, no reuse).
     let inline_env = if !record.env_vars.is_empty() {
         serialize_env_map(&record.env_vars).ok()
     } else {
         None
     };
-    match write_secret(
-        store,
-        |gen| agent_env_key(&pubkey, gen),
-        inline_env.as_deref(),
-        &format!("agent:{pubkey} env_vars"),
-    ) {
+    let auth_val = record.auth_tag.clone();
+    let provider = match &record.backend {
+        BackendKind::Provider { id, config } => {
+            let serialized = if !config.is_null() {
+                serialize_provider_config(config).ok()
+            } else {
+                None
+            };
+            Some((id.clone(), serialized))
+        }
+        _ => None,
+    };
+
+    // Coordinate builders borrow `pubkey`; contexts are owned so they outlive
+    // the batched call. Both are dropped before `record` is mutated below.
+    let env_key = |gen: &str| agent_env_key(&pubkey, gen);
+    let auth_key = |gen: &str| agent_auth_tag_key(&pubkey, gen);
+    let pc_key = |gen: &str| agent_provider_config_key(&pubkey, gen);
+    let env_ctx = format!("agent:{pubkey} env_vars");
+    let auth_ctx = format!("agent:{pubkey} auth_tag");
+    let pc_ctx = provider
+        .as_ref()
+        .map(|(id, _)| format!("agent:{pubkey} provider_config ({id})"));
+
+    let mut fields = vec![
+        FieldSave {
+            coord_key_fn: &env_key,
+            value: inline_env.as_deref(),
+            existing_ref: record.env_vars_ref.as_deref(),
+            context: &env_ctx,
+        },
+        FieldSave {
+            coord_key_fn: &auth_key,
+            value: auth_val.as_deref(),
+            existing_ref: record.auth_tag_ref.as_deref(),
+            context: &auth_ctx,
+        },
+    ];
+    if let Some((_, serialized)) = &provider {
+        fields.push(FieldSave {
+            coord_key_fn: &pc_key,
+            value: serialized.as_deref(),
+            existing_ref: record.provider_config_ref.as_deref(),
+            context: pc_ctx.as_deref().unwrap_or_default(),
+        });
+    }
+
+    // ONE blob mutation for every changed field; unchanged fields reuse their
+    // live generation and write nothing.
+    let outcomes = write_secrets_batched(store, &fields);
+    drop(fields); // end the immutable borrow of `record` before mutating it.
+
+    match &outcomes[0] {
         WriteOutcome::Persisted { gen } => {
-            cancel_gc_candidacy(store, &agent_env_key(&pubkey, &gen));
             record.env_vars.clear();
-            record.env_vars_ref = Some(gen);
+            record.env_vars_ref = Some(gen.clone());
         }
         // Empty projection: preserve an existing ref when unavailable so a
         // failed-hydrate save does not orphan the live generation.
@@ -128,43 +178,23 @@ pub(crate) fn strip_and_persist_agent_secrets_with<S: ProjectionStore>(
             record.env_vars_ref = None;
         }
     }
-    let auth_val = record.auth_tag.clone();
-    match write_secret(
-        store,
-        |gen| agent_auth_tag_key(&pubkey, gen),
-        auth_val.as_deref(),
-        &format!("agent:{pubkey} auth_tag"),
-    ) {
+    match &outcomes[1] {
         WriteOutcome::Persisted { gen } => {
-            cancel_gc_candidacy(store, &agent_auth_tag_key(&pubkey, &gen));
             record.auth_tag = None;
-            record.auth_tag_ref = Some(gen);
+            record.auth_tag_ref = Some(gen.clone());
         }
         WriteOutcome::Nothing if unavailable => {}
         WriteOutcome::KeptInline { .. } | WriteOutcome::Nothing => {
             record.auth_tag_ref = None;
         }
     }
-    if let BackendKind::Provider {
-        ref id,
-        ref mut config,
-    } = record.backend
-    {
-        let serialized = if !config.is_null() {
-            serialize_provider_config(config).ok()
-        } else {
-            None
-        };
-        match write_secret(
-            store,
-            |gen| agent_provider_config_key(&pubkey, gen),
-            serialized.as_deref(),
-            &format!("agent:{pubkey} provider_config ({id})"),
-        ) {
+    if provider.is_some() {
+        match &outcomes[2] {
             WriteOutcome::Persisted { gen } => {
-                cancel_gc_candidacy(store, &agent_provider_config_key(&pubkey, &gen));
-                *config = serde_json::Value::Null;
-                record.provider_config_ref = Some(gen);
+                if let BackendKind::Provider { config, .. } = &mut record.backend {
+                    *config = serde_json::Value::Null;
+                }
+                record.provider_config_ref = Some(gen.clone());
             }
             WriteOutcome::Nothing if unavailable => {}
             WriteOutcome::KeptInline { .. } | WriteOutcome::Nothing => {
@@ -223,16 +253,19 @@ pub(crate) fn strip_and_persist_definition_secrets_with<S: ProjectionStore>(
     } else {
         None
     };
-    match write_secret(
-        store,
-        |gen| definition_env_key(&slug, gen),
-        inline_env.as_deref(),
-        &format!("definition:{slug} env_vars"),
-    ) {
+    let env_key = |gen: &str| definition_env_key(&slug, gen);
+    let env_ctx = format!("definition:{slug} env_vars");
+    let fields = [FieldSave {
+        coord_key_fn: &env_key,
+        value: inline_env.as_deref(),
+        existing_ref: record.env_vars_ref.as_deref(),
+        context: &env_ctx,
+    }];
+    let outcomes = write_secrets_batched(store, &fields);
+    match &outcomes[0] {
         WriteOutcome::Persisted { gen } => {
-            cancel_gc_candidacy(store, &definition_env_key(&slug, &gen));
             record.env_vars.clear();
-            record.env_vars_ref = Some(gen);
+            record.env_vars_ref = Some(gen.clone());
         }
         WriteOutcome::Nothing if unavailable => {}
         WriteOutcome::KeptInline { .. } | WriteOutcome::Nothing => {