jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 22:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

desktop: preserve PMA retry through promotion

Browse Source 
Co-authored-by: Wes <wesbillman@users.noreply.github.com>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>
This commit is contained in:
Wes
2026-08-05 11:58:01 -06:00
committed by Carl
parent fcdaf22232
commit 27f3143e61
2 changed files with 104 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
desktop/src-tauri/src/managed_agents/migration/driver/mod.rs
View File

@@ -4,10 +4,10 @@
//! [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.
//! only code that moves a retained request across the wire: it validates the
//! retained row against its exact JSON before egress, POSTs those bytes, and
//! returns verified evidence plus the immutable attempt identity. The caller
//! owns authority/cache persistence and only then compare-and-clears the row.
//!
//! Design invariants:
//! * **Exact bytes on the wire.** The body is `request_json` verbatim — the
@@ -19,24 +19,22 @@
//! * **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.
//! * **Caller-owned confirmation.** Successful verification returns the exact
//! retained attempt identity. This transport never clears `pending_sync`;
//! the caller first persists relay authority/cache and then compare-and-clears.
//! * **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.
//! and leaves `pending_sync = 1`.
use nostr::{Event, Keys};
use reqwest::Method;
use rusqlite::Connection;
use serde::Deserialize;
use super::{verify_promotion, AggregateResponse, MigrationCandidate};
use super::{verify_promotion, AggregateResponse, MigrationCandidate, PromotionEvidence};
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,
get_retained_managed_agent_aggregate, record_managed_agent_aggregate_error,
};
/// The route the relay mounts for atomic PMA aggregate commits. NIP-98 is
@@ -61,16 +59,27 @@ struct RetainedAggregateBody {
expected_definition_revision: Option<u64>,
}
/// Immutable identity of the retained attempt that produced verified evidence.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RetainedAttempt {
pub owner_pubkey: String,
pub agent_pubkey: String,
pub generation: u64,
pub private_event_id: String,
pub state: String,
}
/// 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 relay served back a faithful active aggregate. Persistence and
/// compare-and-clear remain the caller's responsibility.
Verified {
attempt: RetainedAttempt,
evidence: PromotionEvidence,
},
/// The attempt failed; a diagnostic was persisted and retry is preserved.
Retained { error: String },
@@ -78,11 +87,12 @@ pub(crate) enum SubmitOutcome {
/// 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.
/// Reads the durable row, validates it against its strict `request_json` before
/// any network I/O, POSTs those exact bytes with fresh owner-signed NIP-98 auth,
/// strict-deserializes and verifies the response, then returns evidence with the
/// exact attempt identity. Every failure persists an exact-attempt diagnostic;
/// success deliberately leaves retry pending for caller-owned persistence and
/// compare-and-clear ordering.
#[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,
@@ -116,17 +126,27 @@ pub(crate) async fn submit_retained_aggregate(
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),
};
if row.state != "active"
|| candidate.payload.state != buzz_core_pkg::private_managed_agent::State::Active
|| candidate.payload.generation != generation
|| candidate.signed_event.id.to_hex() != private_event_id
|| candidate.payload.owner_pubkey != owner_pubkey
|| candidate.payload.agent_pubkey != agent_pubkey
{
return record_failure(
"retained active request does not match its owner/agent/generation/event/state coordinate"
.to_string(),
);
}
let response = match submit_to_relay(state, owner_keys, row.request_json.as_bytes()).await {
Ok(response) => response,
Err(error) => return record_failure(error),
};
let evidence = match verify_promotion(&candidate, &response, owner_keys) {
Ok(evidence) => evidence,
@@ -144,22 +164,16 @@ pub(crate) async fn submit_retained_aggregate(
));
}
match mark_managed_agent_aggregate_synced(
conn,
owner_pubkey,
agent_pubkey,
generation,
&private_event_id,
) {
Ok(true) => Ok(SubmitOutcome::Synced {
Ok(SubmitOutcome::Verified {
attempt: RetainedAttempt {
owner_pubkey: row.owner_pubkey,
agent_pubkey: row.agent_pubkey,
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),
}
state: row.state,
},
evidence,
})
}
/// POST the exact request bytes with fresh owner-signed NIP-98 auth and
@@ -208,11 +222,25 @@ fn reconstruct_candidate(body: &[u8], owner_keys: &Keys) -> Result<MigrationCand
.instance_event
.ok_or_else(|| "retained request is missing its instance projection".to_string())?;
let expected_definition_revision = parsed.expected_definition_revision.ok_or_else(|| {
"retained active request is missing expected_definition_revision".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}"))?;
let payload_revision = payload
.active
.as_ref()
.ok_or_else(|| "retained active request head has no active payload".to_string())?
.definition
.revision;
if expected_definition_revision != payload_revision {
return Err(format!(
"retained expected definition revision {expected_definition_revision} does not match head {payload_revision}"
));
}
Ok(MigrationCandidate {
signed_event: parsed.private_event,

51
desktop/src-tauri/src/managed_agents/migration/driver/tests.rs
View File

@@ -3,8 +3,9 @@
//! 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.
//! The happy path proves a faithful read-back returns evidence while leaving
//! caller-owned retry state pending; every negative path proves the row stays
//! pending with a persisted diagnostic.
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
@@ -253,7 +254,7 @@ fn app_state_at(url: String) -> crate::app_state::AppState {
// ── Tests ─────────────────────────────────────────────────────────────────
#[tokio::test]
async fn faithful_read_back_marks_exact_generation_synced_and_posts_verbatim() {
async fn faithful_read_back_returns_evidence_leaves_pending_and_posts_verbatim() {
let fx = Fixture::new();
let dir = tempfile::tempdir().expect("tempdir");
let conn = seeded_conn(&dir, &fx.retained_row());
@@ -275,24 +276,39 @@ async fn faithful_read_back_marks_exact_generation_synced_and_posts_verbatim() {
.await
.expect("drive succeeds");
assert_eq!(
outcome,
SubmitOutcome::Synced {
generation: fx.cas.generation,
private_event_id: fx.candidate.signed_event.id.to_hex(),
match outcome {
SubmitOutcome::Verified { attempt, evidence } => {
assert_eq!(attempt.owner_pubkey, fx.owner_hex());
assert_eq!(attempt.agent_pubkey, fx.agent_hex());
assert_eq!(attempt.generation, fx.cas.generation);
assert_eq!(
attempt.private_event_id,
fx.candidate.signed_event.id.to_hex()
);
assert_eq!(attempt.state, "active");
assert_eq!(evidence.generation, fx.cas.generation);
assert_eq!(
evidence.head_event_id,
fx.candidate.signed_event.id.to_hex()
);
}
);
other => panic!("expected verified evidence, got {other:?}"),
}
// 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())
// Transport success is not permission to clear retry. The caller must first
// persist authority/cache, then compare-and-clear this exact attempt.
let retained = 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);
assert!(
retained.pending_sync,
"transport leaves confirmation pending"
);
assert_eq!(retained.last_error, None);
}
#[tokio::test]
@@ -393,7 +409,7 @@ async fn tampered_read_back_fails_verification_and_preserves_retry() {
}
#[tokio::test]
async fn row_coordinate_disagreeing_with_verified_head_is_not_synced() {
async fn row_coordinate_disagreeing_with_request_is_rejected_before_egress() {
// 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
@@ -404,7 +420,7 @@ async fn row_coordinate_disagreeing_with_verified_head_is_not_synced() {
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 {
let (url, captured) = spawn_stub(StubReply {
status: StatusCode::OK,
body: fx.faithful_response_json().to_string(),
})
@@ -421,9 +437,10 @@ async fn row_coordinate_disagreeing_with_verified_head_is_not_synced() {
.await
.expect("drive resolves");
assert!(matches!(outcome, SubmitOutcome::Retained { .. }));
assert!(
matches!(outcome, SubmitOutcome::Retained { .. }),
"row whose coordinate disagrees with the verified head is not synced"
captured.lock().unwrap().is_none(),
"coordinate drift must be rejected before network submission"
);
let stored = get_retained_managed_agent_aggregate(&conn, &fx.owner_hex(), &fx.agent_hex())
.unwrap()