fix(hosts): recover stale Start receipts and protect private reads

Signed-off-by: Logan Johnson <loganj@squareup.com>
(cherry picked from commit c0ce9204d3e400dd936b62d649e32748fe464ad7)
Signed-off-by: Logan Johnson <loganj@squareup.com>

desktop/src-tauri/src/app_state.rs

@@ -155,12 +155,12 @@ fn identity_from_env() -> Option<Keys> {
 }
 
 /// Build the no-redirect HTTP client used for authenticated relay media
-/// fetches (download / copy).
+/// fetches (download / copy) and owner-private host transport.
 ///
 /// This client is a security boundary, not a convenience: it carries a minted
 /// media `Authorization` header, so it MUST NOT follow redirects. A relay 3xx
-/// to an off-origin or private host would otherwise forward that header across
-/// origins (a redirect-hop SSRF). `redirect::Policy::none()` returns the 3xx
+/// must not forward bodies or make redirect-hop requests, regardless of header
+/// stripping. `redirect::Policy::none()` returns the 3xx
 /// verbatim so the caller can reject it.
 ///
 /// Returned as a `Result` so the fail-closed invariant is testable — callers

desktop/src-tauri/src/commands/host_execution.rs

@@ -88,7 +88,7 @@ pub async fn execute_host_command(
     let catalog = super::discover_acp_providers(app.clone(), Some(true))
         .await
         .unwrap_or_default();
-    let registrations = crate::relay::query_relay_at_with_keys(&state,
+    let registrations = crate::relay::query_private_host_at_with_keys(&state,
         &crate::relay::relay_http_base_url(&relay),
         &[serde_json::json!({"kinds": [50000], "ids": [registration_id], "#p": [expected_owner], "limit": 2})],
         &owner, None).await.map_err(|_| "cannot revalidate execution registration")?;

desktop/src-tauri/src/commands/host_start.rs

@@ -1,7 +1,7 @@
 //! Owner-operated Desktop Start transport. No host-key login and no source
 //! identity/config/file transfer. Command/receipt payloads are encrypted; routing
 //! envelopes and local retry bookkeeping contain no launch secrets.
-use super::host_start_store::{retry_pending, Pending, Store};
+use super::host_start_store::{prepare_receipt, retry_pending, Pending, Store};
 use crate::app_state::AppState;
 use buzz_core_pkg::{
     host,
@@ -28,7 +28,7 @@ pub(super) async fn current_registration(
     relay: &str,
     id: &str,
 ) -> Result<Event, String> {
-    let events = crate::relay::query_relay_at_with_keys(state, &crate::relay::relay_http_base_url(relay),
+    let events = crate::relay::query_private_host_at_with_keys(state, &crate::relay::relay_http_base_url(relay),
         &[serde_json::json!({"kinds":[50000], "ids":[id], "#p":[owner.public_key().to_hex()], "limit":2})], owner, None)
         .await.map_err(|_| "cannot verify destination registration")?;
     let [reg] = events.as_slice() else {
@@ -170,7 +170,7 @@ pub(super) async fn history(
 ) -> Result<Vec<Event>, String> {
     let mut result = Vec::new();
     loop {
-        let page = crate::relay::query_relay_at_with_keys(
+        let page = crate::relay::query_private_host_at_with_keys(
             state,
             &crate::relay::relay_http_base_url(relay),
             &[filter.clone()],
@@ -229,7 +229,7 @@ async fn publish(
     let response = state
         .media_fetch_client
         .post(url)
-        .timeout(std::time::Duration::from_secs(15))
+        .timeout(crate::relay::PRIVATE_HOST_REQUEST_TIMEOUT)
         .header("Authorization", auth)
         .header("Content-Type", "application/json")
         .body(bytes)
@@ -367,12 +367,37 @@ pub async fn pump_host_start(
         );
         store.save()?;
     }
+    // The durable publication owner fsyncs each prepared envelope before send.
     // One revoked/unpublishable operation must not starve another placement.
-    retry_pending(&mut store.journal.received, true, |pending| {
-        publish_pending(&state, &owner, &relay, pending, true)
-    })
-    .await;
-    store.save()?;
+    store
+        .retry_receipts(
+            |mut pending| {
+                let (state, owner, host_keys, relay, receipts, expected_owner) = (
+                    &state,
+                    &owner,
+                    &host_keys,
+                    &relay,
+                    &receipts,
+                    &expected_owner,
+                );
+                async move {
+                    current_registration(state, owner, relay, &pending.registration.id.to_hex())
+                        .await?;
+                    scope(state, expected_owner, relay)?;
+                    prepare_receipt(
+                        &mut pending,
+                        owner,
+                        host_keys,
+                        relay,
+                        receipts,
+                        Timestamp::now().as_secs(),
+                    )?;
+                    Ok(pending)
+                }
+            },
+            |pending| publish_pending(&state, &owner, &relay, pending, true),
+        )
+        .await?;
     retry_pending(&mut store.journal.sent, false, |pending| {
         publish_pending(&state, &owner, &relay, pending, false)
     })
@@ -638,6 +663,7 @@ async fn publish_pending(
         &pending.command
     };
     current_registration(state, owner, relay, &pending.registration.id.to_hex()).await?;
+    scope(state, &owner.public_key().to_hex(), relay)?;
     publish(state, owner, relay, &pending.registration, event).await
 }
 

desktop/src-tauri/src/commands/host_start_recovery_tests.rs

@@ -0,0 +1,428 @@
+//! Durable outbox/restart tests with a controllable relay admission clock.
+//! The relay predicate mirrors ingest's +/-900 seconds; no DB/service is used.
+use super::*;
+use buzz_core_pkg::{
+    host,
+    host_execution::{self, Action, Command, Outcome, Receipt},
+};
+use nostr::{JsonUtil, Keys};
+
+fn observed(owner: &Keys, executor: &Keys) -> (Pending, Receipt) {
+    let registration = host::registration(owner, executor.public_key(), 99).unwrap();
+    let request = Command {
+        v: 1,
+        operation: "ab".repeat(16),
+        relay: "wss://relay.example".into(),
+        agent: Keys::generate().public_key().to_hex(),
+        expires_at: 400,
+        action: Action::Start {
+            runtime: "buzz-agent".into(),
+            revision: "cd".repeat(32),
+        },
+    };
+    let command = host_execution::command(owner, &registration, &request, 100).unwrap();
+    let result = Receipt {
+        v: 1,
+        command: command.id.to_hex(),
+        run: request.run().into(),
+        request,
+        observed_at: 101,
+        outcome: Outcome::Spawned,
+    };
+    let receipt = host_execution::receipt(executor, &registration, &result, 101).unwrap();
+    (
+        Pending {
+            registration,
+            command,
+            receipt: Some(receipt),
+            published: false,
+            error: None,
+            supersedes: None,
+        },
+        result,
+    )
+}
+
+// Exact timestamp condition of handlers/ingest.rs, plus real signature/routing
+// checks. Kept here rather than altering relay admission to accommodate retries.
+fn admit(p: &Pending, owner: &Keys, now: u64) -> Result<(), String> {
+    let event = p.receipt.as_ref().ok_or("missing")?;
+    host_execution::validate_transport(event, &p.registration, owner.public_key())?;
+    if now.abs_diff(event.created_at.as_secs()) > 900 {
+        return Err("timestamp drift".into());
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn late_never_accepted_receipt_survives_restart_without_new_execution() {
+    let dir = tempfile::tempdir().unwrap();
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (pending, observation) = observed(&owner, &executor);
+    let original = pending.receipt.clone().unwrap();
+    let command = pending.command.as_json();
+    let id = pending.command.id.to_hex();
+    let relay = observation.request.relay.clone();
+    let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap();
+    store.journal.received.insert(id.clone(), pending);
+    store.save().unwrap();
+    retry_pending(&mut store.journal.received, true, |_| async {
+        Err("never reached relay".into())
+    })
+    .await;
+    store.save().unwrap();
+    drop(store);
+
+    // More than the admission window AND the command lifetime, after restart.
+    let now = 101 + 901;
+    let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap();
+    assert_eq!(
+        admit(&store.journal.received[&id], &owner, now),
+        Err("timestamp drift".into())
+    );
+    assert!(
+        host_execution::decrypt_command(
+            &executor,
+            &store.journal.received[&id].registration,
+            &store.journal.received[&id].command,
+            &relay,
+            now
+        )
+        .is_err(),
+        "expired intent cannot execute"
+    );
+    prepare_receipt(
+        store.journal.received.get_mut(&id).unwrap(),
+        &owner,
+        &executor,
+        &relay,
+        &[],
+        now,
+    )
+    .unwrap();
+    store.save().unwrap();
+    let renewed = store.journal.received[&id].receipt.clone().unwrap();
+    assert_ne!(renewed.id, original.id);
+    assert_eq!(renewed.created_at.as_secs(), now);
+    assert_eq!(
+        renewed.content, original.content,
+        "same encrypted proof, no new observation"
+    );
+    assert_eq!(renewed.tags, original.tags);
+    assert_eq!(store.journal.received[&id].command.as_json(), command);
+    drop(store); // crash after preparing, before send
+
+    let mut store = Store::open_dir(dir.path(), "owner", &relay).unwrap();
+    let pending = store.journal.received.get_mut(&id).unwrap();
+    prepare_receipt(pending, &owner, &executor, &relay, &[], now + 1).unwrap();
+    assert_eq!(
+        pending.receipt.as_ref().unwrap().as_json(),
+        renewed.as_json()
+    );
+    retry_pending(&mut store.journal.received, true, |p| {
+        let owner = &owner;
+        async move { admit(&p, owner, now + 1) }
+    })
+    .await;
+    assert!(store.journal.received[&id].published);
+    let pending = &store.journal.received[&id];
+    // What the source learns from accepted history is the original exact result.
+    assert_eq!(
+        host_execution::decrypt_receipt(
+            &owner,
+            &pending.registration,
+            &renewed,
+            &pending.command,
+            &observation.request
+        )
+        .unwrap(),
+        observation
+    );
+    store.save().unwrap();
+    drop(store);
+    assert!(
+        Store::open_dir(dir.path(), "owner", &relay)
+            .unwrap()
+            .journal
+            .received[&id]
+            .published
+    );
+}
+
+#[tokio::test]
+async fn accepted_but_ack_lost_uses_history_without_renewing_proof() {
+    let dir = tempfile::tempdir().unwrap();
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (pending, observation) = observed(&owner, &executor);
+    let original = pending.receipt.clone().unwrap();
+    let id = pending.command.id.to_hex();
+    let relay = &observation.request.relay;
+    let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap();
+    store.journal.received.insert(id.clone(), pending);
+    store.save().unwrap();
+    store
+        .retry_receipts(
+            |p| std::future::ready(Ok(p)),
+            |p| {
+                let owner = &owner;
+                async move {
+                    admit(&p, owner, 101)?;
+                    Err("accepted, but ACK lost".into())
+                }
+            },
+        )
+        .await
+        .unwrap();
+    drop(store);
+    let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap();
+    store
+        .retry_receipts(
+            |mut p| {
+                let result = prepare_receipt(
+                    &mut p,
+                    &owner,
+                    &executor,
+                    relay,
+                    std::slice::from_ref(&original),
+                    2000,
+                )
+                .map(|()| p);
+                std::future::ready(result)
+            },
+            |_| async { panic!("accepted history must suppress another publication") },
+        )
+        .await
+        .unwrap();
+    assert!(store.journal.received[&id].published);
+    assert_eq!(
+        store.journal.received[&id]
+            .receipt
+            .as_ref()
+            .unwrap()
+            .as_json(),
+        original.as_json()
+    );
+}
+
+#[test]
+fn history_of_an_earlier_envelope_also_resolves_ambiguous_delivery() {
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (mut pending, observation) = observed(&owner, &executor);
+    let original = pending.receipt.clone().unwrap();
+    prepare_receipt(
+        &mut pending,
+        &owner,
+        &executor,
+        &observation.request.relay,
+        &[],
+        2000,
+    )
+    .unwrap();
+    let renewed = pending.receipt.clone().unwrap();
+    prepare_receipt(
+        &mut pending,
+        &owner,
+        &executor,
+        &observation.request.relay,
+        &[original],
+        3000,
+    )
+    .unwrap();
+    assert!(pending.published);
+    assert_eq!(pending.receipt.unwrap(), renewed);
+}
+
+#[test]
+fn recovery_rejects_wrong_signer_scope_registration_missing_and_tampered_evidence() {
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (pending, observation) = observed(&owner, &executor);
+    let relay = &observation.request.relay;
+    assert!(prepare_receipt(
+        &mut pending.clone(),
+        &owner,
+        &Keys::generate(),
+        relay,
+        &[],
+        2000
+    )
+    .is_err());
+    assert!(prepare_receipt(
+        &mut pending.clone(),
+        &Keys::generate(),
+        &executor,
+        relay,
+        &[],
+        2000
+    )
+    .is_err());
+    assert!(prepare_receipt(
+        &mut pending.clone(),
+        &owner,
+        &executor,
+        "wss://elsewhere.example",
+        &[],
+        2000
+    )
+    .is_err());
+    let mut changed = pending.clone();
+    changed.registration = host::registration(&owner, executor.public_key(), 100).unwrap();
+    assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err());
+    let mut changed = pending.clone();
+    changed.receipt = None;
+    assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err());
+    let mut changed = pending;
+    changed.receipt.as_mut().unwrap().content.push('x');
+    assert!(prepare_receipt(&mut changed, &owner, &executor, relay, &[], 2000).is_err());
+}
+
+#[test]
+fn forged_history_cannot_confirm_delivery_and_invalid_lifetime_cannot_be_renewed() {
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (mut pending, mut observation) = observed(&owner, &executor);
+    let original = pending.receipt.clone().unwrap();
+    let forged = nostr::EventBuilder::new(original.kind, original.content.clone())
+        .tags(original.tags.clone())
+        .allow_self_tagging()
+        .sign_with_keys(&Keys::generate())
+        .unwrap();
+    prepare_receipt(
+        &mut pending,
+        &owner,
+        &executor,
+        &observation.request.relay,
+        &[forged],
+        1002,
+    )
+    .unwrap();
+    assert!(!pending.published);
+    assert_eq!(pending.receipt.as_ref().unwrap().created_at.as_secs(), 1002);
+    observation.request.expires_at = 401; // original creation + COMMAND_TTL + 1
+    let content = nostr::nips::nip44::encrypt(
+        owner.secret_key(),
+        &executor.public_key(),
+        serde_json::to_string(&observation.request).unwrap(),
+        nostr::nips::nip44::Version::V2,
+    )
+    .unwrap();
+    pending.command = nostr::EventBuilder::new(pending.command.kind, content)
+        .tags(pending.command.tags.clone())
+        .allow_self_tagging()
+        .custom_created_at(nostr::Timestamp::from(100))
+        .sign_with_keys(&owner)
+        .unwrap();
+    observation.command = pending.command.id.to_hex();
+    pending.receipt =
+        Some(host_execution::receipt(&executor, &pending.registration, &observation, 101).unwrap());
+    assert!(prepare_receipt(
+        &mut pending,
+        &owner,
+        &executor,
+        &observation.request.relay,
+        &[],
+        1002
+    )
+    .is_err());
+}
+
+#[tokio::test]
+async fn publication_owner_persists_before_send_and_recovers_late_unaccepted_after_restart() {
+    let dir = tempfile::tempdir().unwrap();
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (pending, observation) = observed(&owner, &executor);
+    let relay = &observation.request.relay;
+    let id = pending.command.id.to_hex();
+    let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap();
+    store.journal.received.insert(id.clone(), pending.clone());
+    // An earlier revoked entry must not starve this one.
+    store.journal.received.insert("a-revoked".into(), pending);
+    store.save().unwrap();
+    let disk = store.path.clone();
+    let prepare = |mut p: Pending, now| {
+        prepare_receipt(&mut p, &owner, &executor, relay, &[], now)?;
+        Ok(p)
+    };
+    store
+        .retry_receipts(
+            |p| std::future::ready(prepare(p, 101)),
+            |_| async { Err("never accepted: disconnected".into()) },
+        )
+        .await
+        .unwrap();
+    drop(store);
+    let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap();
+    store
+        .journal
+        .received
+        .get_mut("a-revoked")
+        .unwrap()
+        .registration = host::registration(&owner, executor.public_key(), 100).unwrap();
+    let attempts = std::sync::atomic::AtomicUsize::new(0);
+    store
+        .retry_receipts(
+            |p| std::future::ready(prepare(p, 1002)),
+            |p| {
+                let (disk, id, owner, attempts) = (&disk, &id, &owner, &attempts);
+                async move {
+                    attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    let durable: Journal =
+                        serde_json::from_slice(&fs::read(disk).unwrap()).unwrap();
+                    assert_eq!(
+                        durable.received[id].receipt, p.receipt,
+                        "fsync before publication"
+                    );
+                    admit(&p, owner, 1002)
+                }
+            },
+        )
+        .await
+        .unwrap();
+    assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
+    assert!(store.journal.received[&id].published);
+    assert!(!store.journal.received["a-revoked"].published);
+    assert!(store.journal.received["a-revoked"].error.is_some());
+    let p = &store.journal.received[&id];
+    assert_eq!(
+        host_execution::decrypt_receipt(
+            &owner,
+            &p.registration,
+            p.receipt.as_ref().unwrap(),
+            &p.command,
+            &observation.request
+        )
+        .unwrap(),
+        observation
+    );
+}
+
+#[tokio::test]
+async fn failed_outbox_save_never_sends_a_renewed_envelope() {
+    let dir = tempfile::tempdir().unwrap();
+    let owner = Keys::generate();
+    let executor = Keys::generate();
+    let (pending, observation) = observed(&owner, &executor);
+    let relay = &observation.request.relay;
+    let mut store = Store::open_dir(dir.path(), "owner", relay).unwrap();
+    store
+        .journal
+        .received
+        .insert(pending.command.id.to_hex(), pending);
+    store.save().unwrap();
+    store.path = dir.path().into(); // cannot atomically replace a directory
+    let result = store
+        .retry_receipts(
+            |mut p| {
+                let result =
+                    prepare_receipt(&mut p, &owner, &executor, relay, &[], 1002).map(|()| p);
+                std::future::ready(result)
+            },
+            |_| async { panic!("publication must not precede durable save") },
+        )
+        .await;
+    assert!(result.is_err());
+}

desktop/src-tauri/src/commands/host_start_store.rs

@@ -1,4 +1,5 @@
-//! Encrypted, immutable transport outbox. Execution authority remains in the
+//! Encrypted transport outbox with immutable command and observation payloads.
+//! Only a receipt transport timestamp/signature may be renewed. Authority remains in the
 //! per-placement journal; transport retry must never create a new operation.
 use nostr::Event;
 use serde::{Deserialize, Serialize};
@@ -77,6 +78,53 @@ impl Store {
         })
     }
 
+    /// Prepare/authenticate one durable result, fsync its envelope, then publish.
+    /// No later entry can age an earlier prepared envelope while doing network
+    /// reads. A failed entry remains visible and never starves other placements.
+    pub async fn retry_receipts<
+        Prepared: std::future::Future<Output = Result<Pending, String>> + Send,
+        Published: std::future::Future<Output = Result<(), String>> + Send,
+    >(
+        &mut self,
+        mut prepare: impl FnMut(Pending) -> Prepared,
+        mut publish: impl FnMut(Pending) -> Published,
+    ) -> Result<(), String> {
+        let ids: Vec<_> = self.journal.received.keys().cloned().collect();
+        for id in ids {
+            let Some(pending) = self.journal.received.get(&id).cloned() else {
+                continue;
+            };
+            if pending.published {
+                continue;
+            }
+            match prepare(pending.clone()).await {
+                Ok(mut ready) => {
+                    // Even a crash inside publish leaves exactly these signed
+                    // bytes durable. Keep the original observation ciphertext.
+                    self.journal.received.insert(id.clone(), ready.clone());
+                    self.save()?;
+                    if !ready.published {
+                        match publish(ready.clone()).await {
+                            Ok(()) => {
+                                ready.published = true;
+                                ready.error = None;
+                            }
+                            Err(error) => ready.error = Some(error),
+                        }
+                    }
+                    self.journal.received.insert(id, ready);
+                }
+                Err(error) => {
+                    let mut failed = pending;
+                    failed.error = Some(error);
+                    self.journal.received.insert(id, failed);
+                }
+            }
+            self.save()?;
+        }
+        Ok(())
+    }
+
     pub fn save(&self) -> Result<(), String> {
         if self.journal.sent.len() + self.journal.received.len() + self.journal.moves.len() > 4096 {
             return Err("Start outbox requires archival".into());
@@ -93,9 +141,82 @@ impl Store {
     }
 }
 
+/// Recover delivery of an already durable observation, never execution. The caller
+/// saves the journal BEFORE sending any renewed envelope and still revalidates
+/// current registration/owner authority at publication. Keep ciphertext byte-exact:
+/// even ambiguous acceptance can only yield the same proof, not a new observation.
+pub(super) fn prepare_receipt(
+    pending: &mut Pending,
+    owner: &nostr::Keys,
+    host: &nostr::Keys,
+    relay: &str,
+    history: &[Event],
+    now: u64,
+) -> Result<(), String> {
+    use buzz_core_pkg::host_execution;
+    if pending.published {
+        return Ok(());
+    }
+    // Authenticate the original command at its signed time, including its bounded
+    // lifetime. This is read-only evidence recovery, not admission of an expired
+    // intent to the executor. Wrong host/community/registration fail closed.
+    let request = host_execution::decrypt_command(
+        host,
+        &pending.registration,
+        &pending.command,
+        relay,
+        pending.command.created_at.as_secs(),
+    )?;
+    let event = pending
+        .receipt
+        .as_ref()
+        .ok_or("missing durable Start receipt")?;
+    host_execution::decrypt_receipt(
+        owner,
+        &pending.registration,
+        event,
+        &pending.command,
+        &request,
+    )?;
+    if event.created_at.as_secs() > now.saturating_add(30) {
+        return Err("receipt timestamp is in the future".into());
+    }
+    // ACK loss is not non-acceptance. Match immutable ciphertext/routing, not only
+    // event ID: an earlier envelope may have committed before a later retry.
+    if history.iter().any(|accepted| {
+        accepted.content == event.content
+            && accepted.tags == event.tags
+            && host_execution::decrypt_receipt(
+                owner,
+                &pending.registration,
+                accepted,
+                &pending.command,
+                &request,
+            )
+            .is_ok()
+    }) {
+        pending.published = true;
+        pending.error = None;
+        return Ok(());
+    }
+    // Renew before the relay's +/-900s admission window, leaving time for I/O.
+    // Recent retries reuse exact bytes; observation/request/run remain unchanged.
+    if now.saturating_sub(event.created_at.as_secs()) >= 600 {
+        pending.receipt = Some(
+            nostr::EventBuilder::new(event.kind, event.content.clone())
+                .allow_self_tagging()
+                .tags(event.tags.clone())
+                .custom_created_at(nostr::Timestamp::from(now))
+                .sign_with_keys(host)
+                .map_err(|_| "receipt transport signing failed")?,
+        );
+    }
+    Ok(())
+}
+
 /// Attempt every eligible entry independently. Failed publication remains pending
 /// and visible; an ACK is recorded only for that exact event. Caller fsyncs before
-/// returning. Dropped ACK/restart consequently resends immutable bytes.
+/// returning. Receipt preparation is persisted separately before this send loop.
 pub(super) async fn retry_pending<Fut: std::future::Future<Output = Result<(), String>> + Send>(
     entries: &mut BTreeMap<String, Pending>,
     receipts: bool,
@@ -221,3 +342,7 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+#[path = "host_start_recovery_tests.rs"]
+mod recovery_tests;

desktop/src-tauri/src/commands/hosts.rs

@@ -471,7 +471,7 @@ pub async fn get_host_history_page(
     {
         return Err("Invalid host history scope".into());
     }
-    let events = crate::relay::query_relay_at_with_keys(
+    let events = crate::relay::query_private_host_at_with_keys(
         &state,
         &crate::relay::relay_http_base_url(&relay_url),
         &[filter],

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

@@ -43,9 +43,7 @@ pub(crate) fn local_execution_config(
     app: &AppHandle,
     record: &ManagedAgentRecord,
 ) -> Result<LocalExecutionConfig, String> {
-    if record.backend != BackendKind::Local {
-        return Err("destination agent is not locally provisioned".into());
-    }
+    local_execution_prerequisites(record)?;
     let personas = load_personas(app)?;
     let teams = load_teams(app)?;
     let global = load_global_agent_config(app)?;
@@ -76,6 +74,18 @@ pub(crate) fn local_execution_config(
     })
 }
 
+// Shared by inspection/advertisement and the launch preflight. Only key
+// availability is checked, exactly as at spawn; no key parsing/export is needed.
+fn local_execution_prerequisites(record: &ManagedAgentRecord) -> Result<(), String> {
+    if record.backend != BackendKind::Local {
+        return Err("destination agent is not locally provisioned".into());
+    }
+    if let Some(error) = super::storage::spawn_key_refusal(record) {
+        return Err(error);
+    }
+    Ok(())
+}
+
 /// Remote execution requires an explicit verified owner attestation, even for
 /// legacy local records. A local key in the store is not a remote launch grant.
 pub(crate) fn execution_agent_owner(
@@ -437,6 +447,24 @@ fn exact_generation_matches(actual: &str, expected: &str) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
+    #[test]
+    fn keyless_record_is_not_advertised_as_locally_provisioned() {
+        let mut record: ManagedAgentRecord = serde_json::from_value(serde_json::json!({
+            "pubkey": "synthetic-agent", "name": "test-agent", "private_key_nsec": "",
+            "relay_url": "wss://relay.example", "acp_command": "buzz-acp",
+            "agent_command": "buzz-agent", "agent_args": [], "mcp_command": "",
+            "turn_timeout_seconds": 320, "created_at": "", "updated_at": ""
+        }))
+        .unwrap();
+        assert_eq!(
+            local_execution_prerequisites(&record).unwrap_err(),
+            super::super::storage::spawn_key_refusal(&record).unwrap()
+        );
+        // Availability only: this preflight must not inspect or export key bytes.
+        record.private_key_nsec = "synthetic-nonempty-placeholder".into();
+        assert!(local_execution_prerequisites(&record).is_ok());
+    }
+
     #[test]
     fn stop_fences_successor_and_malformed_or_legacy_generation() {
         assert!(exact_generation_matches(&"aa".repeat(16), &"aa".repeat(16)));

desktop/src-tauri/src/relay.rs

@@ -350,6 +350,9 @@ pub async fn relay_error_message(response: reqwest::Response) -> String {
     format!("relay returned {status}")
 }
 
+mod private_host;
+pub(crate) use private_host::{query_private_host_at_with_keys, PRIVATE_HOST_REQUEST_TIMEOUT};
+
 // ── HTTP bridge: POST /query ────────────────────────────────────────────────
 
 /// Execute a one-shot query via the relay's HTTP bridge (`POST /query`).

desktop/src-tauri/src/relay/private_host.rs

@@ -0,0 +1,41 @@
+//! Selected-owner private host inventory, authorization and execution history.
+//! These POST bodies must never follow redirects, even when sensitive HTTP
+//! headers would be stripped: the filters themselves disclose private metadata.
+use super::*;
+
+/// Per-request deadline includes response-body consumption, not rate-limit wait.
+/// Private host reads/writes deliberately use a shorter budget than WS history.
+pub(crate) const PRIVATE_HOST_REQUEST_TIMEOUT: std::time::Duration =
+    std::time::Duration::from_secs(15);
+
+/// Read private host records with the current unlocked owner's authority, using
+/// the same no-redirect client and deadline as host execution publication.
+pub(crate) async fn query_private_host_at_with_keys(
+    state: &AppState,
+    api_base_url: &str,
+    filters: &[serde_json::Value],
+    keys: &Keys,
+    auth_tag: Option<&str>,
+) -> Result<Vec<nostr::Event>, String> {
+    crate::relay_admission::wait_for_rate_limit().await;
+    assert_expected_signer(
+        Some(&keys.public_key().to_hex()),
+        &state.signing_keys()?.public_key().to_hex(),
+    )?;
+    let url = format!("{}/query", api_base_url.trim_end_matches('/'));
+    let body =
+        serde_json::to_vec(filters).map_err(|e| format!("filter serialization failed: {e}"))?;
+    let auth = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body)?;
+    send_query_request(
+        &state.media_fetch_client,
+        &url,
+        &auth,
+        auth_tag,
+        body,
+        PRIVATE_HOST_REQUEST_TIMEOUT,
+    )
+    .await
+}
+
+#[cfg(test)]
+mod tests;

desktop/src-tauri/src/relay/private_host/tests.rs

@@ -0,0 +1,187 @@
+use super::*;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+
+async fn read_request(stream: &mut TcpStream) -> (String, Vec<u8>) {
+    let mut bytes = Vec::new();
+    loop {
+        let mut buf = [0; 4096];
+        let n = stream.read(&mut buf).await.unwrap();
+        assert_ne!(n, 0);
+        bytes.extend_from_slice(&buf[..n]);
+        if let Some(end) = bytes.windows(4).position(|b| b == b"\r\n\r\n") {
+            let headers = String::from_utf8(bytes[..end].to_vec()).unwrap();
+            let len: usize = headers
+                .lines()
+                .find_map(|line| {
+                    let (key, value) = line.split_once(':')?;
+                    key.eq_ignore_ascii_case("content-length")
+                        .then(|| value.trim().parse().unwrap())
+                })
+                .unwrap();
+            if bytes.len() >= end + 4 + len {
+                return (headers, bytes[end + 4..end + 4 + len].to_vec());
+            }
+        }
+    }
+}
+
+fn state_and_keys() -> (AppState, Keys) {
+    let state = crate::app_state::build_app_state();
+    let keys = Keys::generate();
+    *state.keys.lock().unwrap() = keys.clone();
+    (state, keys)
+}
+
+#[tokio::test]
+async fn private_host_redirects_never_contact_target_or_disclose_filters() {
+    let _serial = crate::relay_admission::TEST_SERIAL.lock().await;
+    crate::relay_admission::reset_rate_limit_gate();
+    for status in [307, 308] {
+        let origin = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let target = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base = format!("http://{}", origin.local_addr().unwrap());
+        let target_url = format!("http://{}/private-leak", target.local_addr().unwrap());
+        let (state, keys) = state_and_keys();
+        let filters =
+            [serde_json::json!({"kinds":[50002],"#p":[keys.public_key().to_hex()],"limit":1000})];
+        let expected_body = serde_json::to_vec(&filters).unwrap();
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = origin.accept().await.unwrap();
+            let (_, body) = read_request(&mut stream).await;
+            assert_eq!(body, expected_body);
+            stream.write_all(format!("HTTP/1.1 {status} Redirect\r\nLocation: {target_url}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
+        });
+        let result = tokio::time::timeout(
+            Duration::from_secs(3),
+            query_private_host_at_with_keys(&state, &base, &filters, &keys, None),
+        )
+        .await
+        .unwrap();
+        assert!(result.unwrap_err().contains(&status.to_string()));
+        server.await.unwrap();
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), target.accept())
+                .await
+                .is_err(),
+            "redirect target must receive no request, including no filter body"
+        );
+    }
+}
+
+#[tokio::test]
+async fn private_host_configured_auth_is_bound_to_exact_url_and_body() {
+    let _serial = crate::relay_admission::TEST_SERIAL.lock().await;
+    crate::relay_admission::reset_rate_limit_gate();
+    let origin = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base = format!("http://{}", origin.local_addr().unwrap());
+    let url = format!("{base}/query");
+    let (state, keys) = state_and_keys();
+    let signer = keys.public_key();
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = origin.accept().await.unwrap();
+        let (headers, body) = read_request(&mut stream).await;
+        let header = |name: &str| {
+            headers
+                .lines()
+                .find_map(|line| {
+                    let (key, value) = line.split_once(':')?;
+                    key.eq_ignore_ascii_case(name).then(|| value.trim())
+                })
+                .unwrap()
+        };
+        assert_eq!(header("x-auth-tag"), "synthetic-configured-auth");
+        let encoded = header("authorization").strip_prefix("Nostr ").unwrap();
+        let bytes = base64::engine::general_purpose::STANDARD
+            .decode(encoded)
+            .unwrap();
+        let auth = nostr::Event::from_json(bytes).unwrap();
+        buzz_core_pkg::verify_event(&auth).unwrap();
+        assert_eq!(auth.pubkey, signer);
+        assert!(auth.tags.iter().any(|t| t.as_slice() == ["u", &url]));
+        assert!(auth.tags.iter().any(|t| t.as_slice() == ["method", "POST"]));
+        assert!(auth
+            .tags
+            .iter()
+            .any(|t| t.as_slice() == ["payload", &hex::encode(Sha256::digest(&body))]));
+        stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n[]").await.unwrap();
+    });
+    assert!(query_private_host_at_with_keys(
+        &state,
+        &base,
+        &[serde_json::json!({"kinds":[50000],"#p":[signer.to_hex()]})],
+        &keys,
+        Some("synthetic-configured-auth")
+    )
+    .await
+    .unwrap()
+    .is_empty());
+    server.await.unwrap();
+}
+
+#[tokio::test]
+async fn private_host_deadline_covers_stalled_headers_and_body() {
+    // Exercise the actual 15s policy (no test-only shorter timeout).
+    let _serial = crate::relay_admission::TEST_SERIAL.lock().await;
+    crate::relay_admission::reset_rate_limit_gate();
+    assert_eq!(PRIVATE_HOST_REQUEST_TIMEOUT, Duration::from_secs(15));
+    for send_headers in [false, true] {
+        let origin = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base = format!("http://{}", origin.local_addr().unwrap());
+        let (state, keys) = state_and_keys();
+        let server = tokio::spawn(async move {
+            let (mut stream, _) = origin.accept().await.unwrap();
+            read_request(&mut stream).await;
+            if send_headers {
+                stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 50\r\n\r\n[").await.unwrap();
+            }
+            std::future::pending::<()>().await;
+        });
+        let start = std::time::Instant::now();
+        let result = tokio::time::timeout(
+            Duration::from_secs(20),
+            query_private_host_at_with_keys(
+                &state,
+                &base,
+                &[serde_json::json!({"kinds":[50002]})],
+                &keys,
+                None,
+            ),
+        )
+        .await
+        .unwrap();
+        assert_eq!(result.unwrap_err(), "relay unreachable: request timed out");
+        assert!(start.elapsed() >= Duration::from_secs(14));
+        server.abort();
+        let _ = server.await;
+    }
+}
+
+#[tokio::test]
+async fn private_host_locked_or_changed_identity_has_no_egress() {
+    let _serial = crate::relay_admission::TEST_SERIAL.lock().await;
+    crate::relay_admission::reset_rate_limit_gate();
+    let origin = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base = format!("http://{}", origin.local_addr().unwrap());
+    let (state, keys) = state_and_keys();
+    for gate in [&state.keyring_locked, &state.identity_lost] {
+        gate.store(true, std::sync::atomic::Ordering::Release);
+        assert!(
+            query_private_host_at_with_keys(&state, &base, &[], &keys, None)
+                .await
+                .is_err()
+        );
+        gate.store(false, std::sync::atomic::Ordering::Release);
+    }
+    assert!(
+        query_private_host_at_with_keys(&state, &base, &[], &Keys::generate(), None)
+            .await
+            .is_err()
+    );
+    assert!(
+        tokio::time::timeout(Duration::from_millis(100), origin.accept())
+            .await
+            .is_err()
+    );
+}

docs/host-execution.md

@@ -4,7 +4,8 @@ The owner-operated native Start outbox, receiver and Hosts picker are implemente
 behind the **non-default `remote-start-preview` Cargo feature**. Default builds
 advertise `accepts_start: false` and reject queuing. Even a preview build advertises
 Start only with a successful same-owner/community receiver pump in the last 15s
-and at least one destination-local provisioned, compatible configuration. A
+and at least one destination-local provisioned, compatible configuration with an
+available agent key (the same availability refusal as launch). A
 runtime catalog entry alone is not launch capability. This is not a release or
 physical-host/provider certification.
 
@@ -64,7 +65,24 @@ immutable ledger replay and before creating any new intent or side effect. Futur
 timestamps are rejected. Expiration never creates a replacement operation.
 
 The native outbox uses restricted atomic writes, directory fsync and OS file locks.
-It retains signed commands and receipts through ACK loss/reconnect/restart.
+It retains signed commands and receipt ciphertext through ACK loss/reconnect/restart.
+Before retrying a durable receipt, the destination checks accepted history. If the
+result never reached the relay and its envelope is at least 10 minutes old, it
+re-signs the **same encrypted observation bytes and routing tags** with a fresh
+transport timestamp and fsyncs before sending. Original command ID, operation,
+run, outcome and observation time do not change. Recent retries reuse exact
+signed bytes; accepted history of any earlier envelope resolves ACK loss without
+creating another observation. Only a saved, authenticated result may be renewed;
+this never executes an expired intent or opens a new ledger operation. Current
+registration and unlocked owner/community authority are still required.
+
+Owner-private host registration/inventory and execution history reads use the
+same native no-redirect client as Start publication. Their 15-second per-request
+deadline includes response-body consumption (rate-limit admission wait is
+separate). Redirects are errors, including 307/308: neither private filter bodies
+nor authentication are sent to a redirect target. Unrelated query transports
+retain their existing policy.
+
 Publication errors are recorded per entry, remain visible and retry independently;
 a revoked old registration does not starve another operation. Corrupt bindings,
 missing receipts and invalid/cyclic supersession chains fail closed. Transport ACK