desktop: PMA aggregate submit/retry driver

Add `managed_agents::migration::driver`, the transport half of the
relay-canonical migration seam. It is the only code that moves a retained
aggregate across the wire: it consumes one immutable row from
`managed_agent_aggregates`, POSTs the row's `request_json` verbatim to
`/api/managed-agents/aggregate` with fresh owner-signed NIP-98 auth,
strict-deserializes the corrected `migration::AggregateResponse`, verifies it
through `verify_promotion`, and flips the durable row to synced only after a
fully verified read-back.

Design invariants:
- Exact bytes on the wire. The body is `request_json` verbatim; the driver
  never re-serializes it, so the stored request and the request the relay
  authenticates are the same bytes. The verification source
  (`MigrationCandidate`) is reconstructed from those same bytes — parse the
  three signed events, decrypt the head under the owner key — so the candidate
  and the wire request cannot diverge and no build-side crypto is duplicated.
- Fresh owner NIP-98 per attempt via `build_nip98_auth_header_for_keys` (unique
  nonce), so a retry never replays a stale token. Mirrors the `/events`
  submit path: egress guard, rate-limit wait, `classify_request_error` /
  `relay_error_message` / `parse_json_response`.
- Only the exact retained (generation, private_event_id) is marked synced.
  Beyond `verify_promotion`'s response-vs-candidate checks, the driver binds
  the *durable row's* coordinate to the verified evidence, so a row whose
  columns disagree with its own request_json (and thus the verified head) is
  never confirmed.
- Errors preserve retry. Every failure path — transport, non-2xx, malformed
  body, reconstruction, verification, coordinate mismatch — persists an
  exact-attempt diagnostic via `record_managed_agent_aggregate_error` and leaves
  `pending_sync = 1`. Nothing here clears retry except a verified read-back.

Transport tests use a loopback axum stub in the shape of the aggregate route
(6, all green): faithful read-back marks the exact generation synced and posts
verbatim; conflict status preserves retry; malformed success body preserves
retry; tampered read-back fails verification and preserves retry; row
coordinate disagreeing with the verified head is not synced; no pending row is
a no-op with no network submission.

Pure transport confined to the new module + one `mod` line; no delete/tombstone,
docs, or relay-crate changes. `managed_agents::` suite: 1026 passed, 0 failed;
`cargo check --lib` clean.

Co-authored-by: Mongo <5c25403eab7271f9f94ddd4f2b270e8cac2c92e2c830c51877cca6ec974ffb3f@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>

desktop/src-tauri/src/managed_agents/migration/driver/mod.rs

