What secure communication protocols prevent embedding inversion during multi-agent RAG?

**Date of Record:** 2026 October 27 **Scope:** Applies when autonomous agents communicate via shared RAG (Retrieval-Augmented Generation) pipelines, exchanging semantic embeddings of user queries or retrieved documents, and the underlying content is sensitive (PII, credentials, or proprietary data). Does not apply to agents operating entirely on public data, or pipelines where the embedding itself is the intended payload.

Core Distinction

The separation between **semantic utility** (the embedding retains enough information for the receiving agent to perform retrieval) and **inversion resistance** (the embedding cannot be used to reconstruct the original text). Relying solely on semantic utility creates the **Shared Context Leak**: an attacker with access to the embeddings can use inversion models to recover the original text, breaking the privacy of the agents' communication. Inversion resistance requires differential privacy noise, bounded to the retrieval task.

What Was Observed

In multi-agent RAG pipelines, failing to prevent embedding inversion produces three primary failure modes:

  1. **PII Reconstruction:** Agent A embeds a user query containing SSN and sends the embedding to Agent B for retrieval. Agent B (or an eavesdropper) inverts the embedding and recovers the SSN.
  2. **Credential Extraction:** Agent A embeds a code snippet containing an API key. The embedding is shared with Agent B. Inversion recovers the key.
  3. **Model Distillation Theft:** Agent A shares embeddings from a proprietary model. An attacker collects embeddings and trains a surrogate model, stealing the proprietary model's capabilities.

Recorded Elements

  1. **`embedding_sensitivity_class`:** The classification of the underlying content (PUBLIC, PII, CREDENTIAL, PROPRIETARY), determining the required noise level.
  2. **`differential_privacy_epsilon`:** The exact epsilon parameter of the noise injected into the embedding, bounding the inversion risk.
  3. **`reconstruction_resistance_receipt`:** The cryptographic proof that the embedding was noise-injected with the required epsilon before transmission.
  4. **`retrieval_utility_guard_hash`:** The cryptographic proof that the noise level does not degrade retrieval utility beyond the acceptable threshold (e.g., recall@10 drops by <5%).

Condition of Validity

* **Applies when:** Embeddings are exchanged between agents, the underlying content is sensitive, and the receiving agent or an eavesdropper could perform inversion attacks. * **Does not apply when:** The embeddings are strictly from public data (no privacy risk), or the agents share a trusted hardware enclave where inversion is computationally infeasible.

Decision Boundary

  1. **Is the underlying content sensitive (PII, credentials, proprietary)?** If no, standard embedding transmission applies. If yes, proceed.
  2. **Is `differential_privacy_epsilon` recorded and appropriate for the sensitivity class?** If no, the embedding is vulnerable to inversion. The agent must add noise before transmission.
  3. **Does the `reconstruction_resistance_receipt` prove the noise was applied?** If no, the agent must re-apply noise and re-verify.
  4. **Conversely, when can an agent transmit embeddings without differential privacy?** Only when the receiving agent is verified to be within the same trust boundary (same enclave, same organization with contractual privacy), or the underlying content is strictly public.

Operational Effect

  1. **Inversion Resistance:** Protects sensitive content from reconstruction attacks by injecting bounded noise.
  2. **Retrieval Utility Preservation:** Ensures the noise does not degrade retrieval quality beyond acceptable thresholds.
  3. **Compliance:** Provides cryptographic proof that embeddings were noise-injected before transmission, satisfying privacy regulations.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class EmbeddingSensitivity(Enum):
    PUBLIC = auto()
    PII = auto()
    CREDENTIAL = auto()
    PROPRIETARY = auto()

class InversionVerdict(Enum):
    SAFE_TRANSMISSION = auto()
    NOISE_REQUIRED = auto()
    UTILITY_DEGRADED = auto()
    TRUST_BOUNDARY_TRUSTED = auto()

@dataclass(frozen=True)
class EmbeddingPayload:
    embedding_hash: str
    sensitivity: EmbeddingSensitivity
    dp_epsilon: float | None
    noise_applied: bool
    retrieval_recall_delta: float

@dataclass(frozen=True)
class EmbeddingPolicy:
    max_epsilon_for_pii: float
    max_epsilon_for_credentials: float
    min_recall_delta: float
    peer_in_same_trust_boundary: bool

@dataclass(frozen=True)
class InversionManifest:
    verdict: InversionVerdict
    directive: str
    manifest_digest: str

class EmbeddingInversionGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_embedding(
        cls, payload: EmbeddingPayload, policy: EmbeddingPolicy
    ) -> InversionManifest:
        
        # Gate 1: Public or trusted peer
        if payload.sensitivity == EmbeddingSensitivity.PUBLIC:
            return InversionManifest(
                verdict=InversionVerdict.SAFE_TRANSMISSION,
                directive="PROCEED: Public content, no privacy risk.",
                manifest_digest=cls._hash(f"PUBLIC:{payload.embedding_hash}")
            )

        if policy.peer_in_same_trust_boundary:
            return InversionManifest(
                verdict=InversionVerdict.TRUST_BOUNDARY_TRUSTED,
                directive="PROCEED: Peer is within the same trust boundary.",
                manifest_digest=cls._hash(f"TRUSTED:{payload.embedding_hash}")
            )

        # Gate 2: Noise requirement
        if not payload.noise_applied or payload.dp_epsilon is None:
            return InversionManifest(
                verdict=InversionVerdict.NOISE_REQUIRED,
                directive=f"HALT: Sensitivity {payload.sensitivity.name} requires differential privacy noise.",
                manifest_digest=cls._hash(f"NO_NOISE:{payload.embedding_hash}")
            )

        # Gate 3: Epsilon threshold
        if payload.sensitivity == EmbeddingSensitivity.PII and payload.dp_epsilon > policy.max_epsilon_for_pii:
            return InversionManifest(
                verdict=InversionVerdict.NOISE_REQUIRED,
                directive=f"HALT: Epsilon {payload.dp_epsilon} > max for PII {policy.max_epsilon_for_pii}.",
                manifest_digest=cls._hash(f"EPSILON_PII:{payload.embedding_hash}")
            )

        if payload.sensitivity == EmbeddingSensitivity.CREDENTIAL and payload.dp_epsilon > policy.max_epsilon_for_credentials:
            return InversionManifest(
                verdict=InversionVerdict.NOISE_REQUIRED,
                directive=f"HALT: Epsilon {payload.dp_epsilon} > max for credentials {policy.max_epsilon_for_credentials}.",
                manifest_digest=cls._hash(f"EPSILON_CRED:{payload.embedding_hash}")
            )

        # Gate 4: Utility check
        if payload.retrieval_recall_delta < policy.min_recall_delta:
            return InversionManifest(
                verdict=InversionVerdict.UTILITY_DEGRADED,
                directive=f"HALT: Recall delta {payload.retrieval_recall_delta:.2f} < min {policy.min_recall_delta:.2f}. Noise too high.",
                manifest_digest=cls._hash(f"UTILITY:{payload.embedding_hash}")
            )

        # Gate 5: Safe
        digest = cls._hash(json.dumps({
            "embedding": payload.embedding_hash,
            "sensitivity": payload.sensitivity.name,
            "epsilon": str(payload.dp_epsilon),
            "recall_delta": f"{payload.retrieval_recall_delta:.3f}"
        }, sort_keys=True))
        
        return InversionManifest(
            verdict=InversionVerdict.SAFE_TRANSMISSION,
            directive=f"PROCEED: DP noise applied (epsilon={payload.dp_epsilon}) with acceptable utility.",
            manifest_digest=digest
        )