merge: integrate targeted transfers with revoke/forget/block

Combine ticket 09 lifecycle APIs with ticket 10 targeted-transfer protocol
and wire forget/block to real cancel_targeted_transfers_for_peer.

Co-authored-by: Cursor <cursoragent@cursor.com>

crates/vnidrop/src/api.rs

@@ -101,6 +101,21 @@ pub struct TargetedTransfer {
     pub updated_at: i64,
 }
 
+/// Pre-approval offer summary. Deliberately omits any reusable share ticket.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, uniffi::Record)]
+pub struct PendingTargetedOffer {
+    pub transfer_id: String,
+    pub sender_endpoint_id: String,
+    pub receiver_endpoint_id: String,
+    pub manifest_id: String,
+    pub content_hash: String,
+    pub transfer_name: String,
+    pub file_count: u64,
+    pub total_size: u64,
+    pub protocol_version: u16,
+    pub received_at: i64,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, uniffi::Enum)]
 pub enum CoreRelayMode {
     Automatic,

crates/vnidrop/src/device_relationship/mod.rs

@@ -847,6 +847,75 @@ impl DeviceRelationshipService {
         Ok(())
     }
 
+    /// Prove possession of the held grant for a Saved peer (targeted offers).
+    pub(crate) async fn prove_saved_possession(
+        &self,
+        peer_endpoint_id: &str,
+        challenge: &Challenge,
+    ) -> Result<(WireProof, u64, u16), VnidropError> {
+        let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| {
+            VnidropError::permission(anyhow::anyhow!("no saved relationship with peer"))
+        })?;
+        if row.state != DeviceRelationshipState::Saved {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "peer is not a saved device"
+            )));
+        }
+        let proof = self
+            .prove_held_possession(
+                peer_endpoint_id,
+                challenge,
+                row.generation,
+                row.minimum_protocol_version,
+            )
+            .await?;
+        Ok((proof, row.generation, row.minimum_protocol_version))
+    }
+
+    /// Verify a Saved peer's held-grant proof against our issued grant.
+    pub(crate) async fn verify_saved_possession(
+        &self,
+        peer_endpoint_id: &str,
+        challenge: &Challenge,
+        proof: &WireProof,
+        generation: u64,
+        protocol_version: u16,
+    ) -> Result<(), VnidropError> {
+        let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| {
+            VnidropError::permission(anyhow::anyhow!("no saved relationship with peer"))
+        })?;
+        if row.state != DeviceRelationshipState::Saved {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "peer is not a saved device"
+            )));
+        }
+        if row.generation != generation {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "relationship generation mismatch"
+            )));
+        }
+        self.verify_issued_possession(
+            peer_endpoint_id,
+            challenge,
+            proof,
+            generation,
+            protocol_version,
+        )
+        .await
+    }
+
+    pub(crate) async fn require_saved(&self, peer_endpoint_id: &str) -> Result<(), VnidropError> {
+        let row = self.find_row(peer_endpoint_id).await?.ok_or_else(|| {
+            VnidropError::permission(anyhow::anyhow!("no saved relationship with peer"))
+        })?;
+        if row.state != DeviceRelationshipState::Saved {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "peer is not a saved device"
+            )));
+        }
+        Ok(())
+    }
+
     async fn prove_held_possession(
         &self,
         peer_endpoint_id: &str,
@@ -1080,7 +1149,10 @@ impl DeviceRelationshipService {
         Ok(())
     }
 
-    async fn peer_addr(&self, peer_endpoint_id: &str) -> Result<EndpointAddr, VnidropError> {
+    pub(crate) async fn peer_addr(
+        &self,
+        peer_endpoint_id: &str,
+    ) -> Result<EndpointAddr, VnidropError> {
         let parsed: EndpointId = peer_endpoint_id
             .parse()
             .context("unusable peer endpoint id")
@@ -1277,10 +1349,10 @@ struct WireGrant {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
-struct WireProof {
+pub(crate) struct WireProof {
-    grant_id: String,
+    pub(crate) grant_id: String,
-    mac: String,
+    pub(crate) mac: String,
-    challenge: String,
+    pub(crate) challenge: String,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/vnidrop/src/lib.rs

@@ -32,10 +32,10 @@ pub use api::{
     CoreEventSink, CoreLimits, CoreNetworkConfig, CoreRelayMode, CoreStorageUsage,
     DeviceRelationship, DeviceRelationshipState, ExperimentalSavedDeviceCapabilities,
     GrantLifetimeSetting, HeldOfferSummary, IncomingOffer, PairingEligibilitySummary,
-    PendingPairing, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedArtifact,
+    PendingPairing, PendingTargetedOffer, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2,
-    ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, SavedDevice, ShareMetadataInput,
+    ReceivedArtifact, ReceivedLocatorKind, ReceiverRequest, RuntimeStatus, SavedDevice,
-    ShareResult, ShareSource, SourceKind, StoredTransfer, TargetedTransfer, TargetedTransferState,
+    ShareMetadataInput, ShareResult, ShareSource, SourceKind, StoredTransfer, TargetedTransfer,
-    TicketInspection, TransferAccessMode, TransferMetadata,
+    TargetedTransferState, TicketInspection, TransferAccessMode, TransferMetadata,
 };
 pub use error::VnidropError;
 pub use runtime::VnidropCore;

crates/vnidrop/src/repository.rs

@@ -324,6 +324,7 @@ impl Repository {
         crate::contacts::ensure_schema(&self.pool).await?;
         crate::secure_secret::ensure_schema(&self.pool).await?;
         crate::device_relationship::DeviceRelationshipService::ensure_schema(&self.pool).await?;
+        crate::targeted_transfer::ensure_schema(&self.pool).await?;
         sqlx::query(
             r#"
             CREATE TABLE IF NOT EXISTS pairing_eligibilities (

crates/vnidrop/src/runtime/contacts.rs

@@ -106,7 +106,7 @@ impl CoreInner {
         // Targeted transfers for this relationship only (ticket 10 fills in).
         // Invitation-domain shares are deliberately not cancelled here.
         self.cancel_targeted_transfers_for_peer(&peer_endpoint_id)
-            .await;
+            .await?;
         self.emit_endpoint(
             "pairing",
             "saved-device-forgotten",
@@ -139,7 +139,7 @@ impl CoreInner {
             .revoke_for_block(&peer_endpoint_id)
             .await?;
         self.cancel_targeted_transfers_for_peer(&peer_endpoint_id)
-            .await;
+            .await?;
         self.offers.discard_from(&peer_endpoint_id).await;
         self.emit_endpoint(
             "pairing",
@@ -182,21 +182,6 @@ impl CoreInner {
             .await
     }
 
-    /// Cancels active or resumable targeted transfers for a peer relationship.
-    ///
-    /// Stub until ticket 10 lands targeted-transfer cancellation. Forget/block
-    /// call this for immediate local effect without touching invitation shares.
-    pub(super) async fn cancel_targeted_transfers_for_peer(&self, peer_endpoint_id: &str) {
-        #[cfg(test)]
-        {
-            self.targeted_cancel_log
-                .lock()
-                .expect("targeted cancel log")
-                .push(peer_endpoint_id.to_string());
-        }
-        let _ = peer_endpoint_id;
-    }
-
     #[cfg(test)]
     pub(super) fn targeted_cancel_log_for_test(&self) -> Vec<String> {
         self.targeted_cancel_log

crates/vnidrop/src/runtime/facade.rs

@@ -484,6 +484,65 @@ impl VnidropCore {
         self.block_on(self.inner.rotate_relationship_grant(peer_endpoint_id))
     }
 
+    /// Create an immutable one-receiver transfer and submit its pre-approval offer.
+    ///
+    /// Blocks until the saved receiver approves or declines. On approval the
+    /// receiver obtains bound authorization via [`Self::respond_to_targeted_offer`].
+    pub fn create_targeted_transfer(
+        &self,
+        receiver_endpoint_id: String,
+        sources: Vec<ShareSource>,
+        transfer_name: Option<String>,
+    ) -> Result<crate::api::TargetedTransfer, VnidropError> {
+        self.block_on(self.inner.create_targeted_transfer(
+            receiver_endpoint_id,
+            sources,
+            transfer_name,
+        ))
+    }
+
+    /// Offline-only pending offers awaiting explicit local approval.
+    pub fn list_pending_targeted_offers(&self) -> Vec<crate::api::PendingTargetedOffer> {
+        self.block_on(self.inner.list_pending_targeted_offers())
+    }
+
+    /// Approve or decline a pending targeted offer.
+    ///
+    /// On approval, returns the recipient-bound authorization capability used
+    /// with [`Self::receive_targeted_transfer`]. Declining returns `None`.
+    pub fn respond_to_targeted_offer(
+        &self,
+        transfer_id: String,
+        accepted: bool,
+    ) -> Result<Option<String>, VnidropError> {
+        self.block_on(self.inner.respond_to_targeted_offer(transfer_id, accepted))
+    }
+
+    /// Pull an approved targeted transfer through existing output-sink machinery.
+    pub fn receive_targeted_transfer(
+        &self,
+        authorization: String,
+        output_dir: String,
+    ) -> Result<(), VnidropError> {
+        self.block_on(
+            self.inner
+                .receive_targeted_transfer(authorization, output_dir),
+        )
+    }
+
+    pub fn get_targeted_transfer(
+        &self,
+        id: String,
+    ) -> Result<Option<crate::api::TargetedTransfer>, VnidropError> {
+        self.block_on(self.inner.get_targeted_transfer(id))
+    }
+
+    pub fn list_targeted_transfers(
+        &self,
+    ) -> Result<Vec<crate::api::TargetedTransfer>, VnidropError> {
+        self.block_on(self.inner.list_targeted_transfers())
+    }
+
     /// Devices the user has chosen to remember.
     pub fn list_contacts(&self) -> Result<Vec<ContactSummary>, VnidropError> {
         self.block_on(self.inner.list_contacts())

crates/vnidrop/src/runtime/mod.rs

@@ -7,6 +7,7 @@
 //! - [`lifecycle`] — cancel/delete/shutdown/status/access
 //! - [`provider`] — blob provider events and per-connection send progress
 //! - [`contacts`] — device history: pairing, forgetting, blocking
+//! - [`targeted`] — saved-device targeted transfers
 
 mod contacts;
 mod delivery;
@@ -16,6 +17,7 @@ mod provider;
 mod receive;
 mod share;
 mod storage;
+mod targeted;
 
 #[cfg(test)]
 pub(crate) use self::contacts::should_poll;
@@ -67,6 +69,7 @@ use crate::{
     repository::Repository,
     secret::load_or_create_secret,
     secure_secret::{start_endpoint_identity, ProfileLock, SecureSecretStore},
+    targeted_transfer::{TargetedOfferInbox, TargetedTransferProtocol},
     ticket::ticket_matches_relay_profile,
     transfer_state::{TransferDirection, TransferStatus},
 };
@@ -109,6 +112,7 @@ pub(super) struct CoreInner {
     pub(super) pairing_eligibility: PairingEligibilityService,
     pub(super) device_relationships: Arc<DeviceRelationshipService>,
     pub(super) offers: OfferInbox,
+    pub(super) targeted_offers: TargetedOfferInbox,
     /// Endpoint → last poll time, for the rate limit above.
     pub(super) last_polled: TokioMutex<HashMap<String, i64>>,
     pub(super) limits: CoreLimits,
@@ -390,6 +394,8 @@ impl CoreInner {
             tracing::warn!(%error, "failed to sweep dead grants");
         }
         let offers = OfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize);
+        let targeted_offers =
+            TargetedOfferInbox::new(event_hub.clone(), limits.max_pending_offers as usize);
         let device_relationships = Arc::new(DeviceRelationshipService::new(
             repository.sqlite_pool(),
             secret_custody.clone(),
@@ -409,6 +415,15 @@ impl CoreInner {
                 RelationshipProtocol::ALPN,
                 RelationshipProtocol::new(device_relationships.clone()),
             )
+            .accept(
+                TargetedTransferProtocol::ALPN,
+                TargetedTransferProtocol::new(
+                    device_relationships.clone(),
+                    targeted_offers.clone(),
+                    limits.clone(),
+                    endpoint.id().to_string(),
+                ),
+            )
             .spawn();
 
         let inner = Arc::new(Self {
@@ -424,6 +439,7 @@ impl CoreInner {
             pairing_eligibility,
             device_relationships,
             offers,
+            targeted_offers,
             last_polled: TokioMutex::new(HashMap::new()),
             relay_mode,
             custom_relay_urls: relay_urls,

crates/vnidrop/src/runtime/targeted.rs

@@ -0,0 +1,342 @@
+//! Create, offer, approve, and receive targeted transfers between Saved devices.
+
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use iroh_blobs::{ticket::BlobTicket, BlobFormat};
+use uuid::Uuid;
+
+use super::CoreInner;
+use crate::{
+    api::{
+        experimental_saved_device_capabilities, PendingTargetedOffer, ShareMetadataInput,
+        ShareSource, TargetedTransfer, TargetedTransferState, TransferAccessMode, TransferMetadata,
+    },
+    error::VnidropError,
+    targeted_transfer::{
+        protocol::{
+            DeliverTargetedAuthorization, SubmitTargetedOffer, TargetedOfferResponse,
+            TargetedTransferProtocol,
+        },
+        TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRow,
+        TargetedTransferStore,
+    },
+    ticket::VnidropTicket,
+    util::{non_empty, now_ms},
+};
+
+const OFFER_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+
+impl CoreInner {
+    pub(super) fn targeted_store(&self) -> TargetedTransferStore {
+        TargetedTransferStore::new(self.repository.sqlite_pool())
+    }
+
+    pub(super) async fn list_pending_targeted_offers(&self) -> Vec<PendingTargetedOffer> {
+        self.targeted_offers.list().await
+    }
+
+    pub(super) async fn get_targeted_transfer(
+        &self,
+        id: String,
+    ) -> Result<Option<TargetedTransfer>, VnidropError> {
+        self.targeted_store().get(&id).await
+    }
+
+    pub(super) async fn list_targeted_transfers(
+        &self,
+    ) -> Result<Vec<TargetedTransfer>, VnidropError> {
+        self.targeted_store().list().await
+    }
+
+    /// Cancel in-flight targeted transfers involving `peer` (for forget/block).
+    pub(crate) async fn cancel_targeted_transfers_for_peer(
+        &self,
+        peer_endpoint_id: &str,
+    ) -> Result<u64, VnidropError> {
+        #[cfg(test)]
+        {
+            self.targeted_cancel_log
+                .lock()
+                .expect("targeted cancel log")
+                .push(peer_endpoint_id.to_string());
+        }
+        self.targeted_offers.discard_from(peer_endpoint_id).await;
+        self.targeted_store().cancel_by_peer(peer_endpoint_id).await
+    }
+
+    pub(super) async fn respond_to_targeted_offer(
+        &self,
+        transfer_id: String,
+        accepted: bool,
+    ) -> Result<Option<String>, VnidropError> {
+        match self.targeted_offers.respond(&transfer_id, accepted).await {
+            Ok(auth) => Ok(auth),
+            Err(crate::targeted_transfer::RespondError::Unknown) => Err(
+                VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer")),
+            ),
+            Err(crate::targeted_transfer::RespondError::SenderGone) => Err(VnidropError::network(
+                anyhow::anyhow!("sender disconnected before approval completed"),
+            )),
+            Err(crate::targeted_transfer::RespondError::AuthorizationTimeout) => Err(
+                VnidropError::network(anyhow::anyhow!("authorization was not delivered in time")),
+            ),
+        }
+    }
+
+    pub(super) async fn create_targeted_transfer(
+        self: &Arc<Self>,
+        receiver_endpoint_id: String,
+        sources: Vec<ShareSource>,
+        transfer_name: Option<String>,
+    ) -> Result<TargetedTransfer, VnidropError> {
+        self.device_relationships
+            .require_saved(&receiver_endpoint_id)
+            .await?;
+
+        let transfer_uuid = Uuid::new_v4().to_string();
+        let protocol_transfer_id = allocate_protocol_transfer_id(&transfer_uuid);
+        let sender_endpoint_id = self.endpoint.id().to_string();
+        let now = now_ms();
+
+        let share = self
+            .share_files(
+                sources,
+                ShareMetadataInput {
+                    transfer_id: protocol_transfer_id,
+                    transfer_name: transfer_name.clone(),
+                    sender_name: None,
+                    access_mode: TransferAccessMode::ApprovalRequired,
+                },
+            )
+            .await
+            .map_err(VnidropError::transfer)?;
+
+        let store = self.targeted_store();
+        let row = TargetedTransferRow {
+            id: transfer_uuid.clone(),
+            protocol_transfer_id,
+            sender_endpoint_id: sender_endpoint_id.clone(),
+            receiver_endpoint_id: receiver_endpoint_id.clone(),
+            manifest_id: share.hash.clone(),
+            content_hash: share.hash.clone(),
+            transfer_name: share.transfer_name.clone(),
+            file_count: share.file_count,
+            total_size: share.total_size,
+            state: TargetedTransferState::Preparing,
+            created_at: now,
+            updated_at: now,
+        };
+        store.insert(&row).await?;
+        store
+            .set_state(
+                &transfer_uuid,
+                TargetedTransferState::Preparing,
+                TargetedTransferState::Offering,
+            )
+            .await?;
+
+        let addr = self
+            .device_relationships
+            .peer_addr(&receiver_endpoint_id)
+            .await?;
+        let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
+        let challenge = tokio::time::timeout(OFFER_CONNECT_TIMEOUT, client.request_challenge())
+            .await
+            .map_err(|_| VnidropError::network(anyhow::anyhow!("device did not answer in time")))?
+            .context("device is not reachable")
+            .map_err(VnidropError::network)?;
+
+        let (proof, generation, relationship_protocol_version) = self
+            .device_relationships
+            .prove_saved_possession(&receiver_endpoint_id, &challenge)
+            .await?;
+
+        let protocol_version =
+            experimental_saved_device_capabilities().targeted_transfer_protocol_version;
+        store
+            .set_state(
+                &transfer_uuid,
+                TargetedTransferState::Offering,
+                TargetedTransferState::AwaitingApproval,
+            )
+            .await?;
+
+        let response = tokio::time::timeout(
+            OFFER_CONNECT_TIMEOUT + std::time::Duration::from_secs(120),
+            client.submit_offer(SubmitTargetedOffer {
+                proof,
+                generation,
+                relationship_protocol_version,
+                protocol_version,
+                transfer_id: transfer_uuid.clone(),
+                sender_endpoint_id: sender_endpoint_id.clone(),
+                receiver_endpoint_id: receiver_endpoint_id.clone(),
+                manifest_id: share.hash.clone(),
+                content_hash: share.hash.clone(),
+                transfer_name: share.transfer_name.clone(),
+                file_count: share.file_count,
+                total_size: share.total_size,
+            }),
+        )
+        .await
+        .map_err(|_| VnidropError::network(anyhow::anyhow!("offer timed out")))?
+        .context("failed to submit targeted offer")
+        .map_err(VnidropError::network)?;
+
+        match response {
+            TargetedOfferResponse::Accepted => {}
+            TargetedOfferResponse::Declined { reason } => {
+                let _ = store
+                    .set_state(
+                        &transfer_uuid,
+                        TargetedTransferState::AwaitingApproval,
+                        TargetedTransferState::Declined,
+                    )
+                    .await;
+                let _ = self.cancel_idle_or_share(protocol_transfer_id).await;
+                return Err(VnidropError::permission(anyhow::anyhow!(
+                    "targeted offer declined: {reason}"
+                )));
+            }
+            TargetedOfferResponse::Refused { reason } => {
+                let _ = store
+                    .set_state(
+                        &transfer_uuid,
+                        TargetedTransferState::AwaitingApproval,
+                        TargetedTransferState::Failed,
+                    )
+                    .await;
+                let _ = self.cancel_idle_or_share(protocol_transfer_id).await;
+                return Err(VnidropError::permission(anyhow::anyhow!(
+                    "targeted offer refused: {reason}"
+                )));
+            }
+        }
+
+        // Bound authorization: only the approved receiver endpoint may fetch.
+        self.access_policy
+            .approve_endpoint(protocol_transfer_id, receiver_endpoint_id.clone())
+            .await;
+
+        let parsed = crate::ticket::parse_transfer_ticket_with_limits(&share.ticket, &self.limits)
+            .map_err(VnidropError::ticket)?;
+        let blob_ticket = BlobTicket::new(
+            parsed.blob_ticket.addr().clone(),
+            parsed.blob_ticket.hash(),
+            BlobFormat::HashSeq,
+        );
+        let authorization = TargetedAuthorization::issue(TargetedAuthorizationDraft {
+            transfer_id: transfer_uuid.clone(),
+            protocol_transfer_id,
+            sender_endpoint_id,
+            receiver_endpoint_id,
+            manifest_id: share.hash.clone(),
+            content_hash: share.hash.clone(),
+            file_count: share.file_count,
+            total_size: share.total_size,
+            protocol_version,
+            transfer_name: share.transfer_name.clone(),
+            blob_ticket: blob_ticket.to_string(),
+        })?;
+        let encoded = authorization.encode()?;
+
+        let deliver = client
+            .deliver_authorization(DeliverTargetedAuthorization {
+                transfer_id: transfer_uuid.clone(),
+                authorization: encoded,
+            })
+            .await
+            .context("failed to deliver targeted authorization")
+            .map_err(VnidropError::network)?;
+        if deliver != crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored {
+            let _ = store
+                .set_state(
+                    &transfer_uuid,
+                    TargetedTransferState::AwaitingApproval,
+                    TargetedTransferState::Failed,
+                )
+                .await;
+            return Err(VnidropError::network(anyhow::anyhow!(
+                "receiver rejected authorization delivery"
+            )));
+        }
+
+        store
+            .set_state(
+                &transfer_uuid,
+                TargetedTransferState::AwaitingApproval,
+                TargetedTransferState::Approved,
+            )
+            .await?;
+
+        store
+            .get(&transfer_uuid)
+            .await?
+            .ok_or_else(|| VnidropError::internal(anyhow::anyhow!("targeted transfer missing")))
+    }
+
+    pub(super) async fn receive_targeted_transfer(
+        self: &Arc<Self>,
+        authorization: String,
+        output_dir: String,
+    ) -> Result<(), VnidropError> {
+        let auth = TargetedAuthorization::decode(&authorization)?;
+        auth.verify_for_receiver(&self.endpoint.id().to_string())?;
+
+        let blob_ticket = BlobTicket::from_str_compat(&auth.blob_ticket)
+            .map_err(|error| VnidropError::ticket(anyhow::anyhow!(error)))?;
+        let metadata = TransferMetadata::new(
+            auth.protocol_transfer_id,
+            non_empty(auth.transfer_name.clone()).unwrap_or_else(|| "transfer".to_string()),
+            None,
+            blob_ticket.hash(),
+            auth.file_count,
+            auth.total_size,
+        );
+        let ticket =
+            VnidropTicket::new_with_relay_urls(blob_ticket, metadata, &self.custom_relay_urls)
+                .encode()
+                .map_err(VnidropError::ticket)?;
+
+        self.receive(ticket, std::path::PathBuf::from(output_dir), None)
+            .await
+            .map_err(VnidropError::transfer)?;
+
+        if let Ok(Some(row)) = self.targeted_store().get_row(&auth.transfer_id).await {
+            let _ = self
+                .targeted_store()
+                .set_state(
+                    &auth.transfer_id,
+                    row.state,
+                    TargetedTransferState::Completed,
+                )
+                .await;
+        }
+        Ok(())
+    }
+}
+
+fn allocate_protocol_transfer_id(transfer_uuid: &str) -> u64 {
+    let hash = blake3::hash(transfer_uuid.as_bytes());
+    let mut bytes = [0u8; 8];
+    bytes.copy_from_slice(&hash.as_bytes()[..8]);
+    // SQLite transfer ids are signed; keep within i64::MAX.
+    let value = u64::from_le_bytes(bytes) & (i64::MAX as u64);
+    if value == 0 {
+        1
+    } else {
+        value
+    }
+}
+
+trait BlobTicketParse {
+    fn from_str_compat(value: &str) -> Result<BlobTicket, String>;
+}
+
+impl BlobTicketParse for BlobTicket {
+    fn from_str_compat(value: &str) -> Result<BlobTicket, String> {
+        use std::str::FromStr;
+        BlobTicket::from_str(value).map_err(|error| error.to_string())
+    }
+}

crates/vnidrop/src/targeted_transfer/auth.rs

@@ -0,0 +1,140 @@
+//! Bound post-approval authorization for a single targeted transfer.
+//!
+//! The pre-approval offer never carries this material. After explicit approval,
+//! the sender issues a capability whose MAC binds the exact recipient, sender,
+//! transfer, manifest, hashes, sizes, and protocol generation. Tampering with
+//! the receiver identity invalidates the MAC; presenting an intact capability
+//! from another endpoint still fails provider ACL and local identity checks.
+
+use data_encoding::{BASE64URL_NOPAD, HEXLOWER};
+use serde::{Deserialize, Serialize};
+
+use crate::error::VnidropError;
+
+const AUTH_CONTEXT: &[u8] = b"vnidrop-targeted-auth-v1";
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct TargetedAuthorization {
+    pub(crate) transfer_id: String,
+    pub(crate) protocol_transfer_id: u64,
+    pub(crate) sender_endpoint_id: String,
+    pub(crate) receiver_endpoint_id: String,
+    pub(crate) manifest_id: String,
+    pub(crate) content_hash: String,
+    pub(crate) file_count: u64,
+    pub(crate) total_size: u64,
+    pub(crate) protocol_version: u16,
+    pub(crate) transfer_name: String,
+    /// BlobTicket string used only after approval to pull through existing sinks.
+    pub(crate) blob_ticket: String,
+    pub(crate) auth_secret: String,
+    pub(crate) mac: String,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct TargetedAuthorizationDraft {
+    pub(crate) transfer_id: String,
+    pub(crate) protocol_transfer_id: u64,
+    pub(crate) sender_endpoint_id: String,
+    pub(crate) receiver_endpoint_id: String,
+    pub(crate) manifest_id: String,
+    pub(crate) content_hash: String,
+    pub(crate) file_count: u64,
+    pub(crate) total_size: u64,
+    pub(crate) protocol_version: u16,
+    pub(crate) transfer_name: String,
+    pub(crate) blob_ticket: String,
+}
+
+impl TargetedAuthorization {
+    pub(crate) fn issue(draft: TargetedAuthorizationDraft) -> Result<Self, VnidropError> {
+        let auth_secret = crate::grant::GrantSecret::generate().encode();
+        let mut auth = Self {
+            transfer_id: draft.transfer_id,
+            protocol_transfer_id: draft.protocol_transfer_id,
+            sender_endpoint_id: draft.sender_endpoint_id,
+            receiver_endpoint_id: draft.receiver_endpoint_id,
+            manifest_id: draft.manifest_id,
+            content_hash: draft.content_hash,
+            file_count: draft.file_count,
+            total_size: draft.total_size,
+            protocol_version: draft.protocol_version,
+            transfer_name: draft.transfer_name,
+            blob_ticket: draft.blob_ticket,
+            auth_secret,
+            mac: String::new(),
+        };
+        auth.mac = HEXLOWER.encode(&auth.compute_mac()?);
+        Ok(auth)
+    }
+
+    pub(crate) fn encode(&self) -> Result<String, VnidropError> {
+        let bytes = serde_json::to_vec(self).map_err(VnidropError::internal)?;
+        Ok(format!("vndta1:{}", BASE64URL_NOPAD.encode(&bytes)))
+    }
+
+    pub(crate) fn decode(value: &str) -> Result<Self, VnidropError> {
+        let encoded = value.strip_prefix("vndta1:").ok_or_else(|| {
+            VnidropError::invalid_input(anyhow::anyhow!("not a targeted authorization"))
+        })?;
+        let bytes = BASE64URL_NOPAD.decode(encoded.as_bytes()).map_err(|_| {
+            VnidropError::invalid_input(anyhow::anyhow!("invalid targeted authorization encoding"))
+        })?;
+        let auth: Self = serde_json::from_slice(&bytes).map_err(|_| {
+            VnidropError::invalid_input(anyhow::anyhow!("invalid targeted authorization payload"))
+        })?;
+        auth.verify_integrity()?;
+        Ok(auth)
+    }
+
+    pub(crate) fn verify_for_receiver(&self, local_endpoint_id: &str) -> Result<(), VnidropError> {
+        self.verify_integrity()?;
+        if self.receiver_endpoint_id != local_endpoint_id {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "targeted authorization is bound to a different receiver"
+            )));
+        }
+        Ok(())
+    }
+
+    fn verify_integrity(&self) -> Result<(), VnidropError> {
+        let expected = self.compute_mac()?;
+        let presented = HEXLOWER.decode(self.mac.as_bytes()).map_err(|_| {
+            VnidropError::invalid_input(anyhow::anyhow!("invalid authorization mac"))
+        })?;
+        if presented.as_slice() != expected.as_slice() {
+            return Err(VnidropError::permission(anyhow::anyhow!(
+                "targeted authorization mac mismatch"
+            )));
+        }
+        Ok(())
+    }
+
+    fn compute_mac(&self) -> Result<[u8; 32], VnidropError> {
+        let secret_bytes = HEXLOWER.decode(self.auth_secret.as_bytes()).map_err(|_| {
+            VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret"))
+        })?;
+        let key: [u8; 32] = secret_bytes.try_into().map_err(|_| {
+            VnidropError::invalid_input(anyhow::anyhow!("invalid authorization secret length"))
+        })?;
+        let mut hasher = blake3::Hasher::new_keyed(&key);
+        hasher.update(AUTH_CONTEXT);
+        for field in [
+            self.transfer_id.as_bytes(),
+            self.sender_endpoint_id.as_bytes(),
+            self.receiver_endpoint_id.as_bytes(),
+            self.manifest_id.as_bytes(),
+            self.content_hash.as_bytes(),
+            self.transfer_name.as_bytes(),
+            self.blob_ticket.as_bytes(),
+        ] {
+            hasher.update(&(field.len() as u64).to_le_bytes());
+            hasher.update(field);
+        }
+        hasher.update(&self.protocol_transfer_id.to_le_bytes());
+        hasher.update(&self.file_count.to_le_bytes());
+        hasher.update(&self.total_size.to_le_bytes());
+        hasher.update(&self.protocol_version.to_le_bytes());
+        Ok(*hasher.finalize().as_bytes())
+    }
+}

crates/vnidrop/src/targeted_transfer/inbox.rs

@@ -0,0 +1,234 @@
+//! Live-session inbox for unapproved targeted-transfer offers.
+//!
+//! Offers are not durable: cancellation, timeout, disconnect, or restart drops
+//! them. Authorization is delivered only after the local user accepts.
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use serde_json::json;
+use tokio::sync::{oneshot, Mutex};
+use uuid::Uuid;
+
+use crate::{api::PendingTargetedOffer, event_hub::EventHub};
+
+const OFFER_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
+
+#[derive(Debug, Clone)]
+pub(crate) struct PendingTargetedOfferRecord {
+    pub(crate) offer: PendingTargetedOffer,
+}
+
+struct DecisionWaiter {
+    decision: oneshot::Sender<bool>,
+}
+
+struct AuthWaiter {
+    auth: oneshot::Sender<String>,
+}
+
+#[derive(Clone)]
+pub(crate) struct TargetedOfferInbox {
+    event_hub: Arc<EventHub>,
+    pending: Arc<Mutex<HashMap<String, PendingTargetedOfferRecord>>>,
+    decisions: Arc<Mutex<HashMap<String, DecisionWaiter>>>,
+    auths: Arc<Mutex<HashMap<String, AuthWaiter>>>,
+    max_pending: usize,
+}
+
+impl TargetedOfferInbox {
+    pub(crate) fn new(event_hub: Arc<EventHub>, max_pending: usize) -> Self {
+        Self {
+            event_hub,
+            pending: Arc::new(Mutex::new(HashMap::new())),
+            decisions: Arc::new(Mutex::new(HashMap::new())),
+            auths: Arc::new(Mutex::new(HashMap::new())),
+            max_pending,
+        }
+    }
+
+    /// Surface a validated offer and block until the local user decides.
+    pub(crate) async fn submit(&self, offer: PendingTargetedOffer) -> TargetedOfferDecision {
+        let transfer_id = offer.transfer_id.clone();
+        let (decision_tx, decision_rx) = oneshot::channel();
+        {
+            let mut pending = self.pending.lock().await;
+            if pending.len() >= self.max_pending {
+                return TargetedOfferDecision::Refused {
+                    reason: "too-many-pending-offers".to_string(),
+                };
+            }
+            if pending
+                .values()
+                .any(|entry| entry.offer.sender_endpoint_id == offer.sender_endpoint_id)
+            {
+                return TargetedOfferDecision::Refused {
+                    reason: "offer-already-pending".to_string(),
+                };
+            }
+            if pending.contains_key(&transfer_id) {
+                return TargetedOfferDecision::Refused {
+                    reason: "duplicate-transfer".to_string(),
+                };
+            }
+            pending.insert(
+                transfer_id.clone(),
+                PendingTargetedOfferRecord {
+                    offer: offer.clone(),
+                },
+            );
+        }
+        self.decisions.lock().await.insert(
+            transfer_id.clone(),
+            DecisionWaiter {
+                decision: decision_tx,
+            },
+        );
+
+        self.event_hub.emit_endpoint(
+            "targeted_transfer",
+            "offer-received",
+            json!({
+                "transfer_id": offer.transfer_id,
+                "sender_endpoint_id": offer.sender_endpoint_id,
+                "file_count": offer.file_count,
+                "total_size": offer.total_size,
+                "manifest_id": offer.manifest_id,
+            }),
+        );
+
+        match tokio::time::timeout(OFFER_WAIT_TIMEOUT, decision_rx).await {
+            Ok(Ok(true)) => TargetedOfferDecision::Accepted,
+            Ok(Ok(false)) => {
+                self.discard(&transfer_id).await;
+                TargetedOfferDecision::Declined {
+                    reason: "receiver-declined".to_string(),
+                }
+            }
+            Ok(Err(_)) | Err(_) => {
+                self.discard(&transfer_id).await;
+                TargetedOfferDecision::Declined {
+                    reason: "no-response".to_string(),
+                }
+            }
+        }
+    }
+
+    pub(crate) async fn list(&self) -> Vec<PendingTargetedOffer> {
+        self.pending
+            .lock()
+            .await
+            .values()
+            .map(|entry| entry.offer.clone())
+            .collect()
+    }
+
+    /// Record the local decision. On accept, wait for sender-issued authorization.
+    pub(crate) async fn respond(
+        &self,
+        transfer_id: &str,
+        accepted: bool,
+    ) -> Result<Option<String>, RespondError> {
+        let exists = self.pending.lock().await.contains_key(transfer_id);
+        if !exists {
+            return Err(RespondError::Unknown);
+        }
+        let waiter = self.decisions.lock().await.remove(transfer_id);
+        let Some(waiter) = waiter else {
+            return Err(RespondError::Unknown);
+        };
+        if !accepted {
+            let _ = waiter.decision.send(false);
+            self.discard(transfer_id).await;
+            self.event_hub.emit_endpoint(
+                "targeted_transfer",
+                "offer-declined",
+                json!({ "transfer_id": transfer_id }),
+            );
+            return Ok(None);
+        }
+
+        let (auth_tx, auth_rx) = oneshot::channel();
+        self.auths
+            .lock()
+            .await
+            .insert(transfer_id.to_string(), AuthWaiter { auth: auth_tx });
+        if waiter.decision.send(true).is_err() {
+            self.auths.lock().await.remove(transfer_id);
+            self.discard(transfer_id).await;
+            return Err(RespondError::SenderGone);
+        }
+
+        match tokio::time::timeout(OFFER_WAIT_TIMEOUT, auth_rx).await {
+            Ok(Ok(auth)) => {
+                self.pending.lock().await.remove(transfer_id);
+                self.event_hub.emit_endpoint(
+                    "targeted_transfer",
+                    "offer-accepted",
+                    json!({ "transfer_id": transfer_id }),
+                );
+                Ok(Some(auth))
+            }
+            Ok(Err(_)) | Err(_) => {
+                self.discard(transfer_id).await;
+                Err(RespondError::AuthorizationTimeout)
+            }
+        }
+    }
+
+    pub(crate) async fn deliver_authorization(
+        &self,
+        transfer_id: &str,
+        authorization: String,
+    ) -> bool {
+        if let Some(waiter) = self.auths.lock().await.remove(transfer_id) {
+            waiter.auth.send(authorization).is_ok()
+        } else {
+            false
+        }
+    }
+
+    #[allow(
+        dead_code,
+        reason = "called via cancel_targeted_transfers_for_peer for ticket 09"
+    )]
+    pub(crate) async fn discard_from(&self, endpoint_id: &str) {
+        let ids: Vec<String> = {
+            let pending = self.pending.lock().await;
+            pending
+                .values()
+                .filter(|entry| entry.offer.sender_endpoint_id == endpoint_id)
+                .map(|entry| entry.offer.transfer_id.clone())
+                .collect()
+        };
+        for id in ids {
+            self.discard(&id).await;
+        }
+    }
+
+    async fn discard(&self, transfer_id: &str) {
+        self.pending.lock().await.remove(transfer_id);
+        if let Some(waiter) = self.decisions.lock().await.remove(transfer_id) {
+            let _ = waiter.decision.send(false);
+        }
+        self.auths.lock().await.remove(transfer_id);
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum TargetedOfferDecision {
+    Accepted,
+    Declined { reason: String },
+    Refused { reason: String },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum RespondError {
+    Unknown,
+    SenderGone,
+    AuthorizationTimeout,
+}
+
+#[allow(dead_code)]
+pub(crate) fn new_offer_id() -> String {
+    Uuid::new_v4().to_string()
+}

crates/vnidrop/src/targeted_transfer/mod.rs

@@ -0,0 +1,267 @@
+//! Immutable one-sender, one-receiver transfers between Saved devices.
+//!
+//! Separate from ordinary multi-receiver shares: own protocol, authorization,
+//! and public APIs. Blob import/streaming/output sinks are reused.
+
+mod auth;
+pub(crate) mod inbox;
+pub(crate) mod protocol;
+mod state;
+
+pub(crate) use auth::{TargetedAuthorization, TargetedAuthorizationDraft};
+pub(crate) use inbox::{RespondError, TargetedOfferInbox};
+pub(crate) use protocol::TargetedTransferProtocol;
+
+use sqlx::{Row, SqlitePool};
+
+use crate::{
+    api::{TargetedTransfer, TargetedTransferState},
+    error::VnidropError,
+    util::now_ms,
+};
+
+pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"
+        CREATE TABLE IF NOT EXISTS targeted_transfers (
+            id TEXT PRIMARY KEY,
+            protocol_transfer_id INTEGER NOT NULL UNIQUE,
+            sender_endpoint_id TEXT NOT NULL,
+            receiver_endpoint_id TEXT NOT NULL,
+            manifest_id TEXT NOT NULL,
+            content_hash TEXT NOT NULL,
+            transfer_name TEXT NOT NULL,
+            file_count INTEGER NOT NULL,
+            total_size INTEGER NOT NULL,
+            state TEXT NOT NULL,
+            created_at INTEGER NOT NULL,
+            updated_at INTEGER NOT NULL
+        );
+        "#,
+    )
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub(crate) struct TargetedTransferStore {
+    pool: SqlitePool,
+}
+
+impl TargetedTransferStore {
+    pub(crate) fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    pub(crate) async fn insert(&self, transfer: &TargetedTransferRow) -> Result<(), VnidropError> {
+        sqlx::query(
+            r#"
+            INSERT INTO targeted_transfers (
+                id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id,
+                manifest_id, content_hash, transfer_name, file_count, total_size,
+                state, created_at, updated_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
+            "#,
+        )
+        .bind(&transfer.id)
+        .bind(transfer.protocol_transfer_id as i64)
+        .bind(&transfer.sender_endpoint_id)
+        .bind(&transfer.receiver_endpoint_id)
+        .bind(&transfer.manifest_id)
+        .bind(&transfer.content_hash)
+        .bind(&transfer.transfer_name)
+        .bind(transfer.file_count as i64)
+        .bind(transfer.total_size as i64)
+        .bind(state_as_str(transfer.state))
+        .bind(transfer.created_at)
+        .bind(transfer.updated_at)
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(())
+    }
+
+    pub(crate) async fn set_state(
+        &self,
+        id: &str,
+        from: TargetedTransferState,
+        to: TargetedTransferState,
+    ) -> Result<(), VnidropError> {
+        from.validate_transition_to(to)?;
+        let result = sqlx::query(
+            r#"
+            UPDATE targeted_transfers
+            SET state = ?2, updated_at = ?3
+            WHERE id = ?1 AND state = ?4
+            "#,
+        )
+        .bind(id)
+        .bind(state_as_str(to))
+        .bind(now_ms())
+        .bind(state_as_str(from))
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        if result.rows_affected() == 0 {
+            return Err(VnidropError::InvalidTransition {
+                reason: format!("{} -> {}", state_as_str(from), state_as_str(to)),
+            });
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn get(&self, id: &str) -> Result<Option<TargetedTransfer>, VnidropError> {
+        let row = sqlx::query(
+            r#"
+            SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id,
+                   file_count, total_size, state, created_at, updated_at
+            FROM targeted_transfers WHERE id = ?1
+            "#,
+        )
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        row.map(row_to_transfer).transpose()
+    }
+
+    pub(crate) async fn get_row(
+        &self,
+        id: &str,
+    ) -> Result<Option<TargetedTransferRow>, VnidropError> {
+        let row = sqlx::query(
+            r#"
+            SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id,
+                   manifest_id, content_hash, transfer_name, file_count, total_size,
+                   state, created_at, updated_at
+            FROM targeted_transfers WHERE id = ?1
+            "#,
+        )
+        .bind(id)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        row.map(row_to_full).transpose()
+    }
+
+    pub(crate) async fn list(&self) -> Result<Vec<TargetedTransfer>, VnidropError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id,
+                   file_count, total_size, state, created_at, updated_at
+            FROM targeted_transfers
+            ORDER BY updated_at DESC
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        rows.into_iter().map(row_to_transfer).collect()
+    }
+
+    #[allow(
+        dead_code,
+        reason = "called via cancel_targeted_transfers_for_peer for ticket 09"
+    )]
+    pub(crate) async fn cancel_by_peer(&self, peer_endpoint_id: &str) -> Result<u64, VnidropError> {
+        let now = now_ms();
+        let result = sqlx::query(
+            r#"
+            UPDATE targeted_transfers
+            SET state = 'cancelled', updated_at = ?2
+            WHERE (sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1)
+              AND state NOT IN ('completed', 'declined', 'cancelled', 'failed', 'deleted')
+            "#,
+        )
+        .bind(peer_endpoint_id)
+        .bind(now)
+        .execute(&self.pool)
+        .await
+        .map_err(VnidropError::repository)?;
+        Ok(result.rows_affected())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct TargetedTransferRow {
+    pub(crate) id: String,
+    pub(crate) protocol_transfer_id: u64,
+    pub(crate) sender_endpoint_id: String,
+    pub(crate) receiver_endpoint_id: String,
+    pub(crate) manifest_id: String,
+    pub(crate) content_hash: String,
+    pub(crate) transfer_name: String,
+    pub(crate) file_count: u64,
+    pub(crate) total_size: u64,
+    pub(crate) state: TargetedTransferState,
+    pub(crate) created_at: i64,
+    pub(crate) updated_at: i64,
+}
+
+fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransfer, VnidropError> {
+    Ok(TargetedTransfer {
+        id: row.get("id"),
+        sender_endpoint_id: row.get("sender_endpoint_id"),
+        receiver_endpoint_id: row.get("receiver_endpoint_id"),
+        manifest_id: row.get("manifest_id"),
+        file_count: row.get::<i64, _>("file_count") as u64,
+        total_size: row.get::<i64, _>("total_size") as u64,
+        state: parse_state(&row.get::<String, _>("state"))?,
+        created_at: row.get("created_at"),
+        updated_at: row.get("updated_at"),
+    })
+}
+
+fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransferRow, VnidropError> {
+    Ok(TargetedTransferRow {
+        id: row.get("id"),
+        protocol_transfer_id: row.get::<i64, _>("protocol_transfer_id") as u64,
+        sender_endpoint_id: row.get("sender_endpoint_id"),
+        receiver_endpoint_id: row.get("receiver_endpoint_id"),
+        manifest_id: row.get("manifest_id"),
+        content_hash: row.get("content_hash"),
+        transfer_name: row.get("transfer_name"),
+        file_count: row.get::<i64, _>("file_count") as u64,
+        total_size: row.get::<i64, _>("total_size") as u64,
+        state: parse_state(&row.get::<String, _>("state"))?,
+        created_at: row.get("created_at"),
+        updated_at: row.get("updated_at"),
+    })
+}
+
+pub(crate) fn state_as_str(state: TargetedTransferState) -> &'static str {
+    match state {
+        TargetedTransferState::Preparing => "preparing",
+        TargetedTransferState::Offering => "offering",
+        TargetedTransferState::AwaitingApproval => "awaiting_approval",
+        TargetedTransferState::Approved => "approved",
+        TargetedTransferState::Connecting => "connecting",
+        TargetedTransferState::Transferring => "transferring",
+        TargetedTransferState::Interrupted => "interrupted",
+        TargetedTransferState::Completed => "completed",
+        TargetedTransferState::Declined => "declined",
+        TargetedTransferState::Cancelled => "cancelled",
+        TargetedTransferState::Failed => "failed",
+        TargetedTransferState::Deleted => "deleted",
+    }
+}
+
+fn parse_state(value: &str) -> Result<TargetedTransferState, VnidropError> {
+    match value {
+        "preparing" => Ok(TargetedTransferState::Preparing),
+        "offering" => Ok(TargetedTransferState::Offering),
+        "awaiting_approval" => Ok(TargetedTransferState::AwaitingApproval),
+        "approved" => Ok(TargetedTransferState::Approved),
+        "connecting" => Ok(TargetedTransferState::Connecting),
+        "transferring" => Ok(TargetedTransferState::Transferring),
+        "interrupted" => Ok(TargetedTransferState::Interrupted),
+        "completed" => Ok(TargetedTransferState::Completed),
+        "declined" => Ok(TargetedTransferState::Declined),
+        "cancelled" => Ok(TargetedTransferState::Cancelled),
+        "failed" => Ok(TargetedTransferState::Failed),
+        "deleted" => Ok(TargetedTransferState::Deleted),
+        other => Err(VnidropError::repository(anyhow::anyhow!(
+            "unknown targeted transfer state: {other}"
+        ))),
+    }
+}

crates/vnidrop/src/targeted_transfer/protocol.rs

@@ -0,0 +1,297 @@
+//! Targeted-transfer control-plane protocol (design §10).
+//!
+//! Separate ALPN from ordinary offers: pre-approval messages carry a manifest
+//! summary and relationship proof only — never a reusable share ticket.
+
+use std::fmt;
+
+use anyhow::Result;
+use iroh::{
+    endpoint::Connection,
+    protocol::{AcceptError, ProtocolHandler},
+    Endpoint, EndpointAddr,
+};
+use irpc::{channel::oneshot, rpc_requests, Client, WithChannels};
+use irpc_iroh::{read_request, IrohLazyRemoteConnection};
+use serde::{Deserialize, Serialize};
+
+use super::{
+    auth::TargetedAuthorization,
+    inbox::{TargetedOfferDecision, TargetedOfferInbox},
+};
+use crate::{
+    api::{experimental_saved_device_capabilities, PendingTargetedOffer},
+    device_relationship::{DeviceRelationshipService, WireProof},
+    error::VnidropError,
+    grant::Challenge,
+    util::now_ms,
+};
+
+#[derive(Clone)]
+pub(crate) struct TargetedTransferProtocol {
+    relationships: std::sync::Arc<DeviceRelationshipService>,
+    inbox: TargetedOfferInbox,
+    limits: crate::api::CoreLimits,
+    local_endpoint_id: String,
+}
+
+impl fmt::Debug for TargetedTransferProtocol {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("TargetedTransferProtocol")
+    }
+}
+
+impl TargetedTransferProtocol {
+    pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/1";
+
+    pub(crate) fn new(
+        relationships: std::sync::Arc<DeviceRelationshipService>,
+        inbox: TargetedOfferInbox,
+        limits: crate::api::CoreLimits,
+        local_endpoint_id: String,
+    ) -> Self {
+        Self {
+            relationships,
+            inbox,
+            limits,
+            local_endpoint_id,
+        }
+    }
+
+    pub(crate) fn client(endpoint: Endpoint, addr: EndpointAddr) -> TargetedTransferClient {
+        TargetedTransferClient {
+            inner: Client::boxed(IrohLazyRemoteConnection::new(
+                endpoint,
+                addr,
+                Self::ALPN.to_vec(),
+            )),
+        }
+    }
+
+    async fn handle_offer(
+        &self,
+        remote_endpoint_id: &str,
+        challenge: &Challenge,
+        offer: SubmitTargetedOffer,
+    ) -> TargetedOfferResponse {
+        let expected = experimental_saved_device_capabilities().targeted_transfer_protocol_version;
+        if offer.protocol_version != expected {
+            return TargetedOfferResponse::Refused {
+                reason: "protocol-incompatible".to_string(),
+            };
+        }
+        if offer.receiver_endpoint_id != self.local_endpoint_id {
+            return TargetedOfferResponse::Refused {
+                reason: "receiver-mismatch".to_string(),
+            };
+        }
+        if offer.sender_endpoint_id != remote_endpoint_id {
+            return TargetedOfferResponse::Refused {
+                reason: "sender-mismatch".to_string(),
+            };
+        }
+        if offer.file_count == 0
+            || offer.file_count > self.limits.max_collection_files
+            || offer.total_size == 0
+            || offer.total_size > self.limits.max_total_bytes
+            || offer.transfer_id.is_empty()
+            || offer.manifest_id.is_empty()
+            || offer.content_hash.is_empty()
+        {
+            return TargetedOfferResponse::Refused {
+                reason: "manifest-limits".to_string(),
+            };
+        }
+        if let Err(error) = self
+            .limits
+            .validate_metadata_text("transfer name", Some(offer.transfer_name.as_str()))
+        {
+            return TargetedOfferResponse::Refused {
+                reason: error.to_string(),
+            };
+        }
+
+        if let Err(error) = self
+            .relationships
+            .verify_saved_possession(
+                remote_endpoint_id,
+                challenge,
+                &offer.proof,
+                offer.generation,
+                offer.relationship_protocol_version,
+            )
+            .await
+        {
+            tracing::debug!(%error, "targeted offer relationship proof rejected");
+            return TargetedOfferResponse::Refused {
+                reason: "unauthenticated".to_string(),
+            };
+        }
+
+        let pending = PendingTargetedOffer {
+            transfer_id: offer.transfer_id,
+            sender_endpoint_id: remote_endpoint_id.to_string(),
+            receiver_endpoint_id: self.local_endpoint_id.clone(),
+            manifest_id: offer.manifest_id,
+            content_hash: offer.content_hash,
+            transfer_name: offer.transfer_name,
+            file_count: offer.file_count,
+            total_size: offer.total_size,
+            protocol_version: offer.protocol_version,
+            received_at: now_ms(),
+        };
+
+        match self.inbox.submit(pending).await {
+            TargetedOfferDecision::Accepted => TargetedOfferResponse::Accepted,
+            TargetedOfferDecision::Declined { reason } => {
+                TargetedOfferResponse::Declined { reason }
+            }
+            TargetedOfferDecision::Refused { reason } => TargetedOfferResponse::Refused { reason },
+        }
+    }
+
+    async fn handle_deliver_authorization(
+        &self,
+        remote_endpoint_id: &str,
+        delivery: DeliverTargetedAuthorization,
+    ) -> DeliverAuthorizationResponse {
+        let Ok(auth) = TargetedAuthorization::decode(&delivery.authorization) else {
+            return DeliverAuthorizationResponse::Rejected;
+        };
+        if auth.sender_endpoint_id != remote_endpoint_id
+            || auth.receiver_endpoint_id != self.local_endpoint_id
+            || auth.transfer_id != delivery.transfer_id
+        {
+            return DeliverAuthorizationResponse::Rejected;
+        }
+        if self
+            .inbox
+            .deliver_authorization(&delivery.transfer_id, delivery.authorization)
+            .await
+        {
+            DeliverAuthorizationResponse::Stored
+        } else {
+            DeliverAuthorizationResponse::Rejected
+        }
+    }
+}
+
+impl ProtocolHandler for TargetedTransferProtocol {
+    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
+        let remote_endpoint_id = connection.remote_id().to_string();
+        let challenge = Challenge::generate();
+        while let Some(message) = read_request::<TargetedTransferMessages>(&connection).await? {
+            match message {
+                TargetedTransferMessage::RequestChallenge(message) => {
+                    let WithChannels { tx, .. } = message;
+                    let _ = tx
+                        .send(ChallengeResponse {
+                            challenge: challenge.clone(),
+                        })
+                        .await;
+                }
+                TargetedTransferMessage::SubmitTargetedOffer(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let response = self
+                        .handle_offer(&remote_endpoint_id, &challenge, inner)
+                        .await;
+                    let _ = tx.send(response).await;
+                }
+                TargetedTransferMessage::DeliverTargetedAuthorization(message) => {
+                    let WithChannels { inner, tx, .. } = message;
+                    let response = self
+                        .handle_deliver_authorization(&remote_endpoint_id, inner)
+                        .await;
+                    let _ = tx.send(response).await;
+                }
+            }
+        }
+        connection.closed().await;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct TargetedTransferClient {
+    inner: Client<TargetedTransferMessages>,
+}
+
+impl TargetedTransferClient {
+    pub(crate) async fn request_challenge(&self) -> Result<Challenge, irpc::Error> {
+        Ok(self.inner.rpc(RequestChallenge).await?.challenge)
+    }
+
+    pub(crate) async fn submit_offer(
+        &self,
+        offer: SubmitTargetedOffer,
+    ) -> Result<TargetedOfferResponse, irpc::Error> {
+        self.inner.rpc(offer).await
+    }
+
+    pub(crate) async fn deliver_authorization(
+        &self,
+        delivery: DeliverTargetedAuthorization,
+    ) -> Result<DeliverAuthorizationResponse, irpc::Error> {
+        self.inner.rpc(delivery).await
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct RequestChallenge;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct ChallengeResponse {
+    challenge: Challenge,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct SubmitTargetedOffer {
+    pub(crate) proof: WireProof,
+    pub(crate) generation: u64,
+    pub(crate) relationship_protocol_version: u16,
+    pub(crate) protocol_version: u16,
+    pub(crate) transfer_id: String,
+    pub(crate) sender_endpoint_id: String,
+    pub(crate) receiver_endpoint_id: String,
+    pub(crate) manifest_id: String,
+    pub(crate) content_hash: String,
+    pub(crate) transfer_name: String,
+    pub(crate) file_count: u64,
+    pub(crate) total_size: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum TargetedOfferResponse {
+    Accepted,
+    Declined { reason: String },
+    Refused { reason: String },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct DeliverTargetedAuthorization {
+    pub(crate) transfer_id: String,
+    pub(crate) authorization: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub(crate) enum DeliverAuthorizationResponse {
+    Stored,
+    Rejected,
+}
+
+#[rpc_requests(message = TargetedTransferMessage)]
+#[derive(Debug, Serialize, Deserialize)]
+enum TargetedTransferMessages {
+    #[rpc(tx = oneshot::Sender<ChallengeResponse>)]
+    RequestChallenge(RequestChallenge),
+    #[rpc(tx = oneshot::Sender<TargetedOfferResponse>)]
+    SubmitTargetedOffer(SubmitTargetedOffer),
+    #[rpc(tx = oneshot::Sender<DeliverAuthorizationResponse>)]
+    DeliverTargetedAuthorization(DeliverTargetedAuthorization),
+}
+
+/// Helper kept for type visibility in callers that map refuse reasons.
+#[allow(dead_code)]
+pub(crate) fn map_offer_error(reason: &str) -> VnidropError {
+    VnidropError::permission(anyhow::anyhow!("targeted offer refused: {reason}"))
+}

crates/vnidrop/src/tests.rs

@@ -38,6 +38,8 @@ mod secure_secret_tests;
 #[cfg(target_os = "windows")]
 #[path = "tests/secure_secret_windows.rs"]
 mod secure_secret_windows_tests;
+#[path = "tests/targeted_transfer.rs"]
+mod targeted_transfer_tests;
 #[path = "tests/ticket.rs"]
 mod ticket_tests;
 #[path = "tests/transfer_state.rs"]

crates/vnidrop/src/tests/targeted_transfer.rs

@@ -0,0 +1,471 @@
+use std::{
+    path::Path,
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+use crate::{
+    secure_secret::FaultInjectingSecretStore, CoreEvent, CoreEventSink, DeviceRelationshipState,
+    PendingTargetedOffer, ShareMetadataInput, ShareSource, SourceKind, TargetedTransferState,
+    TransferAccessMode, VnidropCore, VnidropError,
+};
+
+struct RecordingSink {
+    events: Mutex<Vec<CoreEvent>>,
+}
+
+impl CoreEventSink for RecordingSink {
+    fn on_event(&self, event: CoreEvent) {
+        self.events.lock().unwrap().push(event);
+    }
+}
+
+struct ProtectedNode {
+    _data_dir: tempfile::TempDir,
+    core: Arc<VnidropCore>,
+}
+
+impl ProtectedNode {
+    fn new() -> Self {
+        let data_dir = tempfile::tempdir().unwrap();
+        let sink = Arc::new(RecordingSink {
+            events: Mutex::new(Vec::new()),
+        });
+        let store = Arc::new(FaultInjectingSecretStore::default());
+        let core = VnidropCore::initialize_with_test_secret_store(
+            data_dir.path().to_string_lossy().into_owned(),
+            sink,
+            store,
+        )
+        .expect("protected test core");
+        Self {
+            _data_dir: data_dir,
+            core,
+        }
+    }
+}
+
+impl Drop for ProtectedNode {
+    fn drop(&mut self) {
+        self.core.shutdown();
+    }
+}
+
+fn share_path(core: &VnidropCore, source: &Path, transfer_id: u64) -> crate::ShareResult {
+    core.share_files(
+        vec![ShareSource {
+            kind: SourceKind::Path,
+            value: source.to_string_lossy().into_owned(),
+            display_name: Some("hello.txt".to_string()),
+            is_directory: false,
+        }],
+        ShareMetadataInput {
+            transfer_id,
+            transfer_name: Some("hello.txt".to_string()),
+            sender_name: Some("sender".to_string()),
+            access_mode: TransferAccessMode::ApprovalRequired,
+        },
+    )
+    .unwrap()
+}
+
+fn wait_for_receiver_request(sender: &VnidropCore, transfer_id: u64) -> crate::ReceiverRequest {
+    let started = Instant::now();
+    loop {
+        if let Some(request) = sender
+            .list_receiver_requests(transfer_id)
+            .unwrap()
+            .into_iter()
+            .find(|request| request.status == "requested")
+        {
+            return request;
+        }
+        assert!(
+            started.elapsed() < Duration::from_secs(15),
+            "timed out waiting for receiver request"
+        );
+        std::thread::sleep(Duration::from_millis(25));
+    }
+}
+
+fn complete_transfer(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_id: u64) {
+    let source_dir = tempfile::tempdir().unwrap();
+    let output_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("hello.txt");
+    std::fs::write(&source_path, b"mutual consent").unwrap();
+    let share = share_path(&sender.core, &source_path, transfer_id);
+    let output_dir = output_dir.path().to_string_lossy().to_string();
+    let receiver_core = receiver.core.clone();
+    let ticket = share.ticket.clone();
+    let handle = std::thread::spawn(move || {
+        receiver_core.receive(ticket, output_dir, Some("receiver".to_string()))
+    });
+    let request = wait_for_receiver_request(&sender.core, share.transfer_id);
+    sender
+        .core
+        .respond_receiver_request(request.id, true, None)
+        .unwrap();
+    handle.join().unwrap().unwrap();
+
+    let started = Instant::now();
+    let peer = receiver.core.status().endpoint_id.clone();
+    loop {
+        if sender
+            .core
+            .list_pairing_eligibilities()
+            .unwrap()
+            .iter()
+            .any(|entry| entry.peer_endpoint_id == peer)
+        {
+            break;
+        }
+        assert!(
+            started.elapsed() < Duration::from_secs(10),
+            "eligibility never appeared"
+        );
+        std::thread::sleep(Duration::from_millis(25));
+    }
+}
+
+fn wait_for_relationship(
+    core: &VnidropCore,
+    peer: &str,
+    state: DeviceRelationshipState,
+) -> crate::DeviceRelationship {
+    let started = Instant::now();
+    loop {
+        if let Some(relationship) = core
+            .list_device_relationships()
+            .unwrap()
+            .into_iter()
+            .find(|entry| entry.remote_endpoint_id == peer && entry.state == state)
+        {
+            return relationship;
+        }
+        assert!(
+            started.elapsed() < Duration::from_secs(15),
+            "relationship {peer} never reached {state:?}"
+        );
+        std::thread::sleep(Duration::from_millis(25));
+    }
+}
+
+fn establish_saved(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) {
+    let alice_id = alice.core.status().endpoint_id.clone();
+    let bob_id = bob.core.status().endpoint_id.clone();
+    complete_transfer(alice, bob, transfer_id);
+    assert!(alice
+        .core
+        .request_saved_device_pairing(bob_id.clone())
+        .unwrap());
+    wait_for_relationship(
+        &bob.core,
+        &alice_id,
+        DeviceRelationshipState::PendingIncoming,
+    );
+    assert!(bob
+        .core
+        .respond_to_device_pairing(alice_id.clone(), true)
+        .unwrap());
+    wait_for_relationship(&alice.core, &bob_id, DeviceRelationshipState::Saved);
+    wait_for_relationship(&bob.core, &alice_id, DeviceRelationshipState::Saved);
+}
+
+fn wait_for_pending_offer(core: &VnidropCore) -> PendingTargetedOffer {
+    let started = Instant::now();
+    loop {
+        let pending = core.list_pending_targeted_offers();
+        if let Some(offer) = pending.into_iter().next() {
+            return offer;
+        }
+        assert!(
+            started.elapsed() < Duration::from_secs(20),
+            "timed out waiting for pending targeted offer"
+        );
+        std::thread::sleep(Duration::from_millis(25));
+    }
+}
+
+fn targeted_source(path: &Path) -> ShareSource {
+    ShareSource {
+        kind: SourceKind::Path,
+        value: path.to_string_lossy().into_owned(),
+        display_name: Some("payload.txt".to_string()),
+        is_directory: false,
+    }
+}
+
+#[test]
+fn create_targeted_transfer_is_immutable_and_saved_only() {
+    let alice = ProtectedNode::new();
+    let bob = ProtectedNode::new();
+    let stranger = ProtectedNode::new();
+    let bob_id = bob.core.status().endpoint_id.clone();
+    let stranger_id = stranger.core.status().endpoint_id.clone();
+    establish_saved(&alice, &bob, 10_001);
+
+    let source_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("payload.txt");
+    std::fs::write(&source_path, b"immutable payload").unwrap();
+
+    let stranger_err = alice
+        .core
+        .create_targeted_transfer(
+            stranger_id,
+            vec![targeted_source(&source_path)],
+            Some("payload.txt".to_string()),
+        )
+        .unwrap_err();
+    assert!(matches!(stranger_err, VnidropError::Permission { .. }));
+
+    let bob_core = bob.core.clone();
+    let accept = std::thread::spawn(move || {
+        let offer = wait_for_pending_offer(&bob_core);
+        bob_core
+            .respond_to_targeted_offer(offer.transfer_id, true)
+            .unwrap()
+    });
+
+    let transfer = alice
+        .core
+        .create_targeted_transfer(
+            bob_id.clone(),
+            vec![targeted_source(&source_path)],
+            Some("payload.txt".to_string()),
+        )
+        .unwrap();
+    let _auth = accept.join().unwrap().expect("authorization after approve");
+
+    assert_eq!(transfer.sender_endpoint_id, alice.core.status().endpoint_id);
+    assert_eq!(transfer.receiver_endpoint_id, bob_id);
+    assert_eq!(transfer.file_count, 1);
+    assert_eq!(transfer.total_size, b"immutable payload".len() as u64);
+    assert!(!transfer.id.is_empty());
+    assert!(!transfer.manifest_id.is_empty());
+    assert!(matches!(
+        transfer.state,
+        TargetedTransferState::Approved
+            | TargetedTransferState::Connecting
+            | TargetedTransferState::Transferring
+            | TargetedTransferState::Completed
+    ));
+
+    let listed = alice
+        .core
+        .get_targeted_transfer(transfer.id.clone())
+        .unwrap();
+    let listed = listed.expect("durable targeted transfer");
+    assert_eq!(listed.id, transfer.id);
+    assert_eq!(listed.sender_endpoint_id, transfer.sender_endpoint_id);
+    assert_eq!(listed.receiver_endpoint_id, transfer.receiver_endpoint_id);
+    assert_eq!(listed.manifest_id, transfer.manifest_id);
+    assert_eq!(listed.file_count, transfer.file_count);
+    assert_eq!(listed.total_size, transfer.total_size);
+}
+
+#[test]
+fn preapproval_offer_is_authenticated_without_ordinary_share_ticket() {
+    let alice = ProtectedNode::new();
+    let bob = ProtectedNode::new();
+    let bob_id = bob.core.status().endpoint_id.clone();
+    establish_saved(&alice, &bob, 10_010);
+
+    let source_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("payload.txt");
+    std::fs::write(&source_path, b"offer body").unwrap();
+
+    let bob_core = bob.core.clone();
+    let accept = std::thread::spawn(move || {
+        let offer = wait_for_pending_offer(&bob_core);
+        // Offer surfaces identity + manifest summary only — never a reusable ticket.
+        assert!(!offer.transfer_id.is_empty());
+        assert_eq!(offer.sender_endpoint_id, alice_id_from(&bob_core));
+        assert_eq!(offer.file_count, 1);
+        assert_eq!(offer.total_size, b"offer body".len() as u64);
+        assert!(!offer.manifest_id.is_empty());
+        assert!(!offer.content_hash.is_empty());
+        assert!(offer.protocol_version >= 1);
+        bob_core
+            .respond_to_targeted_offer(offer.transfer_id, true)
+            .unwrap()
+    });
+
+    alice
+        .core
+        .create_targeted_transfer(
+            bob_id,
+            vec![targeted_source(&source_path)],
+            Some("payload.txt".to_string()),
+        )
+        .unwrap();
+    accept.join().unwrap().unwrap();
+}
+
+fn alice_id_from(bob: &VnidropCore) -> String {
+    bob.list_saved_devices()
+        .unwrap()
+        .into_iter()
+        .next()
+        .expect("saved alice")
+        .endpoint_id
+}
+
+#[test]
+fn invalid_offer_never_becomes_observable_pending_approval() {
+    let alice = ProtectedNode::new();
+    let bob = ProtectedNode::new();
+    // No Saved relationship — offer must not surface.
+    let bob_id = bob.core.status().endpoint_id.clone();
+    let source_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("payload.txt");
+    std::fs::write(&source_path, b"nope").unwrap();
+
+    let err = alice
+        .core
+        .create_targeted_transfer(
+            bob_id,
+            vec![targeted_source(&source_path)],
+            Some("payload.txt".to_string()),
+        )
+        .unwrap_err();
+    assert!(matches!(err, VnidropError::Permission { .. }));
+    assert!(bob.core.list_pending_targeted_offers().is_empty());
+}
+
+#[test]
+fn explicit_approval_gates_content_and_binds_authorization_to_receiver() {
+    let alice = ProtectedNode::new();
+    let bob = ProtectedNode::new();
+    let charlie = ProtectedNode::new();
+    let bob_id = bob.core.status().endpoint_id.clone();
+    establish_saved(&alice, &bob, 10_020);
+
+    let source_dir = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("payload.txt");
+    let payload = b"bound authorization payload";
+    std::fs::write(&source_path, payload).unwrap();
+
+    let bob_core = bob.core.clone();
+    let offer_id = Arc::new(Mutex::new(None::<String>));
+    let offer_id_setter = offer_id.clone();
+    let gate = Arc::new(Mutex::new(false));
+    let gate_wait = gate.clone();
+    let accept = std::thread::spawn(move || {
+        let offer = wait_for_pending_offer(&bob_core);
+        *offer_id_setter.lock().unwrap() = Some(offer.transfer_id.clone());
+        // Hold approval until the main thread asserts that content is unavailable.
+        let started = Instant::now();
+        loop {
+            if *gate_wait.lock().unwrap() {
+                break;
+            }
+            assert!(
+                started.elapsed() < Duration::from_secs(20),
+                "approval gate never opened"
+            );
+            std::thread::sleep(Duration::from_millis(25));
+        }
+        bob_core
+            .respond_to_targeted_offer(offer.transfer_id, true)
+            .unwrap()
+    });
+
+    let alice_core = alice.core.clone();
+    let source = targeted_source(&source_path);
+    let create = std::thread::spawn(move || {
+        alice_core.create_targeted_transfer(bob_id, vec![source], Some("payload.txt".to_string()))
+    });
+
+    let started = Instant::now();
+    let transfer_id = loop {
+        if let Some(id) = offer_id.lock().unwrap().clone() {
+            break id;
+        }
+        assert!(
+            started.elapsed() < Duration::from_secs(20),
+            "offer id never observed"
+        );
+        std::thread::sleep(Duration::from_millis(25));
+    };
+
+    let _ = transfer_id;
+    // Without an approved authorization, receive must fail — no content yet.
+    let early_receive = bob.core.receive_targeted_transfer(
+        "not-a-real-authorization".to_string(),
+        tempfile::tempdir()
+            .unwrap()
+            .path()
+            .to_string_lossy()
+            .into_owned(),
+    );
+    assert!(early_receive.is_err());
+
+    *gate.lock().unwrap() = true;
+    let auth = accept.join().unwrap().expect("receiver authorization");
+    create.join().unwrap().unwrap();
+
+    let output = tempfile::tempdir().unwrap();
+    bob.core
+        .receive_targeted_transfer(auth.clone(), output.path().to_string_lossy().into_owned())
+        .unwrap();
+    assert_eq!(
+        std::fs::read(output.path().join("payload.txt")).unwrap(),
+        payload
+    );
+
+    let charlie_output = tempfile::tempdir().unwrap();
+    let leaked = charlie
+        .core
+        .receive_targeted_transfer(auth, charlie_output.path().to_string_lossy().into_owned());
+    assert!(
+        leaked.is_err(),
+        "leaked authorization must not authorize another endpoint"
+    );
+}
+
+#[test]
+fn invitation_multi_receiver_shares_remain_independently_authorized() {
+    let source_dir = tempfile::tempdir().unwrap();
+    let first_output = tempfile::tempdir().unwrap();
+    let second_output = tempfile::tempdir().unwrap();
+    let source_path = source_dir.path().join("shared.txt");
+    std::fs::write(&source_path, b"shared with both receivers").unwrap();
+    let sender = ProtectedNode::new();
+    let first_receiver = ProtectedNode::new();
+    let second_receiver = ProtectedNode::new();
+    let share = sender
+        .core
+        .share_files(
+            vec![ShareSource {
+                kind: SourceKind::Path,
+                value: source_path.to_string_lossy().into_owned(),
+                display_name: Some("shared.txt".to_string()),
+                is_directory: false,
+            }],
+            ShareMetadataInput {
+                transfer_id: 10_030,
+                transfer_name: Some("Existing share".to_string()),
+                sender_name: Some("Sender".to_string()),
+                access_mode: TransferAccessMode::Public,
+            },
+        )
+        .unwrap();
+
+    for (receiver, output) in [
+        (&first_receiver, first_output.path()),
+        (&second_receiver, second_output.path()),
+    ] {
+        receiver
+            .core
+            .receive(
+                share.ticket.clone(),
+                output.to_string_lossy().into_owned(),
+                Some("Receiver".to_string()),
+            )
+            .unwrap();
+        assert_eq!(
+            std::fs::read(output.join("shared.txt")).unwrap(),
+            b"shared with both receivers"
+        );
+    }
+}