diff --git a/desktop/src-tauri/src/managed_agents/migration/driver/mod.rs b/desktop/src-tauri/src/managed_agents/migration/driver/mod.rs new file mode 100644 index 0000000000..5c767d31a7 --- /dev/null +++ b/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, + #[serde(default)] + instance_event: Option, + #[serde(default)] + #[allow(dead_code)] // Bound into the head payload at build time; the relay re-derives it. + expected_definition_revision: Option, +} + +/// 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 { + 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 { + // 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 { + 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::(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 { + 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; diff --git a/desktop/src-tauri/src/managed_agents/migration/driver/tests.rs b/desktop/src-tauri/src/managed_agents/migration/driver/tests.rs new file mode 100644 index 0000000000..65a752e597 --- /dev/null +++ b/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, Arc>>>); + +/// 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>>>) { + let captured: Arc>>> = 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, + 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" + ); +} diff --git a/desktop/src-tauri/src/managed_agents/migration/mod.rs b/desktop/src-tauri/src/managed_agents/migration/mod.rs index da01a6b7ba..390a0d52b3 100644 --- a/desktop/src-tauri/src/managed_agents/migration/mod.rs +++ b/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;