diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8ac6d4db..13016f16 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- **Application-generated EQL values in statements**: SQL literals and bound parameters may now carry EQL v3 storage payloads or query-only SEM operands produced by an application. Proxy authenticates stored ciphertext, requires its authenticated descriptor to name the inferred destination column, and independently re-derives every SEM term before forwarding it without double encryption. Query-only operands contain no ciphertext to authenticate, so Proxy instead validates their version, identifier, term shape, column capabilities, and syntactic query role; they are rejected in storage positions. This includes bare SteVec selector hashes matching `^[0-9a-f]{32}$`: in JSON selector query positions a match is treated as already hashed, while a non-match remains plaintext and is encrypted normally. A matching plaintext selector is inherently ambiguous and is intentionally treated as already hashed. Invalid payloads fail closed with one generic error so validation details cannot be used as an oracle.
+
 ## [3.0.1] - 2026-08-05
 
 ### Added
diff --git a/Cargo.lock b/Cargo.lock
index 269cfe0d..88163a26 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4308,6 +4308,8 @@ checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
 name = "showcase"
 version = "3.0.1"
 dependencies = [
+ "cipherstash-client",
+ "cipherstash-config",
  "rand 0.9.2",
  "rustls",
  "serde",
diff --git a/packages/cipherstash-proxy-integration/src/inbound_ciphertext.rs b/packages/cipherstash-proxy-integration/src/inbound_ciphertext.rs
new file mode 100644
index 00000000..84356665
--- /dev/null
+++ b/packages/cipherstash-proxy-integration/src/inbound_ciphertext.rs
@@ -0,0 +1,394 @@
+//! End-to-end coverage for application-encrypted EQL payloads entering Proxy.
+
+#[cfg(test)]
+mod tests {
+    use crate::common::{clear_with_client, connect_with_tls, random_id, PROXY};
+    use cipherstash_client::{
+        encryption::{Plaintext, QueryOp, ScopedCipher},
+        eql::{
+            encrypt_eql_v3, EqlCiphertextV3, EqlEncryptOpts, EqlOperation, EqlOutputV3, Identifier,
+            PreparedPlaintext,
+        },
+        schema::{column::Index, ColumnConfig, ColumnType},
+        zerokms::{ClientKey, ZeroKMSBuilder},
+        AutoStrategy, IdentifiedBy,
+    };
+    use cipherstash_config::column::{ArrayIndexMode, IndexType, SteVecMode};
+    use std::{borrow::Cow, sync::Arc};
+    use uuid::Uuid;
+
+    async fn cipher() -> Arc<ScopedCipher<AutoStrategy>> {
+        let client_id = env("CS_CLIENT_ID", "CS_ENCRYPT__CLIENT_ID")
+            .parse()
+            .expect("CS_CLIENT_ID must be a UUID");
+        let client_key =
+            ClientKey::from_hex_v1(client_id, &env("CS_CLIENT_KEY", "CS_ENCRYPT__CLIENT_KEY"))
+                .expect("CS_CLIENT_KEY must be valid");
+        let zerokms = ZeroKMSBuilder::auto()
+            .expect("ZeroKMS credentials must be configured")
+            .with_client_key(client_key)
+            .build()
+            .expect("ZeroKMS client must initialize");
+        let keyset_id: Uuid = env("CS_DEFAULT_KEYSET_ID", "CS_ENCRYPT__DEFAULT_KEYSET_ID")
+            .parse()
+            .expect("CS_DEFAULT_KEYSET_ID must be a UUID");
+        Arc::new(
+            ScopedCipher::init(Arc::new(zerokms), Some(IdentifiedBy::Uuid(keyset_id)))
+                .await
+                .expect("scoped cipher must initialize"),
+        )
+    }
+
+    fn env(primary: &str, nested: &str) -> String {
+        std::env::var(primary)
+            .or_else(|_| std::env::var(nested))
+            .unwrap_or_else(|_| panic!("{primary} must be configured"))
+    }
+
+    fn text_search_config(table: &str, column: &str) -> ColumnConfig {
+        ColumnConfig::build(format!("{table}/{column}"))
+            .casts_as(ColumnType::Text)
+            .add_index(Index::new_unique())
+            .add_index(Index::new_ope())
+            .add_index(Index::new_match())
+    }
+
+    fn json_search_config(table: &str, column: &str) -> ColumnConfig {
+        ColumnConfig::build(format!("{table}/{column}"))
+            .casts_as(ColumnType::Json)
+            .add_index(Index::new(IndexType::SteVec {
+                prefix: format!("{table}/{column}"),
+                term_filters: Vec::new(),
+                array_index_mode: ArrayIndexMode::ALL,
+                mode: SteVecMode::default(),
+            }))
+    }
+
+    async fn encrypt_text(table: &str, column: &str, plaintext: &str) -> String {
+        let prepared = PreparedPlaintext::new(
+            Cow::Owned(text_search_config(table, column)),
+            Identifier::new(table, column),
+            Plaintext::from(plaintext),
+            EqlOperation::Store,
+        );
+        let mut outputs =
+            encrypt_eql_v3(cipher().await, vec![prepared], &EqlEncryptOpts::default())
+                .await
+                .expect("application-side encryption must succeed");
+        let EqlOutputV3::Store(ciphertext) = outputs.remove(0) else {
+            panic!("store encryption must return a stored payload");
+        };
+        serde_json::to_string(&ciphertext).unwrap()
+    }
+
+    async fn query_text(table: &str, column: &str, plaintext: &str) -> String {
+        let stored: EqlCiphertextV3 =
+            serde_json::from_str(&encrypt_text(table, column, plaintext).await).unwrap();
+        serde_json::to_string(&stored.into_query_operand()).unwrap()
+    }
+
+    async fn query_json(
+        table: &str,
+        column: &str,
+        plaintext: serde_json::Value,
+    ) -> serde_json::Value {
+        let config = json_search_config(table, column);
+        let index_type = config.indexes[0].index_type.clone();
+        let prepared = PreparedPlaintext::new(
+            Cow::Owned(config),
+            Identifier::new(table, column),
+            Plaintext::Json(Some(plaintext)),
+            EqlOperation::Query(&index_type, QueryOp::Default),
+        );
+        let mut outputs =
+            encrypt_eql_v3(cipher().await, vec![prepared], &EqlEncryptOpts::default())
+                .await
+                .expect("application-side query encryption must succeed");
+        let EqlOutputV3::Query(query) = outputs.remove(0) else {
+            panic!("query encryption must return a query-only payload");
+        };
+        serde_json::to_value(query).unwrap()
+    }
+
+    async fn query_json_selector(table: &str, column: &str, path: &str) -> String {
+        let config = json_search_config(table, column);
+        let index_type = config.indexes[0].index_type.clone();
+        let prepared = PreparedPlaintext::new(
+            Cow::Owned(config),
+            Identifier::new(table, column),
+            Plaintext::from(path),
+            EqlOperation::Query(&index_type, QueryOp::SteVecSelector),
+        );
+        let mut outputs =
+            encrypt_eql_v3(cipher().await, vec![prepared], &EqlEncryptOpts::default())
+                .await
+                .expect("application-side selector encryption must succeed");
+        let EqlOutputV3::Query(query) = outputs.remove(0) else {
+            panic!("selector encryption must return a query-only payload");
+        };
+        let serde_json::Value::String(selector) = serde_json::to_value(query).unwrap() else {
+            panic!("selector encryption must return a bare selector hash");
+        };
+        selector
+    }
+
+    #[tokio::test]
+    async fn accepts_pre_encrypted_parameter_for_storage_and_search() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "encrypted in the application";
+        let payload = encrypt_text("encrypted", "encrypted_text", plaintext).await;
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .unwrap();
+
+        let rows = client
+            .query(
+                "SELECT encrypted_text FROM encrypted WHERE encrypted_text = $1",
+                &[&payload],
+            )
+            .await
+            .unwrap();
+        assert_eq!(rows[0].get::<_, String>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn accepts_pre_encrypted_literal_for_storage() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "application encrypted literal";
+        let payload = encrypt_text("encrypted", "encrypted_text", plaintext).await;
+        let payload = payload.replace('\'', "''");
+
+        client
+            .simple_query(&format!(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ({id}, '{payload}')"
+            ))
+            .await
+            .unwrap();
+
+        let row = client
+            .query_one("SELECT encrypted_text FROM encrypted WHERE id = $1", &[&id])
+            .await
+            .unwrap();
+        assert_eq!(row.get::<_, String>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn accepts_query_only_parameter_for_search() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "queried with application SEM terms";
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &plaintext],
+            )
+            .await
+            .unwrap();
+
+        let payload = query_text("encrypted", "encrypted_text", plaintext).await;
+        let rows = client
+            .query(
+                "SELECT encrypted_text FROM encrypted WHERE encrypted_text = $1",
+                &[&payload],
+            )
+            .await
+            .unwrap();
+        assert_eq!(rows[0].get::<_, String>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn accepts_query_only_literal_for_search() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = "queried with literal SEM terms";
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &plaintext],
+            )
+            .await
+            .unwrap();
+
+        let payload = query_text("encrypted", "encrypted_text", plaintext)
+            .await
+            .replace('\'', "''");
+        let rows = client
+            .simple_query(&format!(
+                "SELECT encrypted_text FROM encrypted WHERE encrypted_text = '{payload}'"
+            ))
+            .await
+            .unwrap();
+        let row = rows
+            .iter()
+            .find_map(|message| match message {
+                tokio_postgres::SimpleQueryMessage::Row(row) => Some(row),
+                _ => None,
+            })
+            .expect("query-only literal must match one row");
+        assert_eq!(row.get(0), Some(plaintext));
+    }
+
+    #[tokio::test]
+    async fn rejects_query_only_parameter_for_storage() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let payload = query_text("encrypted", "encrypted_text", "not writable").await;
+
+        let error = client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .expect_err("query-only payloads must not be accepted for storage");
+        assert_eq!(
+            error.as_db_error().unwrap().message(),
+            "Invalid encrypted value"
+        );
+    }
+
+    #[tokio::test]
+    async fn accepts_query_only_ste_vec_parameter_for_json_search() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = serde_json::json!({
+            "patient": { "name": "Ada Lovelace" },
+            "active": true
+        });
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_jsonb) VALUES ($1, $2)",
+                &[&id, &plaintext],
+            )
+            .await
+            .unwrap();
+
+        let payload = query_json("encrypted", "encrypted_jsonb", plaintext.clone()).await;
+        let rows = client
+            .query(
+                "SELECT encrypted_jsonb FROM encrypted WHERE encrypted_jsonb @> $1",
+                &[&payload],
+            )
+            .await
+            .unwrap();
+        assert_eq!(rows[0].get::<_, serde_json::Value>(0), plaintext);
+    }
+
+    #[tokio::test]
+    async fn accepts_bare_selector_hashes_as_parameters_and_literals() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let plaintext = serde_json::json!({
+            "patient": { "name": "Ada Lovelace" }
+        });
+
+        client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_jsonb) VALUES ($1, $2)",
+                &[&id, &plaintext],
+            )
+            .await
+            .unwrap();
+
+        let selector = query_json_selector("encrypted", "encrypted_jsonb", "$.patient.name").await;
+        let row = client
+            .query_one(
+                "SELECT encrypted_jsonb -> $1 FROM encrypted WHERE id = $2",
+                &[&selector, &id],
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            row.get::<_, serde_json::Value>(0),
+            serde_json::json!("Ada Lovelace")
+        );
+
+        let row = client
+            .simple_query(&format!(
+                "SELECT jsonb_path_query_first(encrypted_jsonb, '{selector}') \
+                 FROM encrypted WHERE id = '{id}'"
+            ))
+            .await
+            .unwrap()
+            .into_iter()
+            .find_map(|message| match message {
+                tokio_postgres::SimpleQueryMessage::Row(row) => row.get(0).map(str::to_owned),
+                _ => None,
+            })
+            .expect("bare selector literal must return an extracted value");
+        assert_eq!(row, "\"Ada Lovelace\"");
+    }
+
+    #[tokio::test]
+    async fn rejects_payload_for_a_different_destination_with_generic_error() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let payload = encrypt_text("some_other_table", "encrypted_text", "secret").await;
+        let mut payload: serde_json::Value = serde_json::from_str(&payload).unwrap();
+        payload["i"]["t"] = "encrypted".into();
+        let payload = payload.to_string();
+
+        let error = client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .expect_err("destination mismatch must fail closed");
+        assert_eq!(
+            error.as_db_error().unwrap().message(),
+            "Invalid encrypted value"
+        );
+    }
+
+    #[tokio::test]
+    async fn rejects_sem_terms_spliced_from_another_plaintext() {
+        let client = connect_with_tls(*PROXY).await;
+        clear_with_client(&client).await;
+        let id = random_id();
+        let x: serde_json::Value = serde_json::from_str(
+            &encrypt_text("encrypted", "encrypted_text", "indexed as x").await,
+        )
+        .unwrap();
+        let mut y: serde_json::Value = serde_json::from_str(
+            &encrypt_text("encrypted", "encrypted_text", "decrypts as y").await,
+        )
+        .unwrap();
+        for term in ["hm", "bf", "ob", "op"] {
+            if let Some(value) = x.get(term) {
+                y[term] = value.clone();
+            }
+        }
+        let payload = y.to_string();
+
+        let error = client
+            .execute(
+                "INSERT INTO encrypted (id, encrypted_text) VALUES ($1, $2)",
+                &[&id, &payload],
+            )
+            .await
+            .expect_err("spliced SEM terms must fail closed");
+        assert_eq!(
+            error.as_db_error().unwrap().message(),
+            "Invalid encrypted value"
+        );
+    }
+}
diff --git a/packages/cipherstash-proxy-integration/src/lib.rs b/packages/cipherstash-proxy-integration/src/lib.rs
index 756ab8d2..d2a713d5 100644
--- a/packages/cipherstash-proxy-integration/src/lib.rs
+++ b/packages/cipherstash-proxy-integration/src/lib.rs
@@ -7,6 +7,7 @@ mod empty_result;
 mod encryption_sanity;
 mod eql_regression;
 mod extended_protocol_error_messages;
