@@ -511,6 +511,10 @@ fn postgres_containment_over_the_encrypted_column_is_jsonb_containment() {
 #[path = "json_tree/generated.rs"]
 mod generated;
 
+/// The element-scoped predicate, through a data plan, against the column.
+#[path = "json_tree/elements.rs"]
+mod elements;
+
 /// Needs an EQL v4 installation. Extraction by selector opens to the leaf;
 /// the subtree walk by a client-derived token returns exactly the nodes
 /// under it and reassembles; the array functions count and list elements;
 
@@ -0,0 +1,339 @@
+//! The element-scoped predicate end to end: documents sealed by Stack
+//! Encrypt's JSON index, the `element_match` operand derived through a data
+//! plan over eql-bindings' target table (the path the Go guest takes), and
+//! `eql_v4.json_tree_element_match` over the stored column.
+
+use super::*;
+
+use eql_bindings::encryption::targets::{self as eql_targets, Target};
+use stack_encrypt::dynamic::{
+    self, record, OpSpec, Operand, TargetDescriptor, TargetError, TargetResolver, ValueFrame,
+};
+use stack_encrypt::{KeysetCipher, Label, Pending, StackCipher};
+
+/// eql-bindings' table as the engine's resolver, naming each type's leaf
+/// and element predicate domains, as a host that answers `element_match`
+/// describes it.
+struct Eql;
+
+fn describe(target: &Target) -> TargetDescriptor {
+    let mut descriptor = TargetDescriptor::new(
+        target.name,
+        target.family,
+        target.suffix,
+        target.plaintext_kind(),
+        target.sql_domain,
+        target.indexes.iter().map(|key| (*key).to_owned()).collect(),
+        target.query.map(str::to_owned),
+        target.query_sql_domain.map(str::to_owned),
+        target.producible,
+        target.reason.map(str::to_owned),
+    );
+    if let Some(domain) = target.leaf_query_sql_domain() {
+        descriptor = descriptor.with_leaf_query_sql_domain(domain);
+    }
+    if let Some(domain) = target.elements_query_sql_domain() {
+        descriptor = descriptor.with_elements_query_sql_domain(domain);
+    }
+    descriptor
+}
+
+impl TargetResolver for Eql {
+    fn targets(&self) -> Vec<TargetDescriptor> {
+        eql_targets::targets().iter().map(describe).collect()
+    }
+    fn encrypt<'a, K: 'static>(
+        &self,
+        name: &str,
+        options: Option<&dynamic::TargetOptions>,
+        keyset: &'a KeysetCipher<'_, K>,
+        label: &Label,
+        plaintext: ValueFrame,
+    ) -> Result<Pending<'a, Vec<u8>, K>, TargetError> {
+        eql_targets::encrypt_with(name, options, keyset, label, plaintext)
+            .map_err(|error| TargetError::Other(Box::new(error)))
+    }
+    fn decrypt<'a, K: 'static>(
+        &self,
+        name: &str,
+        cipher: &'a StackCipher<K>,
+        label: &Label,
+        stored: &[u8],
+    ) -> Result<Pending<'a, ValueFrame, K>, TargetError> {
+        eql_targets::decrypt(name, cipher, label, stored)
+            .map_err(|error| TargetError::Other(Box::new(error)))
+    }
+    fn query<'a, K: 'static>(
+        &self,
+        name: &str,
+        index: &stack_encrypt::target::IndexSpec,
+        keyset: &'a KeysetCipher<'_, K>,
+        label: &Label,
+        plaintext: ValueFrame,
+    ) -> Result<Pending<'a, Vec<u8>, K>, TargetError> {
+        eql_targets::query(name, index, keyset, label, plaintext)
+            .map_err(|error| TargetError::Other(Box::new(error)))
+    }
+}
+
+const RELATIONSHIP: &str = "$.extra.emergency_contacts[*].relationship";
+const CALLS: &str = "$.extra.emergency_contacts[*].previous_call_count";
+
+/// The documents' index options: each contact's relationship and call
+/// count carry ordering terms.
+fn contact_options() -> JsonOptions {
+    JsonOptions::default()
+        .ordering([RELATIONSHIP, CALLS])
+        .unwrap()
+}
+
+/// `profile` under `users/profile`, the column the documents are sealed
+/// under, naming `JsonSearch` with [`contact_options`].
+fn contact_plan() -> record::Plan {
+    let wire = json!({
+        "profile": {
+            "context": ["users", "profile"],
+            "type": "json",
+            "null": "sql",
+            "target": {"name": "JsonSearch", "options": {"ordered": [RELATIONSHIP, CALLS]}}
+        }
+    });
+    record::plan_with(&dynamic::frame::serde::to_vec(&wire).unwrap(), &Eql).unwrap()
+}
+
+fn frame_of(value: &Value) -> ValueFrame {
+    ValueFrame::from_bytes(&dynamic::frame::serde::to_vec(value).unwrap()).unwrap()
+}
+
+/// The operand of `operator` on `profile` for `input`, as the SQL binds
+/// it: an EQL operand's JSON text, or a list of tokens.
+fn derive(
+    runtime: &tokio::runtime::Runtime,
+    keyset: &KeysetCipher<'_, common::Observed>,
+    plan: &record::Plan,
+    operator: &str,
+    input: Value,
+) -> Operand {
+    let term = wait(
+        runtime,
+        dynamic::query_term(
+            keyset,
+            plan,
+            "profile",
+            OpSpec::parse(operator).unwrap(),
+            frame_of(&input),
+            &Eql,
+        )
+        .unwrap(),
+    )
+    .unwrap();
+    term.operand().clone()
+}
+
+fn eql_text(operand: &Operand) -> String {
+    match operand {
+        Operand::One(bytes) => String::from_utf8(bytes.clone()).unwrap(),
+        other => panic!("one EQL operand, got {other:?}"),
+    }
+}
+
+/// Needs an EQL v4 installation. The customer's predicate, a Sibling with
+/// more than five calls in one contact: the element predicate answers it,
+/// where `@>` and an ordering over `json_tree_path_query` answer "some
+/// contact is a Sibling and some has more than five calls". A row with a
+/// contact past the bound fails the query. `json_tree_subtree` is the
+/// indexed read without `k`, and `k` pairs a contact's two fields.
+#[test]
+#[ignore = "requires EQL_TEST_DATABASE_URL pointing to an EQL v4 database"]
+fn postgres_an_element_predicate_holds_of_one_element() {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let (cipher, _) = wait(&runtime, common::cipher());
+    let keyset = cipher.default_keyset();
+    let plan = contact_plan();
+    let mut db = connect();
+    let contacts = |list: Value| json!({"name": "x", "extra": {"emergency_contacts": list}});
+    let documents = [
+        // Each contact holds one half of the predicate.
+        contacts(json!([
+            {"relationship": "Sibling", "previous_call_count": 3},
+            {"relationship": "Friend", "previous_call_count": 7}
+        ])),
+        // One contact holds both.
+        contacts(json!([
+            {"relationship": "Friend", "previous_call_count": 9},
+            {"relationship": "Sibling", "previous_call_count": 7}
+        ])),
+        // Neither.
+        contacts(json!([{"relationship": "Friend", "previous_call_count": 9}])),
+        // No contacts.
+        json!({"name": "y"}),
+        // Folded equality: "sibling" ties with "Sibling".
+        contacts(json!([{"relationship": "sibling", "previous_call_count": 6}])),
+    ];
+    store(
+        &mut db,
+        &runtime,
+        &cipher,
+        "json_tree_elements",
+        &contact_options(),
+        &documents,
+    );
+    let ids = |db: &mut postgres::Client, sql: &str, params: &[&(dyn postgres::types::ToSql + Sync)]| -> Vec<i32> {
+        db.query(sql, params)
+            .unwrap_or_else(|error| panic!("{sql}: {error}"))
+            .iter()
+            .map(|row| row.get(0))
+            .collect()
+    };
+
+    let predicate = |bound: u32| {
+        eql_text(&derive(
+            &runtime,
+            &keyset,
+            &plan,
+            "element_match",
+            json!({"path": "$.extra.emergency_contacts[*]", "wildcard_bound": bound, "predicates": [
+                {"member": "relationship", "op": "eq", "json": "\"Sibling\""},
+                {"member": "previous_call_count", "op": "gt", "json": "5"}
+            ]}),
+        ))
+    };
+    let element_scoped = "SELECT id FROM json_tree_elements \
+         WHERE eql_v4.json_tree_element_match(doc, $1::text::jsonb::eql_v4.query_json_elements) ORDER BY id";
+    assert_eq!(
+        ids(&mut db, element_scoped, &[&predicate(4)]),
+        [2, 5],
+        "one contact a Sibling with more than five calls, the folded spelling included"
+    );
+
+    // The superset: containment on the relationship, an ordering over every
+    // call count.
+    let needle = eql_text(&derive(
+        &runtime,
+        &keyset,
+        &plan,
+        "eq_at",
+        json!({"path": RELATIONSHIP, "json": "\"Sibling\""}),
+    ));
+    let Operand::Text(calls_selector) = derive(&runtime, &keyset, &plan, "selector", json!({"path": CALLS}))
+    else {
+        panic!("a selector is text")
+    };
+    let five = eql_text(&derive(
+        &runtime,
+        &keyset,
+        &plan,
+        "gt_at",
+        json!({"path": CALLS, "json": "5"}),
+    ));
+    let superset = "SELECT id FROM json_tree_elements \
+         WHERE doc @> $1::text::jsonb::eql_v4.query_json \
+           AND EXISTS (SELECT 1 FROM eql_v4.json_tree_path_query(doc, $2::text) AS e \
+                        WHERE e > $3::text::jsonb::eql_v4.query_json_leaf) ORDER BY id";
+    assert_eq!(
+        ids(&mut db, superset, &[&needle, &calls_selector, &five]),
+        [1, 2],
+        "some contact a Sibling and some over five: row 1's halves answer it, and \
+         containment is exact, so the folded spelling does not"
+    );
+
+    // A bound below the longest array: the probe names the element at it.
+    let error = db.query(element_scoped, &[&predicate(1)]).unwrap_err();
+    assert_eq!(
+        error.code(),
+        Some(&postgres::error::SqlState::INVALID_PARAMETER_VALUE),
+        "{error}"
+    );
+    let message = error.as_db_error().map(|db| db.message()).unwrap_or_default();
+    assert!(
+        message.contains("longer than the bound"),
+        "the message says why: {message}"
+    );
+    // At exactly the longest array's length, it answers.
+    assert_eq!(ids(&mut db, element_scoped, &[&predicate(2)]), [2, 5]);
+
+    // `json_tree_subtree` is the indexed read without `k`, and the `k`-th
+    // token of each field's read names a field of one contact.
+    let tokens = |path: &str| match derive(
+        &runtime,
+        &keyset,
+        &plan,
+        "subtree",
+        json!({"path": path, "wildcard_bound": 2}),
+    ) {
+        Operand::TextList(tokens) => tokens,
+        other => panic!("{other:?}"),
+    };
+    let (relationships, counts) = (tokens(RELATIONSHIP), tokens(CALLS));
+    for budget in [0, 1, 64] {
+        for id in 1..=5 {
+            let plain: Vec<String> = db
+                .query(
+                    "SELECT r::jsonb::text FROM json_tree_elements, eql_v4.json_tree_subtree(doc, $1::text[], $2::int) AS r WHERE id = $3",
+                    &[&relationships, &budget, &id],
+                )
+                .unwrap()
+                .iter()
+                .map(|row| row.get(0))
+                .collect();
+            let indexed: Vec<String> = db
+                .query(
+                    "SELECT r.entry::jsonb::text FROM json_tree_elements, eql_v4.json_tree_subtree_indexed(doc, $1::text[], $2::int) AS r WHERE id = $3",
+                    &[&relationships, &budget, &id],
+                )
+                .unwrap()
+                .iter()
+                .map(|row| row.get(0))
+                .collect();
+            assert_eq!(plain, indexed, "row {id}, budget {budget}");
+        }
+    }
+    let paired: Vec<(i32, i32)> = db
+        .query(
+            "SELECT r.k, c.k FROM json_tree_elements,
+                    eql_v4.json_tree_subtree_indexed(doc, $1::text[], 0) AS r
+               JOIN eql_v4.json_tree_subtree_indexed(doc, $2::text[], 0) AS c ON c.k = r.k
+              WHERE id = 2 ORDER BY r.k",
+            &[&relationships, &counts],
+        )
+        .unwrap()
+        .iter()
+        .map(|row| (row.get(0), row.get(1)))
+        .collect();
+    assert_eq!(paired, [(1, 1), (2, 2)], "one pair per contact, by k");
+    let fallback: Vec<i32> = db
+        .query(
+            "SELECT r.k FROM json_tree_elements, eql_v4.json_tree_subtree_indexed(doc, $1::text[], 1) AS r WHERE id = 2",
+            &[&relationships],
+        )
+        .unwrap()
+        .iter()
+        .map(|row| row.get(0))
+        .collect();
+    assert_eq!(fallback, [0], "past the budget, the sealed root alone, at k = 0");
+
+    // A member whose full path the field does not order is refused before
+    // anything is derived.
+    let error = dynamic::check_query(
+        &plan,
+        "profile",
+        OpSpec::ElementMatch,
+        frame_of(&json!({"path": "$.extra.emergency_contacts[*]", "predicates": [
+            {"member": "name", "op": "eq", "json": "\"x\""}
+        ]})),
+        &Eql,
+    )
+    .unwrap_err();
+    assert!(
+        matches!(
+            error,
+            dynamic::Error::Cipher(stack_encrypt::Error::Term(
+                stack_encrypt::sem::TermError::JsonPathNotOrdered
+            ))
+        ),
+        "{error:?}"
+    );
+}