@@ -0,0 +1,226 @@
+//! Desktop PMA aggregate submit/retry driver.
+//!
+//! This is the transport half of the relay-canonical migration seam. The pure
+//! [builder/verifier](super) turns a hydrated record into a signed candidate
+//! and gates the relay read-back; the durable [retention](crate::managed_agents::retention)
+//! layer persists one immutable request per CAS generation. This driver is the
+//! only code that moves a retained request across the wire: it POSTs the exact
+//! retained JSON, verifies the response through [`super::verify_promotion`], and
+//! flips the durable row to synced only when the relay served back a faithful,
+//! owner-signed copy.
+//!
+//! Design invariants:
+//!   * **Exact bytes on the wire.** The body is `request_json` verbatim — the
+//!     driver never re-serializes it, so a stored request and the request the
+//!     relay authenticates are the same bytes. The verification source
+//!     ([`MigrationCandidate`](super::MigrationCandidate)) is reconstructed from
+//!     those same bytes (parse the events, decrypt the head under the owner
+//!     key), so there is no candidate-vs-wire drift and no crypto duplication.
+//!   * **Fresh owner NIP-98 per attempt.** Each submission mints a new
+//!     [`build_nip98_auth_header_for_keys`](crate::relay::build_nip98_auth_header_for_keys)
+//!     header (unique nonce), so a retry never replays a stale token.
+//!   * **Only the exact generation/event is marked synced.** Confirmation binds
+//!     to `(generation, private_event_id)`; a relay that echoes a different
+//!     generation or head id is a verification failure, not a sync.
+//!   * **Errors preserve retry.** Every failure path persists a diagnostic via
+//!     [`record_managed_agent_aggregate_error`](crate::managed_agents::retention::record_managed_agent_aggregate_error)
+//!     and leaves `pending_sync = 1`; nothing here clears retry except a fully
+//!     verified read-back.
+
+use nostr::{Event, Keys};
+use reqwest::Method;
+use rusqlite::Connection;
+use serde::Deserialize;
+
+use super::{verify_promotion, AggregateResponse, MigrationCandidate};
+use crate::app_state::AppState;
+use crate::managed_agents::retention::{
+    get_retained_managed_agent_aggregate, mark_managed_agent_aggregate_synced,
+    record_managed_agent_aggregate_error,
+};
+
+/// The route the relay mounts for atomic PMA aggregate commits. NIP-98 is
+/// signed against this exact path (see `buzz-relay` router).
+const AGGREGATE_PATH: &str = "/api/managed-agents/aggregate";
+
+/// The wire body persisted as `request_json` and POSTed verbatim.
+///
+/// Mirrors the relay's `AggregateBody`; used here only to reconstruct the
+/// verification candidate from the retained bytes, never to re-serialize the
+/// request.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RetainedAggregateBody {
+    private_event: Event,
+    #[serde(default)]
+    definition_event: Option<Event>,
+    #[serde(default)]
+    instance_event: Option<Event>,
+    #[serde(default)]
+    #[allow(dead_code)] // Bound into the head payload at build time; the relay re-derives it.
+    expected_definition_revision: Option<u64>,
+}
+
+/// What one drive of a retained aggregate resolved to.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(dead_code)] // Consumed by the boot/reconcile lane (sibling); today only tests read it.
+pub(crate) enum SubmitOutcome {
+    /// No pending aggregate row for this agent — nothing to submit.
+    Nothing,
+    /// The relay served back a faithful copy and the row was marked synced.
+    Synced {
+        generation: u64,
+        private_event_id: String,
+    },
+    /// The attempt failed; a diagnostic was persisted and retry is preserved.
+    Retained { error: String },
+}
+
+/// Submit (or retry) the latest retained aggregate for one owner→agent pair.
+///
+/// Reads the durable row, POSTs its exact `request_json` with fresh owner-signed
+/// NIP-98 auth, strict-deserializes the response, verifies it against the
+/// candidate reconstructed from the same bytes, and — only on a fully verified
+/// read-back — marks that exact generation/event synced. Every failure persists
+/// an exact-attempt diagnostic without clearing retry.
+#[allow(dead_code)] // Consumed by the boot/reconcile lane (sibling); exercised by this module's tests.
+pub(crate) async fn submit_retained_aggregate(
+    state: &AppState,
+    conn: &Connection,
+    owner_keys: &Keys,
+    owner_pubkey: &str,
+    agent_pubkey: &str,
+) -> Result<SubmitOutcome, String> {
+    let row = match get_retained_managed_agent_aggregate(conn, owner_pubkey, agent_pubkey)? {
+        Some(row) if row.pending_sync => row,
+        _ => return Ok(SubmitOutcome::Nothing),
+    };
+
+    // Fail-fast guard: the retained head id/generation is what we will confirm.
+    // Capture them before any network work so a diagnostic always names the
+    // exact attempt.
+    let generation = row.generation;
+    let private_event_id = row.private_event_id.clone();
+
+    let record_failure = |error: String| -> Result<SubmitOutcome, String> {
+        // A best-effort diagnostic write must not mask the real error; if the
+        // update itself fails we still surface the original failure.
+        let _ = record_managed_agent_aggregate_error(
+            conn,
+            owner_pubkey,
+            agent_pubkey,
+            generation,
+            &private_event_id,
+            &error,
+        );
+        Ok(SubmitOutcome::Retained { error })
+    };
+
+    let response = match submit_to_relay(state, owner_keys, row.request_json.as_bytes()).await {
+        Ok(response) => response,
+        Err(error) => return record_failure(error),
+    };
+
+    // Reconstruct the verification source from the *same bytes* we sent, so the
+    // candidate and the wire request can never diverge.
+    let candidate = match reconstruct_candidate(row.request_json.as_bytes(), owner_keys) {
+        Ok(candidate) => candidate,
+        Err(error) => return record_failure(error),
+    };
+
+    let evidence = match verify_promotion(&candidate, &response, owner_keys) {
+        Ok(evidence) => evidence,
+        Err(error) => return record_failure(format!("verification failed: {error:?}")),
+    };
+
+    // The relay's committed head must be the exact generation/event we retained;
+    // confirming a different coordinate would desync durable state from the
+    // relay. verify_promotion already proved the head is a faithful copy, so a
+    // mismatch here is a protocol violation, not a sync.
+    if evidence.generation != generation || evidence.head_event_id != private_event_id {
+        return record_failure(format!(
+            "read-back coordinate {}:{} does not match retained {generation}:{private_event_id}",
+            evidence.generation, evidence.head_event_id
+        ));
+    }
+
+    match mark_managed_agent_aggregate_synced(
+        conn,
+        owner_pubkey,
+        agent_pubkey,
+        generation,
+        &private_event_id,
+    ) {
+        Ok(true) => Ok(SubmitOutcome::Synced {
+            generation,
+            private_event_id,
+        }),
+        // The row moved under us (superseded by a newer generation) between read
+        // and confirm. Nothing verified was lost; the newer row drives next.
+        Ok(false) => Ok(SubmitOutcome::Nothing),
+        Err(error) => Err(error),
+    }
+}
+
+/// POST the exact request bytes with fresh owner-signed NIP-98 auth and
+/// strict-deserialize the aggregate response.
+async fn submit_to_relay(
+    state: &AppState,
+    owner_keys: &Keys,
+    body: &[u8],
+) -> Result<AggregateResponse, String> {
+    crate::egress_guard::assert_no_key_backup_bytes(body, "managed-agent aggregate submit")?;
+    crate::relay_admission::wait_for_rate_limit().await;
+
+    let base = crate::relay::relay_api_base_url_with_override(state);
+    let url = format!("{}{AGGREGATE_PATH}", base.trim_end_matches('/'));
+    let auth_header =
+        crate::relay::build_nip98_auth_header_for_keys(owner_keys, &Method::POST, &url, body)?;
+
+    let response = state
+        .http_client
+        .post(&url)
+        .header("Authorization", auth_header)
+        .header("Content-Type", "application/json")
+        .body(body.to_vec())
+        .send()
+        .await
+        .map_err(|e| crate::relay::classify_request_error(&e))?;
+
+    if !response.status().is_success() {
+        return Err(crate::relay::relay_error_message(response).await);
+    }
+
+    crate::relay::parse_json_response::<AggregateResponse>(response).await
+}
+
+/// Rebuild the verification [`MigrationCandidate`] from the retained request
+/// bytes: parse the three signed events and decrypt the head under the owner
+/// key. The result is the exact source the relay was asked to store, so
+/// verification compares like against like.
+fn reconstruct_candidate(body: &[u8], owner_keys: &Keys) -> Result<MigrationCandidate, String> {
+    let parsed: RetainedAggregateBody = serde_json::from_slice(body)
+        .map_err(|e| format!("retained request json is not a valid aggregate body: {e}"))?;
+    let definition_event = parsed
+        .definition_event
+        .ok_or_else(|| "retained request is missing its definition projection".to_string())?;
+    let instance_event = parsed
+        .instance_event
+        .ok_or_else(|| "retained request is missing its instance projection".to_string())?;
+
+    let (_, payload) = buzz_core_pkg::private_managed_agent::validate_and_decrypt(
+        &parsed.private_event,
+        owner_keys,
+    )
+    .map_err(|e| format!("retained head does not decrypt under the owner key: {e}"))?;
+
+    Ok(MigrationCandidate {
+        signed_event: parsed.private_event,
+        payload,
+        definition_event,
+        instance_event,
+    })
+}
+
+#[cfg(test)]
+mod tests;