+mod inbound_ciphertext;
 mod insert;
 mod legacy_v2_column;
 mod map_concat;
diff --git a/packages/cipherstash-proxy/src/error.rs b/packages/cipherstash-proxy/src/error.rs
index dffc1493..45904faa 100644
--- a/packages/cipherstash-proxy/src/error.rs
+++ b/packages/cipherstash-proxy/src/error.rs
@@ -75,6 +75,7 @@ impl Error {
             // stores plaintext in a column its operator believes is encrypted
             // (CIP-3688). No configuration may turn that back on.
             Error::Mapping(MappingError::UnmappableEncryptedColumn { .. })
+                | Error::Encrypt(EncryptError::InvalidInboundCiphertext)
         )
     }
 }
@@ -255,6 +256,12 @@ pub enum TlsConfigError {
 
 #[derive(Error, Debug)]
 pub enum EncryptError {
+    /// Deliberately contains no payload or validation detail: inbound
+    /// ciphertext failures are attacker-controlled and detailed responses can
+    /// become an oracle.
+    #[error("Invalid encrypted value")]
+    InvalidInboundCiphertext,
+
     #[error(transparent)]
     CiphertextCouldNotBeSerialised(#[from] serde_json::Error),
 
diff --git a/packages/cipherstash-proxy/src/postgresql/context/mod.rs b/packages/cipherstash-proxy/src/postgresql/context/mod.rs
index d42e015c..4bb75a06 100644
--- a/packages/cipherstash-proxy/src/postgresql/context/mod.rs
+++ b/packages/cipherstash-proxy/src/postgresql/context/mod.rs
@@ -753,6 +753,12 @@ where
         plaintexts: Vec<Option<cipherstash_client::encryption::Plaintext>>,
         columns: &[Option<Column>],
     ) -> Result<Vec<Option<crate::EqlOutput>>, Error> {
+        if plaintexts.iter().all(Option::is_none) {
+            return Ok(std::iter::repeat_with(|| None)
+                .take(plaintexts.len())
+                .collect());
+        }
+
         let keyset_id = self.keyset_identifier();
 
         self.encryption
@@ -1117,6 +1123,18 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn empty_plaintext_batch_does_not_call_encryption_service() {
+        let context = create_context();
+        let output = context
+            .encrypt(vec![None, None], &[None, None])
+            .await
+            .unwrap();
+
+        assert_eq!(output.len(), 2);
+        assert!(output.iter().all(Option::is_none));
+    }
+
     fn statement() -> Statement {
         Statement {
             param_columns: vec![],
diff --git a/packages/cipherstash-proxy/src/postgresql/frontend.rs b/packages/cipherstash-proxy/src/postgresql/frontend.rs
index 0e543642..7a0a762c 100644
--- a/packages/cipherstash-proxy/src/postgresql/frontend.rs
+++ b/packages/cipherstash-proxy/src/postgresql/frontend.rs
@@ -11,7 +11,7 @@ use super::parser::SqlParser;
 use super::protocol::{self};
 use crate::connect::Sender;
 use crate::error::{EncryptError, Error, MappingError};
-use crate::log::{MAPPER, PROTOCOL};
+use crate::log::{ENCRYPT, MAPPER, PROTOCOL};
 use crate::postgresql::context::column::Column;
 use crate::postgresql::context::statement::{
     output_params_from_plan, OutputParam, OutputParamSource,
@@ -21,6 +21,7 @@ use crate::postgresql::context::Portal;
 use crate::postgresql::data::{
     compose_json_selector_path, json_value_selector_plaintext, literal_from_sql, literal_json_value,
 };
+use crate::postgresql::inbound_eql;
 use crate::postgresql::messages::close::Close;
 use crate::postgresql::messages::error_response::ErrorResponseCode;
 use crate::postgresql::messages::ready_for_query::ReadyForQuery;
@@ -648,7 +649,23 @@ where
             return Ok(vec![]);
         }
 
-        let plaintexts = literals_to_plaintext(typed_statement, literal_columns)?;
+        let inbound = literal_values
+            .iter()
+            .zip(literal_columns)
+            .map(|((_, literal), column)| {
+                let (Some(column), Some(value)) = (column, (*literal).clone().into_string()) else {
+                    return Ok(None);
+                };
+                inbound_eql::parse(
+                    value.as_bytes(),
+                    column,
+                    literal_is_query_operand(typed_statement, literal, column),
+                )
+                .map_err(Error::from)
+            })
+            .collect::<Result<Vec<_>, Error>>()?;
+        let skip = inbound.iter().map(Option::is_some).collect::<Vec<_>>();
+        let plaintexts = literals_to_plaintext_skipping(typed_statement, literal_columns, &skip)?;
 
         let start = Instant::now();
 
@@ -660,11 +677,18 @@ where
                 counter!(ENCRYPTION_ERROR_TOTAL).increment(1);
             })?;
 
-        for ((_, literal), encrypted) in literal_values.iter().zip(encrypted.iter_mut()) {
-            project_query_operand(
-                typed_statement.query_operands.contains_literal(literal),
-                encrypted,
-            );
+        self.merge_inbound_eql(&mut encrypted, inbound, literal_columns)
+            .await?;
+
+        for (((_, literal), column), encrypted) in literal_values
+            .iter()
+            .zip(literal_columns)
+            .zip(encrypted.iter_mut())
+        {
+            let query_operand = column
+                .as_ref()
+                .is_some_and(|column| literal_is_query_operand(typed_statement, literal, column));
+            project_query_operand(query_operand, encrypted);
         }
 
         debug!(target: MAPPER,
@@ -1156,8 +1180,13 @@ where
         bind: &Bind,
         statement: &Statement,
     ) -> Result<Vec<Option<crate::EqlOutput>>, Error> {
-        let plaintexts =
-            bind.to_plaintext(&statement.output_params, &statement.postgres_param_types)?;
+        let inbound = bind.inbound_eql(&statement.output_params)?;
+        let skip = inbound.iter().map(Option::is_some).collect::<Vec<_>>();
+        let plaintexts = bind.to_plaintext_skipping(
+            &statement.output_params,
+            &statement.postgres_param_types,
+            &skip,
+        )?;
 
         // Encryption is positional over the OUTPUT params — the values actually
         // sent — not over what the client bound.
@@ -1179,6 +1208,9 @@ where
                 counter!(ENCRYPTION_ERROR_TOTAL).increment(1);
             })?;
 
+        self.merge_inbound_eql(&mut encrypted, inbound, &output_param_columns)
+            .await?;
+
         for (output, encrypted) in statement.output_params.iter().zip(encrypted.iter_mut()) {
             project_query_operand(output.query_operand, encrypted);
         }
@@ -1205,6 +1237,93 @@ where
         Ok(encrypted)
     }
 
+    /// Merge application-generated EQL values into the encryption output.
+    /// Stored payloads are authenticated and independently verified. Query-only
+    /// payloads have no ciphertext to authenticate and are accepted only after
+    /// role-aware structural validation in `inbound_eql::parse`.
+    async fn merge_inbound_eql(
+        &self,
+        encrypted: &mut [Option<EqlOutput>],
+        inbound: Vec<Option<inbound_eql::InboundEql>>,
+        columns: &[Option<Column>],
+    ) -> Result<(), Error> {
+        let mut positions = Vec::new();
+        for (index, (payload, column)) in inbound.into_iter().zip(columns).enumerate() {
+            match payload {
+                Some(inbound_eql::InboundEql::Query(query)) => {
+                    encrypted[index] = Some(EqlOutput::Query(query));
+                }
+                Some(inbound_eql::InboundEql::Store(ciphertext)) => {
+                    let Some(column) = column else {
+                        return Err(EncryptError::InvalidInboundCiphertext.into());
+                    };
+                    positions.push((index, ciphertext, column.clone()));
+                }
+                None => {}
+            }
+        }
+        if positions.is_empty() {
+            return Ok(());
+        }
+
+        let ciphertexts = positions
+            .iter()
+            .map(|(_, ciphertext, _)| Some(ciphertext.clone()))
+            .collect();
+        let plaintexts = self.context.decrypt(ciphertexts).await.map_err(|err| {
+            warn!(
+                target: ENCRYPT,
+                client_id = self.context.client_id,
+                msg = "Inbound EQL ciphertext authentication failed",
+                error = ?err,
+            );
+            EncryptError::InvalidInboundCiphertext
+        })?;
+
+        // Re-encrypt the authenticated plaintext for the inferred destination
+        // and compare every derived SEM term. This detects term splicing: the
+        // AEAD tag authenticates `c`, but the searchable metadata sits outside
+        // it in the EQL envelope.
+        let verification_columns = positions
+            .iter()
+            .map(|(_, _, column)| Some(column.clone()))
+            .collect::<Vec<_>>();
+        let derived = self
+            .context
+            .encrypt(plaintexts, &verification_columns)
+            .await
+            .map_err(|err| {
+                warn!(
+                    target: ENCRYPT,
+                    client_id = self.context.client_id,
+                    msg = "Inbound EQL ciphertext metadata verification failed",
+                    error = ?err,
+                );
+                EncryptError::InvalidInboundCiphertext
+            })?;
+
+        for ((index, ciphertext, _), derived) in positions.into_iter().zip(derived) {
+            let Some(EqlOutput::Store(derived)) = derived else {
+                warn!(
+                    target: ENCRYPT,
+                    client_id = self.context.client_id,
+                    msg = "Inbound EQL ciphertext SEM terms did not match plaintext",
+                );
+                return Err(EncryptError::InvalidInboundCiphertext.into());
+            };
+            if !inbound_eql::sem_terms_match(&ciphertext, derived) {
+                warn!(
+                    target: ENCRYPT,
+                    client_id = self.context.client_id,
+                    msg = "Inbound EQL ciphertext SEM terms did not match plaintext",
+                );
+                return Err(EncryptError::InvalidInboundCiphertext.into());
+            }
+            encrypted[index] = Some(EqlOutput::Store(ciphertext));
+        }
+        Ok(())
+    }
+
     fn type_check<'a>(
         &self,
         statement: &'a ast::Statement,
@@ -1388,48 +1507,76 @@ fn project_query_operand(query_operand: bool, encrypted: &mut Option<EqlOutput>)
     }
 }
 