desktop/src-tauri/src/managed_agents/migration/driver/tests.rs

@@ -0,0 +1,463 @@
+//! Transport tests for the PMA aggregate submit/retry driver.
+//!
+//! Each test builds a real signed candidate, serializes it into the exact
+//! `request_json` the retention layer would persist, stands up a loopback axum
+//! stub in the shape of the relay's aggregate route, and drives one submission.
+//! The happy path proves a faithful read-back marks the row synced; every
+//! negative path proves the row stays pending with a persisted diagnostic.
+
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+
+use axum::{extract::State as AxumState, http::StatusCode, routing::post, Router};
+use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA};
+use nostr::{EventBuilder, Keys, Kind};
+use serde_json::json;
+
+use super::super::{build_migration_candidate, CasMetadata, MigrationCandidate};
+use super::*;
+use crate::managed_agents::retention::{
+    get_retained_managed_agent_aggregate, open_retention_db, retain_managed_agent_aggregate,
+    RetainedManagedAgentAggregate,
+};
+use crate::managed_agents::types::{BackendKind, ManagedAgentRecord, RespondTo};
+
+const CAPABLE: [&str; 1] = [crate::managed_agents::authority::NIP_PMA_AGGREGATE_TOKEN];
+const CREATED_AT: u64 = 1_700_000_000;
+
+// ── Fixtures ────────────────────────────────────────────────────────────────
+
+fn signed_projection(owner_keys: &Keys, kind: u32, d: &str, content: &str) -> nostr::Event {
+    EventBuilder::new(Kind::Custom(kind as u16), content)
+        .tags([nostr::Tag::parse(["d", d]).unwrap()])
+        .custom_created_at(nostr::Timestamp::from(CREATED_AT))
+        .sign_with_keys(owner_keys)
+        .expect("sign projection")
+}
+
+fn record_for(agent_keys: &Keys) -> ManagedAgentRecord {
+    ManagedAgentRecord {
+        pubkey: agent_keys.public_key().to_hex(),
+        name: "Test Agent".to_string(),
+        persona_id: Some("persona-1".to_string()),
+        team_id: Some("team-1".to_string()),
+        private_key_nsec: String::new(),
+        auth_tag: Some("[\"auth\",\"STALE\",\"\",\"deadbeef\"]".to_string()),
+        relay_url: "wss://relay.example".to_string(),
+        avatar_url: None,
+        acp_command: "buzz-acp".to_string(),
+        agent_command: "goose".to_string(),
+        agent_command_override: None,
+        agent_args: vec![],
+        mcp_command: "buzz-dev-mcp".to_string(),
+        turn_timeout_seconds: 0,
+        idle_timeout_seconds: None,
+        max_turn_duration_seconds: None,
+        parallelism: 1,
+        system_prompt: Some("You are a test agent.".to_string()),
+        model: Some("claude-opus-4".to_string()),
+        provider: Some("anthropic".to_string()),
+        persona_source_version: None,
+        env_vars: BTreeMap::new(),
+        start_on_app_launch: false,
+        auto_restart_on_config_change: true,
+        runtime_pid: None,
+        backend: BackendKind::Local,
+        backend_agent_id: None,
+        provider_binary_path: None,
+        persona_team_dir: None,
+        persona_name_in_team: None,
+        created_at: "2025-01-01T00:00:00Z".to_string(),
+        updated_at: "2025-01-01T00:00:00Z".to_string(),
+        last_started_at: None,
+        last_stopped_at: None,
+        last_exit_code: None,
+        last_error: None,
+        last_error_code: None,
+        respond_to: RespondTo::default(),
+        respond_to_allowlist: vec![],
+        display_name: None,
+        slug: Some("sample-slug".to_string()),
+        runtime: Some("goose".to_string()),
+        name_pool: vec![],
+        is_builtin: false,
+        is_active: true,
+        shared: false,
+        source_team: None,
+        source_team_persona_slug: None,
+        catalog_source: None,
+        definition_respond_to: None,
+        definition_respond_to_allowlist: vec![],
+        definition_parallelism: None,
+        relay_mesh: None,
+        relay_authority: crate::managed_agents::authority::RelayAuthority::legacy(),
+    }
+}
+
+struct Fixture {
+    owner_keys: Keys,
+    agent_keys: Keys,
+    candidate: MigrationCandidate,
+    cas: CasMetadata,
+}
+
+impl Fixture {
+    fn new() -> Self {
+        let owner_keys = Keys::generate();
+        let agent_keys = Keys::generate();
+        let agent_hex = agent_keys.public_key().to_hex();
+        let definition_event = signed_projection(
+            &owner_keys,
+            KIND_PERSONA,
+            "def-slug",
+            "{\"definition\":true}",
+        );
+        let instance_event = signed_projection(
+            &owner_keys,
+            KIND_MANAGED_AGENT,
+            &agent_hex,
+            "{\"instance\":true}",
+        );
+        let cas = CasMetadata {
+            generation: 1,
+            previous_event_id: None,
+            definition_revision: 7,
+        };
+        let candidate = build_migration_candidate(
+            &record_for(&agent_keys),
+            &owner_keys,
+            &agent_keys,
+            definition_event,
+            instance_event,
+            &cas,
+            CAPABLE,
+            CREATED_AT,
+        )
+        .expect("candidate builds");
+        Self {
+            owner_keys,
+            agent_keys,
+            candidate,
+            cas,
+        }
+    }
+
+    fn owner_hex(&self) -> String {
+        self.owner_keys.public_key().to_hex()
+    }
+
+    fn agent_hex(&self) -> String {
+        self.agent_keys.public_key().to_hex()
+    }
+
+    /// The exact `request_json` the retention layer would persist and the
+    /// driver POSTs verbatim: the relay's `AggregateBody` shape.
+    fn request_json(&self) -> String {
+        json!({
+            "private_event": self.candidate.signed_event,
+            "definition_event": self.candidate.definition_event,
+            "instance_event": self.candidate.instance_event,
+            "expected_definition_revision": self.cas.definition_revision,
+        })
+        .to_string()
+    }
+
+    /// A faithful relay read-back — the relay stored and serves back exactly
+    /// what we submitted.
+    fn faithful_response_json(&self) -> serde_json::Value {
+        json!({
+            "event_id": self.candidate.signed_event.id.to_hex(),
+            "generation": self.cas.generation,
+            "state": "active",
+            "accepted": true,
+            "inserted": true,
+            "private_event": self.candidate.signed_event,
+            "definition_event": self.candidate.definition_event,
+            "instance_event": self.candidate.instance_event,
+            "definition_revision": self.cas.definition_revision,
+        })
+    }
+
+    fn retained_row(&self) -> RetainedManagedAgentAggregate {
+        RetainedManagedAgentAggregate {
+            owner_pubkey: self.owner_hex(),
+            agent_pubkey: self.agent_hex(),
+            generation: self.cas.generation,
+            private_event_id: self.candidate.signed_event.id.to_hex(),
+            state: "active".to_string(),
+            request_json: self.request_json(),
+            pending_sync: true,
+            last_error: None,
+        }
+    }
+}
+
+// ── Stub relay ──────────────────────────────────────────────────────────────
+
+/// A scripted reply: the HTTP status and body the stub returns.
+#[derive(Clone)]
+struct StubReply {
+    status: StatusCode,
+    body: String,
+}
+
+type StubState = (Arc<StubReply>, Arc<Mutex<Option<Vec<u8>>>>);
+
+/// Spawn a loopback stub relay serving the aggregate route with a scripted
+/// reply, capturing the raw request body it received. Returns the base URL and
+/// a handle to the captured body.
+async fn spawn_stub(reply: StubReply) -> (String, Arc<Mutex<Option<Vec<u8>>>>) {
+    let captured: Arc<Mutex<Option<Vec<u8>>>> = Arc::new(Mutex::new(None));
+    let state: StubState = (Arc::new(reply), captured.clone());
+    let app =
+        Router::new()
+            .route(
+                "/api/managed-agents/aggregate",
+                post(
+                    |AxumState((reply, captured)): AxumState<StubState>,
+                     body: axum::body::Bytes| async move {
+                        *captured.lock().unwrap() = Some(body.to_vec());
+                        (reply.status, reply.body.clone())
+                    },
+                ),
+            )
+            .with_state(state);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind stub relay");
+    let addr = listener.local_addr().expect("stub relay addr");
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.ok();
+    });
+    (format!("http://{addr}"), captured)
+}
+
+/// Open the retention db, seed one pending row, and hand back a fresh
+/// connection for the driver to read/write committed state through.
+fn seeded_conn(
+    dir: &tempfile::TempDir,
+    row: &RetainedManagedAgentAggregate,
+) -> rusqlite::Connection {
+    let path = dir.path().join("retention.db");
+    let mut writer = open_retention_db(&path).expect("open db");
+    retain_managed_agent_aggregate(&mut writer, row).expect("retain aggregate row");
+    open_retention_db(&path).expect("reopen db")
+}
+
+fn app_state_at(url: String) -> crate::app_state::AppState {
+    let state = crate::app_state::build_app_state();
+    *state.relay_url_override.lock().unwrap() = Some(url);
+    state
+}
+
+// ── Tests ─────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn faithful_read_back_marks_exact_generation_synced_and_posts_verbatim() {
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let conn = seeded_conn(&dir, &fx.retained_row());
+
+    let (url, captured) = spawn_stub(StubReply {
+        status: StatusCode::OK,
+        body: fx.faithful_response_json().to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive succeeds");
+
+    assert_eq!(
+        outcome,
+        SubmitOutcome::Synced {
+            generation: fx.cas.generation,
+            private_event_id: fx.candidate.signed_event.id.to_hex(),
+        }
+    );
+
+    // Posted bytes are the retained request verbatim.
+    let posted = captured.lock().unwrap().clone().expect("body captured");
+    assert_eq!(posted, fx.request_json().into_bytes());
+
+    // The row is no longer pending and carries no error.
+    let confirmed = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
+        .unwrap()
+        .unwrap();
+    assert!(!confirmed.pending_sync, "verified row is marked synced");
+    assert_eq!(confirmed.last_error, None);
+}
+
+#[tokio::test]
+async fn conflict_status_preserves_retry_with_diagnostic() {
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let conn = seeded_conn(&dir, &fx.retained_row());
+
+    let (url, _captured) = spawn_stub(StubReply {
+        status: StatusCode::CONFLICT,
+        body: json!({ "error": "stale generation" }).to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive resolves");
+
+    assert!(matches!(outcome, SubmitOutcome::Retained { .. }));
+    let row = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
+        .unwrap()
+        .unwrap();
+    assert!(row.pending_sync, "rejected attempt stays pending");
+    assert!(row.last_error.is_some(), "diagnostic persisted");
+}
+
+#[tokio::test]
+async fn malformed_success_body_preserves_retry() {
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let conn = seeded_conn(&dir, &fx.retained_row());
+
+    let (url, _captured) = spawn_stub(StubReply {
+        status: StatusCode::OK,
+        body: "{ not json".to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive resolves");
+
+    assert!(matches!(outcome, SubmitOutcome::Retained { .. }));
+    let row = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
+        .unwrap()
+        .unwrap();
+    assert!(row.pending_sync);
+    assert!(row.last_error.is_some());
+}
+
+#[tokio::test]
+async fn tampered_read_back_fails_verification_and_preserves_retry() {
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let conn = seeded_conn(&dir, &fx.retained_row());
+
+    // A well-formed response whose head id was swapped for a different value —
+    // deserializes cleanly but must fail verify_promotion.
+    let mut tampered = fx.faithful_response_json();
+    tampered["event_id"] = json!("0".repeat(64));
+    let (url, _captured) = spawn_stub(StubReply {
+        status: StatusCode::OK,
+        body: tampered.to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive resolves");
+
+    assert!(matches!(outcome, SubmitOutcome::Retained { .. }));
+    let row = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
+        .unwrap()
+        .unwrap();
+    assert!(row.pending_sync, "unverified read-back never clears retry");
+    assert!(row.last_error.is_some());
+}
+
+#[tokio::test]
+async fn row_coordinate_disagreeing_with_verified_head_is_not_synced() {
+    // The durable row's (generation, private_event_id) columns are the pair the
+    // driver confirms. If they disagree with the request_json the row carries —
+    // and thus with the verified read-back — the guard must refuse to mark it
+    // synced even though the relay served a faithful copy of the request bytes.
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let mut row = fx.retained_row();
+    row.private_event_id = "f".repeat(64); // column diverges from request_json head id
+    let conn = seeded_conn(&dir, &row);
+
+    let (url, _captured) = spawn_stub(StubReply {
+        status: StatusCode::OK,
+        body: fx.faithful_response_json().to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive resolves");
+
+    assert!(
+        matches!(outcome, SubmitOutcome::Retained { .. }),
+        "row whose coordinate disagrees with the verified head is not synced"
+    );
+    let stored = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
+        .unwrap()
+        .unwrap();
+    assert!(stored.pending_sync, "mismatched row stays pending");
+    assert!(stored.last_error.is_some(), "diagnostic persisted");
+}
+
+#[tokio::test]
+async fn no_pending_row_is_a_noop() {
+    let fx = Fixture::new();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let conn = open_retention_db(&dir.path().join("retention.db")).expect("open db");
+
+    let (url, captured) = spawn_stub(StubReply {
+        status: StatusCode::OK,
+        body: fx.faithful_response_json().to_string(),
+    })
+    .await;
+    let state = app_state_at(url);
+
+    let outcome = submit_retained_aggregate(
+        &state,
+        &conn,
+        &fx.owner_keys,
+        &fx.owner_hex(),
+        &fx.agent_hex(),
+    )
+    .await
+    .expect("drive resolves");
+
+    assert_eq!(outcome, SubmitOutcome::Nothing);
+    assert!(
+        captured.lock().unwrap().is_none(),
+        "no row means no network submission"
+    );
+}

desktop/src-tauri/src/managed_agents/migration/mod.rs

@@ -707,5 +707,10 @@ fn require_owner_projection(
     }
 }
 
+/// The transport half of this seam: submits retained aggregates over HTTP and
+/// verifies the read-back through [`verify_promotion`]. Kept in a sibling
+/// module so the pure builder/verifier above stays free of HTTP/persistence.
+pub(crate) mod driver;
+
 #[cfg(test)]
 mod tests;