+/// JSON path/accessor literals are query operands even though they are passed
+/// as bare text and therefore need no query-domain cast. All other literals
+/// use the predicate roles recorded by EQL Mapper.
+fn literal_is_query_operand(
+    typed_statement: &TypeCheckedStatement<'_>,
+    literal: &ast::Value,
+    column: &Column,
+) -> bool {
+    matches!(
+        column.eql_term,
+        EqlTermVariant::JsonAccessor | EqlTermVariant::JsonPath
+    ) || typed_statement.query_operands.contains_literal(literal)
+}
+
 fn literals_to_plaintext(
     typed_statement: &TypeCheckedStatement<'_>,
     literal_columns: &Vec<Option<Column>>,
+) -> Result<Vec<Option<Plaintext>>, Error> {
+    literals_to_plaintext_skipping(typed_statement, literal_columns, &[])
+}
+
+fn literals_to_plaintext_skipping(
+    typed_statement: &TypeCheckedStatement<'_>,
+    literal_columns: &Vec<Option<Column>>,
+    skip: &[bool],
 ) -> Result<Vec<Option<Plaintext>>, Error> {
     let literals = typed_statement.literal_values();
 
     let plaintexts = literals
         .iter()
         .zip(literal_columns)
-        .map(|((eql_term, val), col)| match col {
-            Some(col) => {
-                let plaintext = match eql_term.variant() {
-                    EqlTermVariant::JsonValueSelector => {
-                        json_value_selector_literal_plaintext(typed_statement, val)
-                    }
-                    // A selector that carries a collapsed chain keys the composed
-                    // path, not the one segment it spells. Only a selector the
-                    // mapper recorded a chain for: a single access has no record
-                    // and takes the ordinary single-segment route below.
-                    EqlTermVariant::JsonAccessor
-                        if typed_statement
-                            .json_accessor_paths
-                            .for_literal(val)
-                            .is_some() =>
-                    {
-                        json_accessor_path_literal_plaintext(typed_statement, val)
-                    }
-                    _ => literal_from_sql(val, col.eql_term(), col.cast_type()),
-                };
+        .enumerate()
+        .map(|(index, ((eql_term, val), col))| {
+            if skip.get(index).copied().unwrap_or(false) {
+                return Ok(None);
+            }
+            match col {
+                Some(col) => {
+                    let plaintext = match eql_term.variant() {
+                        EqlTermVariant::JsonValueSelector => {
+                            json_value_selector_literal_plaintext(typed_statement, val)
+                        }
+                        // A selector that carries a collapsed chain keys the composed
+                        // path, not the one segment it spells. Only a selector the
+                        // mapper recorded a chain for: a single access has no record
+                        // and takes the ordinary single-segment route below.
+                        EqlTermVariant::JsonAccessor
+                            if typed_statement
+                                .json_accessor_paths
+                                .for_literal(val)
+                                .is_some() =>
+                        {
+                            json_accessor_path_literal_plaintext(typed_statement, val)
+                        }
+                        _ => literal_from_sql(val, col.eql_term(), col.cast_type()),
+                    };
 
-                plaintext.map_err(|err| {
-                    debug!(
-                        target: MAPPER,
-                        msg = "Could not convert literal value",
-                        value = ?val,
-                        cast_type = ?col.cast_type(),
-                        error = err.to_string()
-                    );
-                    MappingError::InvalidParameter(Box::new(col.to_owned())).into()
-                })
+                    plaintext.map_err(|err| {
+                        debug!(
+                            target: MAPPER,
+                            msg = "Could not convert literal value",
+                            value = ?val,
+                            cast_type = ?col.cast_type(),
+                            error = err.to_string()
+                        );
+                        MappingError::InvalidParameter(Box::new(col.to_owned())).into()
+                    })
+                }
+                None => Ok(None),
             }
-            None => Ok(None),
         })
         .collect::<Result<Vec<_>, Error>>()?;
     Ok(plaintexts)
diff --git a/packages/cipherstash-proxy/src/postgresql/inbound_eql.rs b/packages/cipherstash-proxy/src/postgresql/inbound_eql.rs
new file mode 100644
index 00000000..83246c0e
--- /dev/null
+++ b/packages/cipherstash-proxy/src/postgresql/inbound_eql.rs
@@ -0,0 +1,543 @@
+use crate::{error::EncryptError, postgresql::Column, EqlCiphertext, EqlQueryPayload};
+use cipherstash_client::{
+    eql::{EncryptedPayloadV3, EQL_SCHEMA_VERSION_V3},
+    schema::column::IndexType,
+};
+use eql_mapper::EqlTermVariant;
+use serde_json::Value;
+
+/// Tokenized SteVec selectors are 16 bytes rendered as lowercase hexadecimal.
+///
+/// Plaintext selectors can also match this format. In a JSON selector query
+/// position that ambiguity is intentionally resolved in favour of treating a
+/// match as an application-generated query operand.
+const SELECTOR_HASH_LEN: usize = 32;
+
+/// An application-generated EQL value entering Proxy.
+#[derive(Debug)]
+pub enum InboundEql {
+    /// A stored payload carrying source ciphertext. This must be authenticated
+    /// and have its SEM terms independently verified before it can be used.
+    Store(EqlCiphertext),
+    /// A query operand carrying SEM terms only. It can never be written and has
+    /// no source ciphertext with which to authenticate its metadata.
+    Query(EqlQueryPayload),
+}
+
+/// Parse a value only when its fields advertise it as an EQL storage payload or
+/// query operand. Query-only payloads are valid exclusively in syntactic query
+/// positions. Ordinary JSON (including an object with a `c` key) remains
+/// plaintext; malformed advertised payloads fail closed.
+pub fn parse(
+    bytes: &[u8],
+    column: &Column,
+    query_operand: bool,
+) -> Result<Option<InboundEql>, EncryptError> {
+    if query_operand
+        && matches!(
+            column.eql_term,
+            EqlTermVariant::JsonAccessor | EqlTermVariant::JsonPath
+        )
+        && is_selector_hash(bytes)
+    {
+        let selector = String::from_utf8(bytes.to_vec())
+            .map_err(|_| EncryptError::InvalidInboundCiphertext)?;
+        let query = EqlQueryPayload::Selector(selector);
+        validate_query_metadata(&query, column)?;
+        return Ok(Some(InboundEql::Query(query)));
+    }
+
+    let Ok(value) = serde_json::from_slice::<Value>(bytes) else {
+        return Ok(None);
+    };
+    let Some(object) = value.as_object() else {
+        return Ok(None);
+    };
+
+    let storage_shaped = object.contains_key("v")
+        && object.contains_key("i")
+        && (object.contains_key("c") || object.contains_key("h") || object.contains_key("sv"));
+    if storage_shaped {
+        let ciphertext: EqlCiphertext =
+            serde_json::from_value(value).map_err(|_| EncryptError::InvalidInboundCiphertext)?;
+        validate_storage_metadata(&ciphertext, column)?;
+        return Ok(Some(InboundEql::Store(ciphertext)));
+    }
+
+    let scalar_query_shaped = object.contains_key("v")
+        && object.contains_key("i")
+        && ["hm", "bf", "ob", "op"]
+            .iter()
+            .any(|term| object.contains_key(*term));
+    let ste_vec_query_shaped = object.len() == 1 && object.contains_key("sv");
+    if !scalar_query_shaped && !ste_vec_query_shaped {
+        return Ok(None);
+    }
+    if !query_operand {
+        return Err(EncryptError::InvalidInboundCiphertext);
+    }
+
+    let query: EqlQueryPayload =
+        serde_json::from_value(value).map_err(|_| EncryptError::InvalidInboundCiphertext)?;
+    validate_query_metadata(&query, column)?;
+    Ok(Some(InboundEql::Query(query)))
+}
+
+/// Equivalent to the static selector-hash regex `^[0-9a-f]{32}$`.
+fn is_selector_hash(bytes: &[u8]) -> bool {
+    bytes.len() == SELECTOR_HASH_LEN
+        && bytes
+            .iter()
+            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
+}
+
+fn validate_storage_metadata(
+    ciphertext: &EqlCiphertext,
+    column: &Column,
+) -> Result<(), EncryptError> {
+    if ciphertext.version() != EQL_SCHEMA_VERSION_V3
+        || ciphertext.identifier() != &column.identifier
+    {
+        return Err(EncryptError::InvalidInboundCiphertext);
+    }
+
+    // The descriptor is covered by the encrypted record's AEAD tag. Requiring
+    // the canonical table/column descriptor cryptographically binds a payload
+    // to its claimed destination, unlike the self-reported `i` field alone.
+    let expected_descriptor = format!("{}/{}", column.identifier.table, column.identifier.column);
+    let descriptor = match ciphertext {
+        EqlCiphertext::Encrypted(payload) => &payload.ciphertext.descriptor,
+        EqlCiphertext::SteVec(payload) => &payload.key_header.descriptor,
+    };
+    if descriptor != &expected_descriptor {
+        return Err(EncryptError::InvalidInboundCiphertext);
+    }
+
+    match ciphertext {
+        EqlCiphertext::Encrypted(payload) => validate_scalar_terms(payload, column),
+        EqlCiphertext::SteVec(payload) => {
+            let configured = column
+                .config
+                .indexes
+                .iter()
+                .any(|index| matches!(index.index_type, IndexType::SteVec { .. }));
+            if !configured || payload.ste_vec.is_empty() {
+                return Err(EncryptError::InvalidInboundCiphertext);
+            }
+            Ok(())
+        }
+    }
+}
+
+fn validate_query_metadata(query: &EqlQueryPayload, column: &Column) -> Result<(), EncryptError> {
+    match query {
+        EqlQueryPayload::Encrypted(payload) => {
+            if payload.version != EQL_SCHEMA_VERSION_V3 || payload.identifier != column.identifier {
+                return Err(EncryptError::InvalidInboundCiphertext);
+            }
+
+            match column.eql_term {
+                EqlTermVariant::Full | EqlTermVariant::Partial | EqlTermVariant::Tokenized => {
+                    validate_scalar_term_presence(
+                        payload.hmac_256.is_some(),
+                        payload.bloom_filter.is_some(),
+                        payload.ore_block_u64_8_256.is_some(),
+                        payload.ope_cllw.is_some(),
+                        column,
+                    )
+                }
+                EqlTermVariant::JsonOrd => {
+                    let ste_vec_configured = column
+                        .config
+                        .indexes
+                        .iter()
+                        .any(|index| matches!(index.index_type, IndexType::SteVec { .. }));
+                    if !ste_vec_configured
+                        || payload.hmac_256.is_some()
+                        || payload.bloom_filter.is_some()
+                        || payload.ore_block_u64_8_256.is_some()
+                        || payload.ope_cllw.is_none()
+                    {
+                        return Err(EncryptError::InvalidInboundCiphertext);
+                    }
+                    Ok(())
+                }
+                _ => Err(EncryptError::InvalidInboundCiphertext),
+            }
+        }
+        EqlQueryPayload::SteVec(payload) => {
+            let configured = column
+                .config
+                .indexes
+                .iter()
+                .any(|index| matches!(index.index_type, IndexType::SteVec { .. }));
+            let query_shape = matches!(
+                column.eql_term,
+                EqlTermVariant::Full | EqlTermVariant::Partial | EqlTermVariant::JsonValueSelector
+            );
+            if !configured || !query_shape || payload.ste_vec.is_empty() {
+                return Err(EncryptError::InvalidInboundCiphertext);
+            }
+            Ok(())
+        }
+        EqlQueryPayload::Selector(selector) => {
+            let configured = column
+                .config
+                .indexes
+                .iter()
+                .any(|index| matches!(index.index_type, IndexType::SteVec { .. }));
+            let query_shape = matches!(
+                column.eql_term,
+                EqlTermVariant::JsonAccessor | EqlTermVariant::JsonPath
+            );
+            if !configured || !query_shape || !is_selector_hash(selector.as_bytes()) {
+                return Err(EncryptError::InvalidInboundCiphertext);
+            }
+            Ok(())
+        }
+    }
+}
+
+/// Compare all searchable metadata after the plaintext has been authenticated
+/// and independently re-encrypted for the inferred destination column.
+/// `into_query_operand` removes only record ciphertext/key material, leaving
+/// the identifier and every scalar or SteVec SEM term. Bloom-filter positions
+/// are compared without regard to order; all other terms compare exactly.
+pub fn sem_terms_match(inbound: &EqlCiphertext, derived: EqlCiphertext) -> bool {
+    if let (EqlCiphertext::Encrypted(inbound), EqlCiphertext::Encrypted(derived)) =
+        (inbound, &derived)
+    {
+        return inbound.version == derived.version
+            && inbound.identifier == derived.identifier
+            && inbound.hmac_256 == derived.hmac_256
+            && bloom_filters_match(&inbound.bloom_filter, &derived.bloom_filter)
+            && inbound.ore_block_u64_8_256 == derived.ore_block_u64_8_256
+            && inbound.ope_cllw == derived.ope_cllw;
+    }
+
+    match (
+        serde_json::to_value(inbound.clone().into_query_operand()),
+        serde_json::to_value(derived.into_query_operand()),
+    ) {
+        (Ok(inbound), Ok(derived)) => inbound == derived,
+        _ => false,
+    }
+}
+
+fn bloom_filters_match(inbound: &Option<Vec<i16>>, derived: &Option<Vec<i16>>) -> bool {
+    match (inbound, derived) {
+        (Some(inbound), Some(derived)) => {
+            // Bloom-filter positions are a set. Their generation order is not
+            // stable, so comparing the serialized arrays directly rejects
+            // equivalent terms produced by independent encryptions.
+            let mut inbound = inbound.clone();
+            let mut derived = derived.clone();
+            inbound.sort_unstable();
+            derived.sort_unstable();
+            inbound == derived
+        }
+        (None, None) => true,
+        _ => false,
+    }
+}
+
+fn validate_scalar_terms(
+    payload: &EncryptedPayloadV3,
+    column: &Column,
+) -> Result<(), EncryptError> {
+    validate_scalar_term_presence(
+        payload.hmac_256.is_some(),
+        payload.bloom_filter.is_some(),
+        payload.ore_block_u64_8_256.is_some(),
+        payload.ope_cllw.is_some(),
+        column,
+    )
+}
+
+fn validate_scalar_term_presence(
+    has_hmac: bool,
+    has_bloom: bool,
+    has_ore: bool,
+    has_ope: bool,
+    column: &Column,
+) -> Result<(), EncryptError> {
+    let mut hmac = false;
+    let mut bloom = false;
+    let mut ore = false;
+    let mut ope = false;
+    for index in &column.config.indexes {
+        match index.index_type {
+            IndexType::Unique { .. } => hmac = true,
+            IndexType::Match { .. } => bloom = true,
+            IndexType::Ore => ore = true,
+            IndexType::Ope => ope = true,
+            IndexType::SteVec { .. } => return Err(EncryptError::InvalidInboundCiphertext),
+        }
+    }
+
+    if has_hmac != hmac || has_bloom != bloom || has_ore != ore || has_ope != ope {
+        return Err(EncryptError::InvalidInboundCiphertext);
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use cipherstash_client::schema::{ColumnConfig, ColumnMode, ColumnType};
+    use cipherstash_client::zerokms::EncryptedRecord;
+    use cipherstash_config::column::{ArrayIndexMode, Index, SteVecMode};
+    use eql_mapper::EqlTermVariant;
+    use uuid::Uuid;
+
+    fn column() -> Column {
+        Column {
+            identifier: crate::Identifier::new("users", "email"),
+            config: ColumnConfig {
+                name: "email".into(),
+                in_place: true,
+                cast_type: ColumnType::Text,
+                indexes: vec![],
+                mode: ColumnMode::Encrypted,
+            },
+            postgres_type: postgres_types::Type::TEXT,
+            eql_term: EqlTermVariant::Full,
+        }
+    }
+
+    fn payload(identifier: crate::Identifier) -> EqlCiphertext {
+        EqlCiphertext::Encrypted(EncryptedPayloadV3 {
+            version: EQL_SCHEMA_VERSION_V3,
+            identifier,
+            ciphertext: EncryptedRecord {
+                iv: Default::default(),
+                ciphertext: vec![1; 16],
+                tag: vec![2; 16],
+                descriptor: "users/email".into(),
+                keyset_id: Some(Uuid::nil()),
+                decryption_policy: None,
+            },
+            hmac_256: None,
+            bloom_filter: None,
+            ore_block_u64_8_256: None,
+            ope_cllw: None,
+        })
+    }
+
+    fn ste_vec_column() -> Column {
+        let mut column = column();
+        column.config.cast_type = ColumnType::Json;
+        column.config.indexes.push(Index::new(IndexType::SteVec {
+            prefix: "users/email".into(),
+            term_filters: Vec::new(),
+            array_index_mode: ArrayIndexMode::ALL,
+            mode: SteVecMode::default(),
+        }));
+        column.postgres_type = postgres_types::Type::JSONB;
+        column
+    }
+
+    #[test]
+    fn ordinary_json_is_plaintext() {
+        assert!(parse(br#"{"name":"Ada"}"#, &column(), false)
+            .unwrap()
+            .is_none());
+    }
+
+    #[test]
+    fn ordinary_json_with_a_c_key_is_plaintext() {
+        assert!(parse(br#"{"c":"customer code"}"#, &column(), false)
+            .unwrap()
+            .is_none());
+    }
+
+    #[test]
+    fn malformed_payload_shape_fails_closed() {
+        assert!(matches!(
+            parse(br#"{"v":3,"i":"users.email","c":"bad"}"#, &column(), false),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn destination_identifier_must_match() {
+        let ciphertext = payload(crate::Identifier::new("users", "phone"));
+        assert!(matches!(
+            validate_storage_metadata(&ciphertext, &column()),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn authenticated_descriptor_must_match_destination() {
+        let mut ciphertext = payload(crate::Identifier::new("users", "email"));
+        let EqlCiphertext::Encrypted(payload) = &mut ciphertext else {
+            unreachable!()
+        };
+        payload.ciphertext.descriptor = "accounts/email".into();
+        assert!(matches!(
+            validate_storage_metadata(&ciphertext, &column()),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn configured_sem_terms_must_be_present() {
+        let mut column = column();
+        column
+            .config
+            .indexes
+            .push(cipherstash_client::schema::column::Index::new_unique());
+        let ciphertext = payload(column.identifier.clone());
+        assert!(matches!(
+            validate_storage_metadata(&ciphertext, &column),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn independently_derived_sem_terms_must_match() {
+        let derived = payload(crate::Identifier::new("users", "email"));
+        let mut spliced = derived.clone();
+        let EqlCiphertext::Encrypted(payload) = &mut spliced else {
+            unreachable!()
+        };
+        payload.hmac_256 = Some("term from another plaintext".into());
+
+        assert!(!sem_terms_match(&spliced, derived));
+    }
+
+    #[test]
+    fn bloom_filter_order_does_not_affect_sem_term_matching() {
+        let mut inbound = payload(crate::Identifier::new("users", "email"));
+        let mut derived = inbound.clone();
+        if let EqlCiphertext::Encrypted(payload) = &mut inbound {
+            payload.bloom_filter = Some(vec![3, 1, 2]);
+        }
+        if let EqlCiphertext::Encrypted(payload) = &mut derived {
+            payload.bloom_filter = Some(vec![1, 2, 3]);
+        }
+
+        assert!(sem_terms_match(&inbound, derived));
+    }
+
+    #[test]
+    fn query_only_scalar_payload_is_accepted_for_a_query_operand() {
+        let mut column = column();
+        column
+            .config
+            .indexes
+            .push(cipherstash_client::schema::column::Index::new_unique());
+        let mut ciphertext = payload(column.identifier.clone());
+        let EqlCiphertext::Encrypted(payload) = &mut ciphertext else {
+            unreachable!()
+        };
+        payload.hmac_256 = Some("application-generated SEM term".into());
+        let query = serde_json::to_vec(&ciphertext.into_query_operand()).unwrap();
+
+        assert!(matches!(
+            parse(&query, &column, true),
+            Ok(Some(InboundEql::Query(_)))
+        ));
+    }
+
+    #[test]
+    fn query_only_scalar_payload_is_rejected_for_storage() {
+        let mut column = column();
+        column
+            .config
+            .indexes
+            .push(cipherstash_client::schema::column::Index::new_unique());
+        let mut ciphertext = payload(column.identifier.clone());
+        let EqlCiphertext::Encrypted(payload) = &mut ciphertext else {
+            unreachable!()
+        };
+        payload.hmac_256 = Some("application-generated SEM term".into());
+        let query = serde_json::to_vec(&ciphertext.into_query_operand()).unwrap();
+
+        assert!(matches!(
+            parse(&query, &column, false),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn query_only_scalar_identifier_must_match_destination() {
+        let mut column = column();
+        column
+            .config
+            .indexes
+            .push(cipherstash_client::schema::column::Index::new_unique());
+        let mut ciphertext = payload(crate::Identifier::new("users", "phone"));
+        let EqlCiphertext::Encrypted(payload) = &mut ciphertext else {
+            unreachable!()
+        };
+        payload.hmac_256 = Some("application-generated SEM term".into());
+        let query = serde_json::to_vec(&ciphertext.into_query_operand()).unwrap();
+
+        assert!(matches!(
+            parse(&query, &column, true),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn query_only_json_ordering_term_is_accepted() {
+        let mut column = ste_vec_column();
+        column.eql_term = EqlTermVariant::JsonOrd;
+        let query = serde_json::to_vec(&serde_json::json!({
+            "v": EQL_SCHEMA_VERSION_V3,
+            "i": { "t": "users", "c": "email" },
+            "op": "application-generated ordering term"
+        }))
+        .unwrap();
+
+        assert!(matches!(
+            parse(&query, &column, true),
+            Ok(Some(InboundEql::Query(EqlQueryPayload::Encrypted(_))))
+        ));
+    }
+
+    #[test]
+    fn query_only_ste_vec_payload_is_accepted_for_a_query_operand() {
+        let query = br#"{"sv":[{"s":"application-generated selector"}]}"#;
+
+        assert!(matches!(
+            parse(query, &ste_vec_column(), true),
+            Ok(Some(InboundEql::Query(EqlQueryPayload::SteVec(_))))
+        ));
+    }
+
+    #[test]
+    fn query_only_ste_vec_payload_is_rejected_for_storage() {
+        let query = br#"{"sv":[{"s":"application-generated selector"}]}"#;
+
+        assert!(matches!(
+            parse(query, &ste_vec_column(), false),
+            Err(EncryptError::InvalidInboundCiphertext)
+        ));
+    }
+
+    #[test]
+    fn bare_selector_hash_is_accepted_for_a_json_accessor_query_operand() {
+        let mut column = ste_vec_column();
+        column.eql_term = EqlTermVariant::JsonAccessor;
+
+        assert!(matches!(
+            parse(b"0123456789abcdef0123456789abcdef", &column, true),
+            Ok(Some(InboundEql::Query(EqlQueryPayload::Selector(selector))))
+                if selector == "0123456789abcdef0123456789abcdef"
+        ));
+    }
+
+    #[test]
+    fn selector_that_does_not_match_hash_format_remains_plaintext() {
+        let mut column = ste_vec_column();
+        column.eql_term = EqlTermVariant::JsonAccessor;
+
+        assert!(parse(b"patient.name", &column, true).unwrap().is_none());
+        assert!(parse(b"0123456789ABCDEF0123456789ABCDEF", &column, true)
+            .unwrap()
+            .is_none());
+    }
+}
diff --git a/packages/cipherstash-proxy/src/postgresql/messages/bind.rs b/packages/cipherstash-proxy/src/postgresql/messages/bind.rs
index b3f382f5..79cece8d 100644
--- a/packages/cipherstash-proxy/src/postgresql/messages/bind.rs
+++ b/packages/cipherstash-proxy/src/postgresql/messages/bind.rs
@@ -10,6 +10,7 @@ use crate::postgresql::data::{
     json_value_selector_plaintext,
 };
 use crate::postgresql::format_code::FormatCode;
+use crate::postgresql::inbound_eql;
 use crate::postgresql::protocol::BytesMutReadString;
 use crate::{EqlOutput, EqlQueryPayload};
 use crate::{SIZE_I16, SIZE_I32};
@@ -68,10 +69,23 @@ impl Bind {
         &self,
         output_params: &[OutputParam],
         param_types: &[i32],
+    ) -> Result<Vec<Option<Plaintext>>, Error> {
+        self.to_plaintext_skipping(output_params, param_types, &[])
+    }
+
+    pub fn to_plaintext_skipping(
+        &self,
+        output_params: &[OutputParam],
+        param_types: &[i32],
+        skip: &[bool],
     ) -> Result<Vec<Option<Plaintext>>, Error> {
         output_params
             .iter()
-            .map(|output| {
+            .enumerate()
+            .map(|(output_index, output)| {
+                if skip.get(output_index).copied().unwrap_or(false) {
+                    return Ok(None);
+                }
                 let Some(col) = &output.column else {
                     // Native param: forwarded verbatim, nothing to encrypt.
                     return Ok(None);
@@ -114,6 +128,37 @@ impl Bind {
             .collect()
     }
 
+    /// Detect application-generated storage or query payloads before decoding
+    /// parameters as their configured plaintext PostgreSQL types.
+    pub fn inbound_eql(
+        &self,
+        output_params: &[OutputParam],
+    ) -> Result<Vec<Option<inbound_eql::InboundEql>>, Error> {
+        output_params
+            .iter()
+            .map(|output| {
+                let Some(column) = &output.column else {
+                    return Ok(None);
+                };
+                let OutputParamSource::Input(input) = output.source else {
+                    return Ok(None);
+                };
+                let Some(param) = self.param_values.get(input) else {
+                    return Ok(None);
+                };
+                if param.is_null() {
+                    return Ok(None);
+                }
+                let bytes = if param.is_binary() && param.bytes.first() == Some(&1) {
+                    param.json_bytes()
+                } else {
+                    &param.bytes
+                };
+                inbound_eql::parse(bytes, column, output.query_operand).map_err(Error::from)
+            })
+            .collect()
+    }
+
     /// Composes `{"path", "value"}` — the input to `SteVecValueSelector` — from
     /// the operands of a JSON field equality.
     ///
diff --git a/packages/cipherstash-proxy/src/postgresql/mod.rs b/packages/cipherstash-proxy/src/postgresql/mod.rs
index 71c8df24..c743832e 100644
--- a/packages/cipherstash-proxy/src/postgresql/mod.rs
+++ b/packages/cipherstash-proxy/src/postgresql/mod.rs
@@ -6,6 +6,7 @@ mod error_handler;
 mod format_code;
 mod frontend;
 mod handler;
+mod inbound_eql;
 mod message_buffer;
 mod messages;
 mod parser;
diff --git a/packages/cipherstash-proxy/src/proxy/zerokms/zerokms.rs b/packages/cipherstash-proxy/src/proxy/zerokms/zerokms.rs
index bb563c19..b602006d 100644
--- a/packages/cipherstash-proxy/src/proxy/zerokms/zerokms.rs
+++ b/packages/cipherstash-proxy/src/proxy/zerokms/zerokms.rs
@@ -16,7 +16,7 @@ use cipherstash_client::{
         PreparedPlaintext,
     },
     schema::column::IndexType,
-    zerokms::{Decryptable, EncryptedRecord, RecordWithNonce, RetrieveKeyPayload},
+    zerokms::{Decryptable, EncryptedRecord, IdentifiedBy, RecordWithNonce, RetrieveKeyPayload},
 };
 use eql_mapper::EqlTermVariant;
 use metrics::{counter, histogram};
@@ -164,7 +164,15 @@ impl ZeroKms {
         info!(target: ZEROKMS, msg = "Initializing ZeroKMS ScopedCipher (cache miss)", ?keyset_id);
         counter!(KEYSET_CIPHER_CACHE_MISS_TOTAL).increment(1);
 
-        let identified_by = keyset_id.as_ref().map(|id| id.0.clone());
+        // A connection-level keyset takes precedence. Otherwise, scope the
+        // cipher to Proxy's configured default instead of passing `None` and
+        // silently falling back to the ZeroKMS client's account default. The
+        // two defaults are not required to be the same, and using the account
+        // default would derive different searchable-encryption terms.
+        let identified_by = keyset_id
+            .as_ref()
+            .map(|id| id.0.clone())
+            .or_else(|| self.default_keyset_id.map(IdentifiedBy::Uuid));
 
         let start = Instant::now();
         let result = ScopedCipher::init(zerokms_client, identified_by).await;
diff --git a/packages/eql-mapper/src/inference/infer_type_impls/expr.rs b/packages/eql-mapper/src/inference/infer_type_impls/expr.rs
index 525fc6ec..17773b51 100644
--- a/packages/eql-mapper/src/inference/infer_type_impls/expr.rs
+++ b/packages/eql-mapper/src/inference/infer_type_impls/expr.rs
@@ -414,9 +414,8 @@ impl<'ast> InferType<'ast, Expr> for TypeInferencer<'ast> {
 
                 // The operands of a predicate reach PostgreSQL as query
                 // operands — terms only, never a ciphertext. Record them so the
-                // proxy projects their payloads accordingly. Containment
-                // (`@>`/`<@`) is deliberately excluded: its needle is a whole
-                // document and keeps its full payload.
+                // proxy projects their payloads accordingly. JSON containment
+                // uses a SteVec query needle rather than a stored document.
                 if matches!(
                     op,
                     BinaryOperator::Eq
@@ -426,6 +425,8 @@ impl<'ast> InferType<'ast, Expr> for TypeInferencer<'ast> {
                         | BinaryOperator::Gt
                         | BinaryOperator::GtEq
                         | BinaryOperator::AtAt
+                        | BinaryOperator::AtArrow
+                        | BinaryOperator::ArrowAt
                 ) {
                     self.record_query_operands([&**left, &**right]);
                 }
diff --git a/packages/eql-mapper/src/lib.rs b/packages/eql-mapper/src/lib.rs
index 28dbfbbe..4de54147 100644
--- a/packages/eql-mapper/src/lib.rs
+++ b/packages/eql-mapper/src/lib.rs
@@ -2624,7 +2624,7 @@ mod test {
             tables: {
                 patients: {
                     id,
-                    notes (EQL: JsonLike + Contain),
+                    notes (EQL("eql_v3_json_search"): JsonLike + Contain),
                 }
             }
         });
@@ -2635,14 +2635,18 @@ mod test {
 
         match type_check(schema, &statement) {
             Ok(typed) => {
+                if matches!(op, "@>" | "<@") {
+                    let literal = typed.literal_values()[0].1;
+                    assert!(typed.query_operands.contains_literal(literal));
+                }
                 match typed.transform(test_helpers::dummy_encrypted_json_selector(
                     &statement,
                     vec![ast::Value::SingleQuotedString("medications".to_owned())],
                 )) {
                     Ok(statement) => {
                         let expected = match op {
-                            "@>" => "SELECT id, eql_v3.jsonb_contains(notes, '<encrypted-selector(medications)>'::JSONB::public.eql_v3_text_search) AS meds FROM patients".to_string(),
-                            "<@" => "SELECT id, eql_v3.jsonb_contained_by(notes, '<encrypted-selector(medications)>'::JSONB::public.eql_v3_text_search) AS meds FROM patients".to_string(),
+                            "@>" => "SELECT id, eql_v3.jsonb_contains(notes, '<encrypted-selector(medications)>'::JSONB::eql_v3.query_json) AS meds FROM patients".to_string(),
+                            "<@" => "SELECT id, eql_v3.jsonb_contained_by(notes, '<encrypted-selector(medications)>'::JSONB::eql_v3.query_json) AS meds FROM patients".to_string(),
                             // -> / ->> field access: functionalised to eql_v3."->"/"->>",
                             // with the field selector passed as encrypted text.
                             "->" => "SELECT id, eql_v3.\"->\"(notes, '<encrypted-selector(medications)>') AS meds FROM patients".to_string(),
@@ -2664,7 +2668,7 @@ mod test {
             tables: {
                 patients: {
                     id,
-                    notes (EQL: JsonLike + Contain),
+                    notes (EQL("eql_v3_json_search"): JsonLike + Contain),
                 }
             }
         });
@@ -2685,7 +2689,7 @@ mod test {
             tables: {
                 patients: {
                     id,
-                    notes (EQL: JsonLike + Contain),
+                    notes (EQL("eql_v3_json_search"): JsonLike + Contain),
                 }
             }
         });
@@ -2775,7 +2779,7 @@ mod test {
             tables: {
                 patients: {
                     id,
-                    notes (EQL: JsonLike + Contain),
+                    notes (EQL("eql_v3_json_search"): JsonLike + Contain),
                 }
             }
         });
@@ -2795,12 +2799,14 @@ mod test {
             "Expected @> to be transformed to eql_v3.jsonb_contains, got: {sql}"
         );
 
-        // CRITICAL: Verify the parameter is cast to enable GIN index usage
-        // The cast ::JSONB::public.eql_v3_text_search is required for GIN indexes to work
+        // A containment needle is a term-only query operand. It must use the
+        // query_json domain so no source ciphertext is required.
         assert!(
-            sql.contains("::JSONB::public.eql_v3_text_search") || sql.contains("::jsonb::public.eql_v3_text_search"),
-            "Expected parameter to be cast as ::JSONB::public.eql_v3_text_search for GIN index support, got: {sql}"
+            sql.contains("::JSONB::eql_v3.query_json")
+                || sql.contains("::jsonb::eql_v3.query_json"),
+            "Expected parameter to be cast as ::JSONB::eql_v3.query_json, got: {sql}"
         );
+        assert!(transformed.params.outputs()[0].query_operand);
     }
 
     #[test]
@@ -2809,7 +2815,7 @@ mod test {
             tables: {
                 patients: {
                     id,
-                    notes (EQL: JsonLike + Contain),
+                    notes (EQL("eql_v3_json_search"): JsonLike + Contain),
                 }
             }
         });
@@ -2829,11 +2835,13 @@ mod test {
             "Expected <@ to be transformed to eql_v3.jsonb_contained_by, got: {sql}"
         );
 
-        // CRITICAL: Verify the parameter is cast to enable GIN index usage
+        // The contained value is also a term-only query operand.
         assert!(
-            sql.contains("::JSONB::public.eql_v3_text_search") || sql.contains("::jsonb::public.eql_v3_text_search"),
-            "Expected parameter to be cast as ::JSONB::public.eql_v3_text_search for GIN index support, got: {sql}"
+            sql.contains("::JSONB::eql_v3.query_json")
+                || sql.contains("::jsonb::eql_v3.query_json"),
+            "Expected parameter to be cast as ::JSONB::eql_v3.query_json, got: {sql}"
         );
+        assert!(transformed.params.outputs()[0].query_operand);
     }
 
     #[test]
diff --git a/packages/eql-mapper/src/query_operands.rs b/packages/eql-mapper/src/query_operands.rs
index f167e83a..0d876428 100644
--- a/packages/eql-mapper/src/query_operands.rs
+++ b/packages/eql-mapper/src/query_operands.rs
@@ -27,9 +27,9 @@ use crate::Param;
 ///
 /// Membership is decided syntactically, by the predicate an operand belongs to
 /// — the same contexts whose rewrite rules cast to a `eql_v3.query_*` twin:
-/// comparisons (`=`, `<>`, `<`, `<=`, `>`, `>=`), `LIKE`/`ILIKE` and `@@`.
-/// Everything else — `INSERT` values, `UPDATE` assignments, containment needles
-/// — is a stored value and keeps its full payload.
+/// comparisons (`=`, `<>`, `<`, `<=`, `>`, `>=`), containment (`@>`/`<@`),
+/// `LIKE`/`ILIKE` and `@@`. Everything else — including `INSERT` values and
+/// `UPDATE` assignments — is a stored value and keeps its full payload.
 #[derive(Debug, Default)]
 pub struct QueryOperands<'ast> {
     params: HashSet<Param>,
diff --git a/packages/eql-mapper/src/transformation_rules/helpers.rs b/packages/eql-mapper/src/transformation_rules/helpers.rs
index 81461ac1..c30625c3 100644
--- a/packages/eql-mapper/src/transformation_rules/helpers.rs
+++ b/packages/eql-mapper/src/transformation_rules/helpers.rs
@@ -64,9 +64,9 @@ pub(crate) fn query_operand_domain(eql_term: &EqlTerm) -> Option<(String, String
 /// own domain, carrying the ciphertext plus every search term the column
 /// indexes.
 ///
-/// This is what an `INSERT` value, an `UPDATE` assignment and a containment
-/// needle all need — as opposed to a predicate operand, which needs only the
-/// terms of [`query_operand_domain`].
+/// This is what an `INSERT` value or an `UPDATE` assignment needs — as opposed
+/// to a predicate operand, which needs only the terms of
+/// [`query_operand_domain`].
 ///
 /// Returns `None` for a JSON selector, which is bare text in every position.
 pub(crate) fn full_payload_domain(eql_term: &EqlTerm) -> Option<(String, String)> {
diff --git a/packages/eql-mapper/src/transformation_rules/rewrite_containment_ops.rs b/packages/eql-mapper/src/transformation_rules/rewrite_containment_ops.rs
index 0acfe477..2eda2d32 100644
--- a/packages/eql-mapper/src/transformation_rules/rewrite_containment_ops.rs
+++ b/packages/eql-mapper/src/transformation_rules/rewrite_containment_ops.rs
@@ -13,7 +13,7 @@ use sqltk::{NodeKey, NodePath, Visitable};
 use crate::unifier::{Type, Value};
 use crate::EqlMapperError;
 
-use super::helpers::{cast_encrypted_operand, full_payload_domain};
+use super::helpers::{cast_encrypted_operand, query_operand_domain};
 use super::TransformationRule;
 
 /// Rewrites JSON binary operators on encrypted columns to `eql_v3` function
@@ -115,17 +115,15 @@ impl<'ast> TransformationRule<'ast> for RewriteContainmentOps<'ast> {
                     _ => return Ok(false),
                 };
 
-                // A containment needle is a whole encrypted document, so it
-                // casts to the column domain, not to a query twin. A `->`/`->>`
-                // selector takes no cast at all — `full_payload_domain` returns
-                // `None` for it — because `eql_v3."->"(json, text)` wants the
-                // bare encrypted selector text.
-                cast_encrypted_operand(&self.node_types, original_left, left, full_payload_domain);
+                // Containment uses a term-only SteVec query needle. A `->`/`->>`
+                // selector also remains query-only but takes no cast because
+                // `eql_v3."->"(json, text)` wants the bare encrypted selector.
+                cast_encrypted_operand(&self.node_types, original_left, left, query_operand_domain);
                 cast_encrypted_operand(
                     &self.node_types,
                     original_right,
                     right,
-                    full_payload_domain,
+                    query_operand_domain,
                 );
 
                 // Use mem::replace to move (not copy) the original nodes,
diff --git a/packages/showcase/Cargo.toml b/packages/showcase/Cargo.toml
index 5881d1ad..d8a7f124 100644
--- a/packages/showcase/Cargo.toml
+++ b/packages/showcase/Cargo.toml
@@ -5,6 +5,8 @@ edition.workspace = true
 description = "Healthcare data model demonstrating EQL v3 searchable encryption with realistic encrypted application patterns"
 
 [dependencies]
+cipherstash-client = { workspace = true, features = ["tokio"] }
+cipherstash-config = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 uuid = { version = "1.11.0", features = ["serde", "v4"] }
diff --git a/packages/showcase/README.md b/packages/showcase/README.md
index 6dae109d..aa1297d7 100644
--- a/packages/showcase/README.md
+++ b/packages/showcase/README.md
@@ -466,12 +466,30 @@ mise run test:integration:showcase
 
 The showcase will execute and display:
 
-1. **Original Healthcare Query**: Aspirin prescription lookup
-2. **Field Access Operations**: Testing `->` and `->>`
-3. **Containment Operations**: Testing `@>` and `<@`
-4. **JSONPath Functions**: Testing `jsonb_path_*` functions
-5. **Comparison Operations**: Numeric, string, date, and float comparisons
-6. **Complex Nested Queries**: JOINs, aggregations, and subqueries
+1. **Application-side EQL**: Insert pre-encrypted storage payloads and search with query-only SEM payloads, using both parameters and SQL literals
+2. **Original Healthcare Query**: Aspirin prescription lookup
+3. **Field Access Operations**: Testing `->` and `->>`
+4. **Containment Operations**: Testing `@>` and `<@`
+5. **JSONPath Functions**: Testing `jsonb_path_*` functions
+6. **Comparison Operations**: Numeric, string, date, and float comparisons
+7. **Complex Nested Queries**: JOINs, aggregations, and subqueries
+
+### Application-side Encryption
+
+The `pre_encrypted` example constructs the same `eql_v3_json_search` column configuration declared by `patients.pii`, encrypts patient PII or its query SEM terms with `cipherstash-client`, and sends the resulting EQL payload through Proxy. The application-side `ColumnConfig` uses the canonical `patients/pii` descriptor; this authenticated descriptor binds stored ciphertext to its destination. The example demonstrates both supported input forms and both payload roles:
+
+```sql
+INSERT INTO patients (id, pii) VALUES ($1, $2); -- payload parameter
+INSERT INTO patients (id, pii) VALUES ('...', '{...}'); -- payload literal
+SELECT id FROM patients WHERE pii @> $1; -- query-only SEM parameter
+SELECT id FROM patients WHERE pii @> '{...}'; -- query-only SEM literal
+SELECT pii -> $1 FROM patients; -- bare selector-hash parameter
+SELECT jsonb_path_query_first(pii, '<hash>') FROM patients; -- selector-hash literal
+```
+
+For stored payloads, Proxy authenticates the ciphertext, checks that its identifier and authenticated descriptor match `patients.pii`, and independently re-derives every SEM term from the decrypted plaintext. Query-only payloads deliberately contain no source ciphertext, so authentication is impossible and unnecessary: Proxy validates that their shape is valid for `patients.pii` and accepts them only in query positions, where incorrect terms can only produce incorrect query results and cannot poison stored data. Both forms are forwarded without double encryption.
+
+A SteVec path selector is a bare tokenized-selector hash matching `^[0-9a-f]{32}$`. In a JSON selector query position, Proxy treats a matching value as already hashed; anything not matching that format is treated as plaintext and encrypted normally. This is intentionally ambiguous: plaintext selectors are a superset of the hash format, so a genuine plaintext selector consisting of exactly 32 lowercase hexadecimal characters is also treated as already hashed. Applications with such a field name must currently query it using an application-generated selector hash.
 
 Each test section provides detailed output showing:
 - ✅ Successful query execution
@@ -499,4 +517,4 @@ Examples:
 
 ⚠️ **Chained Operators**: The `->` operator cannot be chained on `ste_vec` encrypted columns. Use JSONPath functions like `jsonb_path_query_first()` for deep nested access instead.
 
-This showcase proves that EQL v3 provides comprehensive JSONB support for encrypted data, enabling sophisticated healthcare applications while maintaining strong privacy protections.
\ No newline at end of file
+This showcase proves that EQL v3 provides comprehensive JSONB support for encrypted data, enabling sophisticated healthcare applications while maintaining strong privacy protections.
diff --git a/packages/showcase/src/main.rs b/packages/showcase/src/main.rs
index 9cc67097..f6c76226 100644
--- a/packages/showcase/src/main.rs
+++ b/packages/showcase/src/main.rs
@@ -53,6 +53,7 @@
 mod common;
 mod data;
 mod model;
+mod pre_encrypted;
 mod schema;
 
 use common::{connect_with_tls, trace, PROXY};
@@ -75,6 +76,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     setup_schema().await;
     insert_test_data().await;
     create_enhanced_jsonb_test_data().await;
+    pre_encrypted::run_examples().await?;
 
     let client = connect_with_tls(*PROXY).await;
 
@@ -156,6 +158,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("   • Healthcare-compliant database schema with proper foreign keys");
     println!("   • Realistic medical data with nested objects, arrays, and mixed data types");
     println!("   • Secure querying of encrypted data while maintaining privacy");
+    println!("   • Application-side encryption passed through Proxy as parameters and literals");
     println!();
     println!("✨ EQL v3 provides comprehensive JSONB support for encrypted healthcare data!");
 
diff --git a/packages/showcase/src/pre_encrypted.rs b/packages/showcase/src/pre_encrypted.rs
new file mode 100644
index 00000000..cdd6bff5
--- /dev/null
+++ b/packages/showcase/src/pre_encrypted.rs
@@ -0,0 +1,228 @@
+//! Application-side EQL examples for storage and search.
+//!
+//! Proxy accepts storage and query-only payloads as either bound parameters or
+//! SQL literals, applies role-appropriate validation, and avoids encrypting
+//! them a second time.
+
+use crate::common::{connect_with_tls, PROXY};
+use cipherstash_client::{
+    encryption::{Plaintext, QueryOp, ScopedCipher},
+    eql::{
+        encrypt_eql_v3, EqlEncryptOpts, EqlOperation, EqlOutputV3, Identifier, PreparedPlaintext,
+    },
+    schema::{ColumnConfig, ColumnType},
+    zerokms::{ClientKey, ZeroKMSBuilder},
+    AutoStrategy, IdentifiedBy,
+};
+use cipherstash_config::column::{ArrayIndexMode, Index, IndexType, SteVecMode};
+use serde_json::{json, Value};
+use std::{borrow::Cow, sync::Arc};
+use uuid::Uuid;
+
+pub async fn run_examples() -> Result<(), Box<dyn std::error::Error>> {
+    println!("\n🔐 === Application-side EQL encryption ===");
+    let client = connect_with_tls(*PROXY).await;
+
+    // Example 1: bind an application-encrypted payload as a parameter.
+    let parameter_id = Uuid::parse_str("a1b2c3d4-e5f6-4a5b-8c9d-123456789021")?;
+    let parameter_pii = json!({
+        "first_name": "Ada",
+        "last_name": "Lovelace",
+        "email": "ada@example.com",
+        "date_of_birth": "1815-12-10"
+    });
+    let parameter_payload = encrypt_patient_pii(parameter_pii.clone()).await?;
+    client
+        .execute(
+            "INSERT INTO patients (id, pii) VALUES ($1, $2)",
+            &[&parameter_id, &parameter_payload],
+        )
+        .await?;
+    println!("✅ Inserted application-encrypted PII as a bound parameter");
+
+    // Example 2: the same wire payload can be supplied as a SQL literal.
+    let literal_id = Uuid::parse_str("a1b2c3d4-e5f6-4a5b-8c9d-123456789022")?;
+    let literal_pii = json!({
+        "first_name": "Grace",
+        "last_name": "Hopper",
+        "email": "grace@example.com",
+        "date_of_birth": "1906-12-09"
+    });
+    let literal_payload = encrypt_patient_pii(literal_pii.clone()).await?;
+    let literal_payload = literal_payload.to_string().replace('\'', "''");
+    client
+        .simple_query(&format!(
+            "INSERT INTO patients (id, pii) VALUES ('{literal_id}', '{literal_payload}')"
+        ))
+        .await?;
+    println!("✅ Inserted application-encrypted PII as a SQL literal");
+
+    // Both rows still decrypt normally when selected through Proxy.
+    for (id, expected) in [
+        (parameter_id, parameter_pii.clone()),
+        (literal_id, literal_pii.clone()),
+    ] {
+        let row = client
+            .query_one("SELECT pii FROM patients WHERE id = $1", &[&id])
+            .await?;
+        assert_eq!(row.get::<_, Value>(0), expected);
+    }
+    println!("✅ Proxy authenticated and decrypted both application-encrypted values");
+
+    // Example 3: a query-only EQL payload contains SteVec SEM terms but no
+    // source ciphertext. Proxy validates its query role and forwards it without
+    // attempting authentication or encrypting it a second time.
+    let parameter_query = query_patient_pii(parameter_pii).await?;
+    let row = client
+        .query_one(
+            "SELECT id FROM patients WHERE pii @> $1",
+            &[&parameter_query],
+        )
+        .await?;
+    assert_eq!(row.get::<_, Uuid>(0), parameter_id);
+    println!("✅ Queried with application-generated SEM terms as a bound parameter");
+
+    // Example 4: query-only payloads are also accepted as SQL literals in
+    // predicate positions.
+    let literal_query = query_patient_pii(literal_pii)
+        .await?
+        .to_string()
+        .replace('\'', "''");
+    let rows = client
+        .simple_query(&format!(
+            "SELECT id FROM patients WHERE pii @> '{literal_query}'"
+        ))
+        .await?;
+    let matched = rows.iter().any(|message| match message {
+        tokio_postgres::SimpleQueryMessage::Row(row) => {
+            row.get(0) == Some(literal_id.to_string().as_str())
+        }
+        _ => false,
+    });
+    assert!(matched);
+    println!("✅ Queried with application-generated SEM terms as a SQL literal");
+
+    // Example 5: SteVec path selectors are bare, 32-character lowercase hex
+    // query terms. Proxy recognises and forwards an application-generated hash
+    // instead of hashing it again.
+    let parameter_selector = query_patient_selector("$.first_name").await?;
+    let row = client
+        .query_one(
+            "SELECT pii -> $1 FROM patients WHERE id = $2",
+            &[&parameter_selector, &parameter_id],
+        )
+        .await?;
+    assert_eq!(row.get::<_, Value>(0), json!("Ada"));
+    println!("✅ Queried with an application-generated selector hash parameter");
+
+    // Example 6: selector hashes work as literals too. A plaintext selector
+    // matching the same format is ambiguous and is intentionally interpreted
+    // as already hashed; see the showcase README for the compatibility rule.
+    let literal_selector = query_patient_selector("$.first_name").await?;
+    let rows = client
+        .simple_query(&format!(
+            "SELECT jsonb_path_query_first(pii, '{literal_selector}') \
+             FROM patients WHERE id = '{literal_id}'"
+        ))
+        .await?;
+    let selected = rows.iter().find_map(|message| match message {
+        tokio_postgres::SimpleQueryMessage::Row(row) => row.get(0),
+        _ => None,
+    });
+    assert_eq!(selected, Some("\"Grace\""));
+    println!("✅ Queried with an application-generated selector hash literal");
+    Ok(())
+}
+
+async fn encrypt_patient_pii(value: Value) -> Result<Value, Box<dyn std::error::Error>> {
+    let config = patient_pii_config();
+    let prepared = PreparedPlaintext::new(
+        Cow::Owned(config),
+        Identifier::new("patients", "pii"),
+        Plaintext::Json(Some(value)),
+        EqlOperation::Store,
+    );
+    let mut outputs = encrypt_eql_v3(
+        scoped_cipher().await?,
+        vec![prepared],
+        &EqlEncryptOpts::default(),
+    )
+    .await?;
+    let EqlOutputV3::Store(ciphertext) = outputs.remove(0) else {
+        return Err("store encryption returned a query payload".into());
+    };
+    Ok(serde_json::to_value(ciphertext)?)
+}
+
+async fn query_patient_pii(value: Value) -> Result<Value, Box<dyn std::error::Error>> {
+    let config = patient_pii_config();
+    let index_type = config.indexes[0].index_type.clone();
+    let prepared = PreparedPlaintext::new(
+        Cow::Owned(config),
+        Identifier::new("patients", "pii"),
+        Plaintext::Json(Some(value)),
+        EqlOperation::Query(&index_type, QueryOp::Default),
+    );
+    let mut outputs = encrypt_eql_v3(
+        scoped_cipher().await?,
+        vec![prepared],
+        &EqlEncryptOpts::default(),
+    )
+    .await?;
+    let EqlOutputV3::Query(query) = outputs.remove(0) else {
+        return Err("query encryption returned a storage payload".into());
+    };
+    Ok(serde_json::to_value(query)?)
+}
+
+async fn query_patient_selector(path: &str) -> Result<String, Box<dyn std::error::Error>> {
+    let config = patient_pii_config();
+    let index_type = config.indexes[0].index_type.clone();
+    let prepared = PreparedPlaintext::new(
+        Cow::Owned(config),
+        Identifier::new("patients", "pii"),
+        Plaintext::from(path),
+        EqlOperation::Query(&index_type, QueryOp::SteVecSelector),
+    );
+    let mut outputs = encrypt_eql_v3(
+        scoped_cipher().await?,
+        vec![prepared],
+        &EqlEncryptOpts::default(),
+    )
+    .await?;
+    let EqlOutputV3::Query(query) = outputs.remove(0) else {
+        return Err("selector encryption returned a storage payload".into());
+    };
+    let Value::String(selector) = serde_json::to_value(query)? else {
+        return Err("selector encryption returned a non-selector query payload".into());
+    };
+    Ok(selector)
+}
+
+fn patient_pii_config() -> ColumnConfig {
+    ColumnConfig::build("patients/pii")
+        .casts_as(ColumnType::Json)
+        .add_index(Index::new(IndexType::SteVec {
+            prefix: "patients/pii".into(),
+            term_filters: Vec::new(),
+            array_index_mode: ArrayIndexMode::ALL,
+            mode: SteVecMode::default(),
+        }))
+}
+
+async fn scoped_cipher() -> Result<Arc<ScopedCipher<AutoStrategy>>, Box<dyn std::error::Error>> {
+    let client_id = env("CS_CLIENT_ID", "CS_ENCRYPT__CLIENT_ID")?.parse()?;
+    let client_key =
+        ClientKey::from_hex_v1(client_id, &env("CS_CLIENT_KEY", "CS_ENCRYPT__CLIENT_KEY")?)?;
+    let zerokms = ZeroKMSBuilder::auto()?
+        .with_client_key(client_key)
+        .build()?;
+    let keyset_id: Uuid = env("CS_DEFAULT_KEYSET_ID", "CS_ENCRYPT__DEFAULT_KEYSET_ID")?.parse()?;
+    Ok(Arc::new(
+        ScopedCipher::init(Arc::new(zerokms), Some(IdentifiedBy::Uuid(keyset_id))).await?,
+    ))
+}
+
+fn env(primary: &str, nested: &str) -> Result<String, std::env::VarError> {
+    std::env::var(primary).or_else(|_| std::env::var(nested))
+}