diff --git a/DEFAULT_CONFIG.json5 b/DEFAULT_CONFIG.json5
index 77d03f0e1f..3c4180380c 100644
--- a/DEFAULT_CONFIG.json5
+++ b/DEFAULT_CONFIG.json5
@@ -349,6 +349,29 @@
   //   publishers: [
   //     // key_expression
   //   ],
+  //   /// Router upstream (northbound) declaration aggregation. Opt-in; effective only on a
+  //   /// router's north-bound forwarding (distinct from the session-local subscribers/publishers
+  //   /// above). A downstream subscriber/queryable whose key-expression is included by one of
+  //   /// these prefixes is folded into a single `${prefix}` declaration toward the upstream and
+  //   /// suppressed upstream (kept locally for downward routing), so an upstream router holds one
+  //   /// resource per configured prefix instead of one per forwarded child key-expression. The
+  //   /// folded queryable is advertised complete=false (presence, not authority); `target=AllComplete`
+  //   /// queries are still forwarded through the aggregate to the real child queryables.
+  //   /// Trade-offs when enabled — apply per-key policy/introspection at the router that performs the
+  //   /// fold (where child keys are still visible): upstream per-key declaration ACL/QoS/interceptors
+  //   /// and admin-space enumeration see only the `${prefix}` aggregate; the wildcard aggregate may
+  //   /// forward data toward the folding router for unsubscribed keys; liveliness tokens are NOT
+  //   /// folded. Each prefix should cover the key-expression sub-tree a single downstream branch owns;
+  //   /// prefixes should be disjoint, not a bare `**` root, and not under the `@/` admin key-space
+  //   /// (misconfigured prefixes are warned about at startup).
+  //   upstream: {
+  //     subscribers: [
+  //       // key_expression
+  //     ],
+  //     queryables: [
+  //       // key_expression
+  //     ],
+  //   },
   // },
 
   // /// Namespace prefix.
diff --git a/commons/zenoh-config/src/lib.rs b/commons/zenoh-config/src/lib.rs
index 846a10ae5b..69d5cafbcb 100644
--- a/commons/zenoh-config/src/lib.rs
+++ b/commons/zenoh-config/src/lib.rs
@@ -655,6 +655,44 @@ validated_struct::validator! {
             subscribers: Vec<OwnedKeyExpr>,
             /// A list of key-expressions for which all included publishers will be aggregated into.
             publishers: Vec<OwnedKeyExpr>,
+            /// Router upstream (northbound) declaration aggregation. **Opt-in** and effective only on
+            /// a router's north-bound forwarding — distinct from the session-local `subscribers`/
+            /// `publishers` above. When a north-bound router HAT forwards a downstream subscriber or
+            /// queryable whose key-expression is *included* by one of these prefixes, the per-key
+            /// children are folded into a single `${prefix}` declaration toward the upstream and the
+            /// children are suppressed upstream (they remain registered locally for downward
+            /// routing), so an upstream router holds one routing resource per configured prefix
+            /// instead of one per forwarded child key-expression. The folded queryable is advertised
+            /// `complete=false` (presence, not authority) so it never shadows a distinct complete
+            /// source; `target=AllComplete` queries are still forwarded through the aggregate to the
+            /// real child queryables. Empty by default → behaviour is identical to stock.
+            ///
+            /// **Trade-offs when enabled — apply per-key policy and introspection at the router that
+            /// performs the fold (downstream of the aggregation point), where the child key-expressions
+            /// are still visible:**
+            /// - Per-key declaration ACL / QoS-overwrite / interceptors at the *upstream* node act on
+            ///   the `${prefix}` aggregate, not the individual child keys (granularity degrades to the
+            ///   prefix).
+            /// - Upstream admin-space enumeration (`@/${zid}/router/subscriber|queryable/**`) shows the
+            ///   aggregate, not the per-key children.
+            /// - The aggregate is a wildcard, so data published to *unsubscribed* keys under the prefix
+            ///   is still forwarded toward the folding router (which drops it); keep prefixes as tight
+            ///   as the actually-subscribed sub-tree.
+            /// - Only subscribers and queryables are folded. Liveliness tokens are intentionally NOT
+            ///   folded: a liveliness sample's key-expression *is* the token's own key (there is no
+            ///   reply fan-out to recover per-key identity, unlike a wildcard queryable `get`), so a
+            ///   single `${prefix}/**` token could neither enumerate the live per-key set nor emit a
+            ///   per-key removal when one child token is undeclared.
+            /// - Each prefix should cover the key-expression sub-tree a single downstream branch owns;
+            ///   prefixes should be disjoint, not a bare `**` root, and not under the `@/` admin
+            ///   key-space (misconfigured prefixes are warned about at startup).
+            pub upstream: #[derive(Default)]
+            UpstreamAggregationConf {
+                /// Key-expression prefixes whose downstream subscribers are folded upstream.
+                subscribers: Vec<OwnedKeyExpr>,
+                /// Key-expression prefixes whose downstream queryables are folded upstream.
+                queryables: Vec<OwnedKeyExpr>,
+            },
         },
 
         /// Overwrite QoS options for Zenoh messages by key expression (ignores Zenoh API QoS config)
diff --git a/zenoh/src/net/routing/dispatcher/queries.rs b/zenoh/src/net/routing/dispatcher/queries.rs
index f32b62f06f..6e0a546f2d 100644
--- a/zenoh/src/net/routing/dispatcher/queries.rs
+++ b/zenoh/src/net/routing/dispatcher/queries.rs
@@ -369,10 +369,12 @@ impl Face {
                 }
             }
             QueryTarget::AllComplete => {
-                for qabl in qabls
-                    .iter()
-                    .filter(|q| q.info.is_none_or(|info| info.complete) && filter(q))
-                {
+                for qabl in qabls.iter().filter(|q| {
+                    // B-A2: also forward to transparent forwarders (non-complete entries toward a
+                    // router whose resource covers the query, e.g. a northbound aggregate hiding
+                    // complete children) so the next router re-applies AllComplete against them.
+                    (q.info.is_none_or(|info| info.complete) || q.forwarder) && filter(q)
+                }) {
                     route.insert(qabl.dir.dst_face.id, || {
                         let mut dir = qabl.dir.clone();
                         let rid = insert_pending_query(&mut dir.dst_face, query.clone());
diff --git a/zenoh/src/net/routing/dispatcher/resource.rs b/zenoh/src/net/routing/dispatcher/resource.rs
index 47efeec034..2389a15ec9 100644
--- a/zenoh/src/net/routing/dispatcher/resource.rs
+++ b/zenoh/src/net/routing/dispatcher/resource.rs
@@ -86,6 +86,14 @@ pub(crate) struct QueryTargetQabl {
     pub(crate) dir: Direction,
     pub(crate) info: Option<QueryableInfoType>,
     pub(crate) region: Region,
+    /// A *transparent forwarder*: a non-complete route entry toward a router whose resource
+    /// covers the query key-expr (e.g. a northbound `${prefix}/**` aggregate that suppresses its
+    /// per-key — possibly `complete=true` — children upstream). `QueryTarget::AllComplete` forwards
+    /// to such entries so the next router can re-apply the filter against its real children;
+    /// forwarding to a genuinely-incomplete remote is harmless (dropped one hop later). It never
+    /// affects `BestMatching` (which only short-circuits on `info.complete`), so it cannot shadow a
+    /// distinct complete source. Always `false` for terminal/face-local queryables.
+    pub(crate) forwarder: bool,
 }
 
 impl QueryTargetQabl {
@@ -109,6 +117,7 @@ impl QueryTargetQabl {
                 distance: if ctx.face.is_local { 0 } else { 1 },
             }),
             region: *region,
+            forwarder: false,
         })
     }
 }
diff --git a/zenoh/src/net/routing/dispatcher/tables.rs b/zenoh/src/net/routing/dispatcher/tables.rs
index 12fe09ea4d..decca5596b 100644
--- a/zenoh/src/net/routing/dispatcher/tables.rs
+++ b/zenoh/src/net/routing/dispatcher/tables.rs
@@ -25,7 +25,7 @@ use std::{
 
 use uhlc::HLC;
 use zenoh_config::{unwrap_or_default, Config};
-use zenoh_keyexpr::keyexpr;
+use zenoh_keyexpr::{keyexpr, OwnedKeyExpr};
 use zenoh_protocol::{
     core::{Bound, ExprId, Region, WireExpr, ZenohIdProto},
     network::Mapping,
@@ -145,6 +145,16 @@ pub(crate) struct TablesData {
 
     pub(crate) hats: RegionMap<HatTablesData>,
     pub(crate) routes_version: RoutesVersion,
+
+    /// Northbound declaration aggregation (`aggregation.upstream`). Prefixes whose downstream
+    /// subscribers/queryables a north-bound router HAT folds into one `${prefix}` declaration
+    /// upstream. `*_resources` are the corresponding aggregate `Resource`s, pre-created once at
+    /// gateway build (parallel to the prefix vecs) so the fold path never creates resources and the
+    /// aggregate's match-set is wired exactly once. Empty ⇒ behaviour identical to stock zenoh.
+    pub(crate) agg_sub_prefixes: Vec<OwnedKeyExpr>,
+    pub(crate) agg_sub_resources: Vec<Arc<Resource>>,
+    pub(crate) agg_qabl_prefixes: Vec<OwnedKeyExpr>,
+    pub(crate) agg_qabl_resources: Vec<Arc<Resource>>,
 }
 
 impl Debug for TablesData {
@@ -184,6 +194,8 @@ impl TablesData {
             Duration::from_millis(unwrap_or_default!(config.queries_default_timeout()));
         let interests_timeout =
             Duration::from_millis(unwrap_or_default!(config.routing().interests().timeout()));
+        let agg_sub_prefixes = config.aggregation().upstream().subscribers().clone();
+        let agg_qabl_prefixes = config.aggregation().upstream().queryables().clone();
         #[cfg(feature = "stats")]
         let mut stats_keys = zenoh_stats::StatsKeysTree::default();
         #[cfg(feature = "stats")]
@@ -210,6 +222,10 @@ impl TablesData {
             #[cfg(feature = "stats")]
             stats,
             routes_version: 0,
+            agg_sub_prefixes,
+            agg_sub_resources: Vec::new(),
+            agg_qabl_prefixes,
+            agg_qabl_resources: Vec::new(),
         })
     }
 
@@ -294,6 +310,130 @@ pub struct Tables {
 }
 
 impl Tables {
+    /// Pre-create the aggregate `Resource` for every configured `aggregation.upstream` prefix, once,
+    /// at gateway build (where the full `Tables` is available for `make_resource`). The northbound
+    /// fold path then only *looks these up* — it never creates resources — and because each
+    /// aggregate persists for the gateway's lifetime its match-set is wired exactly once, so
+    /// reconnect/churn cannot accumulate duplicate match cross-links. No-op when unconfigured.
+    pub(crate) fn precreate_upstream_aggregates(&mut self) {
+        self.warn_on_aggregation_misconfig();
+        // Create + wire each UNIQUE prefix's aggregate exactly once (a prefix may be listed in both
+        // `subscribers` and `queryables`), then fill the parallel resource vecs by lookup. This
+        // avoids re-running `match_resource` on a shared prefix (which would duplicate match links).
+        let mut by_ke: HashMap<OwnedKeyExpr, Arc<Resource>> = HashMap::new();
+        let prefixes: Vec<OwnedKeyExpr> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .chain(self.data.agg_qabl_prefixes.iter())
+            .cloned()
+            .collect();
+        for ke in &prefixes {
+            if !by_ke.contains_key(ke) {
+                let res = self.make_aggregate_resource(ke);
+                by_ke.insert(ke.clone(), res);
+            }
+        }
+        let sub_res: Vec<Arc<Resource>> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .map(|ke| by_ke[ke].clone())
+            .collect();
+        let qabl_res: Vec<Arc<Resource>> = self
+            .data
+            .agg_qabl_prefixes
+            .iter()
+            .map(|ke| by_ke[ke].clone())
+            .collect();
+        self.data.agg_sub_resources = sub_res;
+        self.data.agg_qabl_resources = qabl_res;
+    }
+
+    /// Emit operator-facing warnings for suspicious `aggregation.upstream` prefixes (root wildcards,
+    /// admin-space prefixes, duplicates, mutual inclusion) plus a one-time note about the trade-offs.
+    /// Soft: never rejects config — folding still applies to the prefixes exactly as given.
+    fn warn_on_aggregation_misconfig(&self) {
+        let all: Vec<(&str, &OwnedKeyExpr)> = self
+            .data
+            .agg_sub_prefixes
+            .iter()
+            .map(|k| ("subscribers", k))
+            .chain(
+                self.data
+                    .agg_qabl_prefixes
+                    .iter()
+                    .map(|k| ("queryables", k)),
+            )
+            .collect();
+        if all.is_empty() {
+            return;
+        }
+        tracing::info!(
+            "northbound aggregation enabled for {} prefix declaration(s): upstream per-key declaration \
+             ACL/QoS/interceptors and upstream admin-space enumeration see only the ${{prefix}} aggregate, \
+             and the wildcard aggregate may forward data toward the edge for unsubscribed keys — apply \
+             per-key policy and per-key introspection at the owning edge",
+            all.len()
+        );
+        for (kind, ke) in &all {
+            let s = ke.as_str();
+            if s == "**" || s == "*" {
+                tracing::warn!(
+                    "aggregation.upstream.{kind} prefix `{s}` is a root wildcard and would fold EVERY \
+                     matching declaration into one aggregate — almost certainly a misconfiguration"
+                );
+            }
+            if s == "@" || s.starts_with("@/") {
+                tracing::warn!(
+                    "aggregation.upstream.{kind} prefix `{s}` targets the `@/` admin key-space; folding \
+                     admin declarations breaks per-node introspection — almost certainly a misconfiguration"
+                );
+            }
+        }
+        // Duplicates / mutual inclusion: matching is first-match, so a narrower prefix listed after a
+        // broader one that includes it never receives children (its aggregate stays empty).
+        for i in 0..all.len() {
+            for j in (i + 1)..all.len() {
+                let (a, b) = (all[i].1, all[j].1);
+                if a == b {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` is listed more than once (redundant)",
+                        a.as_str()
+                    );
+                } else if a.includes(b) {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` includes `{}`: children of the narrower prefix \
+                         fold into the broader one (first match), so `{}` may never aggregate anything",
+                        a.as_str(),
+                        b.as_str(),
+                        b.as_str()
+                    );
+                } else if b.includes(a) {
+                    tracing::warn!(
+                        "aggregation.upstream prefix `{}` includes `{}`: children of the narrower prefix \
+                         fold into the broader one (first match), so `{}` may never aggregate anything",
+                        b.as_str(),
+                        a.as_str(),
+                        a.as_str()
+                    );
+                }
+            }
+        }
+    }
+
+    fn make_aggregate_resource(&mut self, ke: &OwnedKeyExpr) -> Arc<Resource> {
+        // Compute matches BEFORE creating the resource so it does not appear in its own match-set
+        // (mirrors `Face::with_mapped_expr`), then wire it in so a wildcard route resolving to the
+        // aggregate still reaches the (later-declared) children.
+        let mut matches = Resource::get_matches(&self.data, ke);
+        let mut root = self.data.root_res.clone();
+        let mut res = Resource::make_resource(self, &mut root, ke.as_str());
+        matches.push(Arc::downgrade(&res));
+        Resource::match_resource(&self.data, &mut res, matches);
+        res
+    }
+
     pub(crate) fn sourced_subscribers(&self) -> HashMap<Arc<Resource>, Sources> {
         self.hats
             .values()
diff --git a/zenoh/src/net/routing/gateway.rs b/zenoh/src/net/routing/gateway.rs
index ad86398622..33a3513db3 100644
--- a/zenoh/src/net/routing/gateway.rs
+++ b/zenoh/src/net/routing/gateway.rs
@@ -188,9 +188,14 @@ impl<'conf> GatewayBuilder<'conf> {
             stats,
         )?;
 
+        let mut tables = Tables { data, hats };
+        // Pre-create the northbound-aggregation aggregate resources (needs the fully-assembled
+        // `Tables`); no-op unless `aggregation.upstream` is configured.
+        tables.precreate_upstream_aggregates();
+
         Ok(Gateway {
             tables: Arc::new(TablesLock {
-                tables: RwLock::new(Tables { data, hats }),
+                tables: RwLock::new(tables),
                 ctrl_lock: Mutex::new(()),
                 queries_lock: RwLock::new(()),
             }),
diff --git a/zenoh/src/net/routing/hat/broker/queries.rs b/zenoh/src/net/routing/hat/broker/queries.rs
index 50c2ce36db..8ec04d2aff 100644
--- a/zenoh/src/net/routing/hat/broker/queries.rs
+++ b/zenoh/src/net/routing/hat/broker/queries.rs
@@ -190,18 +190,17 @@ impl Hat {
             return false;
         };
 
-        let is_matching_res = |qabl: &Resource| {
+        let is_matching = |qabl: &Resource, info: &QueryableInfoType| {
             let Some(ke) = qabl.keyexpr() else {
                 bug!("Queryable resource should not be root");
                 return false;
             };
-
-            ke.includes(key_expr)
-        };
-        let is_matching_info = |info: &QueryableInfoType| !complete || info.complete;
-
-        let is_matching = |qabl: &Resource, info: &QueryableInfoType| {
-            is_matching_res(qabl) && is_matching_info(info)
+            // Under AllComplete (`complete`), match a complete queryable OR a strict wildcard
+            // superset (`ke != key_expr`) — a transparent forwarder (e.g. a northbound aggregate
+            // that suppresses its complete children upstream). A same-key non-complete queryable
+            // stays non-matching. Mirrors the AllComplete route-path forwarder in
+            // dispatcher::compute_final_route (B-A2).
+            ke.includes(key_expr) && (!complete || info.complete || ke != key_expr)
         };
 
         let compute_other_matches = || {
diff --git a/zenoh/src/net/routing/hat/client/queries.rs b/zenoh/src/net/routing/hat/client/queries.rs
index b682441856..6774162bdf 100644
--- a/zenoh/src/net/routing/hat/client/queries.rs
+++ b/zenoh/src/net/routing/hat/client/queries.rs
@@ -191,6 +191,7 @@ impl HatQueriesTrait for Hat {
                             node_id: DEFAULT_NODE_ID,
                         },
                         region: self.region,
+                        forwarder: false,
                     });
                 }
             }
diff --git a/zenoh/src/net/routing/hat/peer/queries.rs b/zenoh/src/net/routing/hat/peer/queries.rs
index cff92c1a80..7b3aad8982 100644
--- a/zenoh/src/net/routing/hat/peer/queries.rs
+++ b/zenoh/src/net/routing/hat/peer/queries.rs
@@ -317,6 +317,7 @@ impl HatQueriesTrait for Hat {
                         },
                         info: None,
                         region: self.region(),
+                        forwarder: false,
                     });
                 }
             }
@@ -334,6 +335,7 @@ impl HatQueriesTrait for Hat {
                     },
                     info: None,
                     region: self.region(),
+                    forwarder: false,
                 });
             }
         }
diff --git a/zenoh/src/net/routing/hat/router/mod.rs b/zenoh/src/net/routing/hat/router/mod.rs
index ccc5bd9ea8..73b93a8a94 100644
--- a/zenoh/src/net/routing/hat/router/mod.rs
+++ b/zenoh/src/net/routing/hat/router/mod.rs
@@ -105,6 +105,14 @@ pub(crate) struct Hat {
     router_qabls: HashSet<Arc<Resource>>,
     routers_net: Option<Network>,
     routers_trees_worker: TreesComputationWorker,
+    /// Northbound declaration aggregation: the per-key children folded into each configured
+    /// `aggregation.upstream` prefix, keyed by index into `TablesData::agg_sub_prefixes` /
+    /// `agg_qabl_prefixes`. The aggregate `Resource` itself is pre-created in `TablesData`; here we
+    /// only track folded children so the aggregate is declared once (first fold) and withdrawn on
+    /// the last. Queryable children additionally carry their `QueryableInfoType` so the aggregate's
+    /// advertised info is the merge of its children. Empty ⇒ stock behaviour.
+    north_agg_subs: HashMap<usize, HashSet<Arc<Resource>>>,
+    north_agg_qabls: HashMap<usize, HashMap<Arc<Resource>, QueryableInfoType>>,
     #[cfg(test)]
     disable_async_tree_computation: bool,
 }
@@ -124,6 +132,8 @@ impl Hat {
             router_tokens: HashSet::new(),
             routers_net: None,
             routers_trees_worker: TreesComputationWorker::new(region),
+            north_agg_subs: HashMap::new(),
+            north_agg_qabls: HashMap::new(),
             #[cfg(test)]
             disable_async_tree_computation: false,
         }
diff --git a/zenoh/src/net/routing/hat/router/pubsub.rs b/zenoh/src/net/routing/hat/router/pubsub.rs
index 00319ad0ee..f145d4c018 100644
--- a/zenoh/src/net/routing/hat/router/pubsub.rs
+++ b/zenoh/src/net/routing/hat/router/pubsub.rs
@@ -273,6 +273,109 @@ impl Hat {
     }
 }
 
+impl Hat {
+    // ===== Northbound declaration aggregation (subscribers) ===================================
+
+    /// Index into `TablesData::agg_sub_prefixes` of the first configured prefix that *includes*
+    /// `res`'s key-expr — only on a north-bound HAT. `None` ⇒ stock per-key propagation. A child
+    /// whose key-expr *equals* the prefix folds too: it becomes a child of its own aggregate (which
+    /// advertises the same `${prefix}`), keeping a real exact-prefix declaration and the synthetic
+    /// aggregate on ONE refcounted path so neither can spuriously withdraw the other.
+    fn matching_agg_sub_prefix(&self, tables: &TablesData, res: &Arc<Resource>) -> Option<usize> {
+        if !self.region().bound().is_north() || tables.agg_sub_prefixes.is_empty() {
+            return None;
+        }
+        let child_ke = res.keyexpr()?;
+        tables
+            .agg_sub_prefixes
+            .iter()
+            .position(|prefix| prefix.includes(child_ke))
+    }
+
+    /// Fold a downstream subscriber into the pre-created `${prefix}` aggregate for `idx`: record the
+    /// child and, on the first child, declare the aggregate northbound (idempotent on `router_subs`).
+    fn fold_subscriber_upstream(
+        &mut self,
+        ctx: DispatcherContext,
+        idx: usize,
+        child: &Arc<Resource>,
+        sub_info: &SubscriberInfo,
+    ) {
+        let Some(mut aggregate) = ctx.tables.agg_sub_resources.get(idx).cloned() else {
+            return; // precreate keeps resources parallel to prefixes; defensive
+        };
+        self.north_agg_subs
+            .entry(idx)
+            .or_default()
+            .insert(child.clone());
+        if !self
+            .res_hat(&aggregate)
+            .router_subs
+            .contains(&ctx.tables.zid)
+        {
+            self.res_hat_mut(&mut aggregate)
+                .router_subs
+                .insert(ctx.tables.zid);
+            self.router_subs.insert(aggregate.clone());
+            self.disable_data_routes(&mut aggregate);
+            self.propagate_sourced_subscriber(
+                ctx.tables,
+                &aggregate,
+                sub_info,
+                None,
+                &ctx.tables.zid,
+            );
+        }
+    }
+
+    /// If `res` is a child folded into some northbound aggregate, remove it and — when the aggregate
+    /// has no remaining children — withdraw the aggregate. Returns `true` iff `res` was a folded
+    /// child (caller must then skip the stock per-key undeclare path). The aggregate `Resource`
+    /// itself persists (pre-created), so its match-set is never rebuilt.
+    fn remove_subscriber_from_aggregate(
+        &mut self,
+        ctx: DispatcherContext,
+        res: &Arc<Resource>,
+    ) -> bool {
+        let mut hit: Option<usize> = None;
+        for (idx, children) in self.north_agg_subs.iter_mut() {
+            if children.remove(res) {
+                hit = Some(*idx);
+                break;
+            }
+        }
+        let Some(idx) = hit else {
+            return false;
+        };
+        let empty = self.north_agg_subs.get(&idx).map_or(true, |c| c.is_empty());
+        if empty {
+            self.north_agg_subs.remove(&idx);
+            if let Some(mut aggregate) = ctx.tables.agg_sub_resources.get(idx).cloned() {
+                if self
+                    .res_hat(&aggregate)
+                    .router_subs
+                    .contains(&ctx.tables.zid)
+                {
+                    self.res_hat_mut(&mut aggregate)
+                        .router_subs
+                        .remove(&ctx.tables.zid);
+                    if self.res_hat(&aggregate).router_subs.is_empty() {
+                        self.router_subs.retain(|r| !Arc::ptr_eq(r, &aggregate));
+                    }
+                    self.disable_data_routes(&mut aggregate);
+                    self.propagate_forget_sourced_subscriber(
+                        ctx.tables,
+                        &aggregate,
+                        None,
+                        &ctx.tables.zid,
+                    );
+                }
+            }
+        }
+        true
+    }
+}
+
 impl HatPubSubTrait for Hat {
     #[tracing::instrument(level = "debug", skip(tables), ret)]
     fn sourced_subscribers(&self, tables: &TablesData) -> HashMap<Arc<Resource>, Sources> {
@@ -470,6 +573,17 @@ impl HatPubSubTrait for Hat {
             return;
         };
 
+        // Northbound aggregation: if this north-bound router HAT forwards a downstream subscriber
+        // whose key-expr is included by a configured `aggregation.upstream.subscribers` prefix, fold
+        // it into the pre-created `${prefix}` aggregate and suppress the per-key child upstream. The
+        // child is intentionally NOT inserted into `router_subs`, so `pubsub_tree_change` /
+        // `sourced_subscribers` cannot re-leak it; downward routing is unaffected (the child stays
+        // registered in the source region's HAT).
+        if let Some(idx) = self.matching_agg_sub_prefix(ctx.tables, &res) {
+            self.fold_subscriber_upstream(ctx, idx, &res, &other_info);
+            return;
+        }
+
         if !self.res_hat(&res).router_subs.contains(&ctx.tables.zid) {
             self.res_hat_mut(&mut res)
                 .router_subs
@@ -481,12 +595,19 @@ impl HatPubSubTrait for Hat {
     }
 
     #[tracing::instrument(level = "debug", skip(ctx), ret)]
-    fn unpropagate_subscriber(&mut self, ctx: DispatcherContext, mut res: Arc<Resource>) {
+    fn unpropagate_subscriber(&mut self, mut ctx: DispatcherContext, mut res: Arc<Resource>) {
         if self.owns(ctx.src_face) {
             // NOTE(regions): see Hat::unregister_subscriber
             return;
         }
 
+        // Northbound aggregation: a folded child was never propagated per-key (so the
+        // `debug_assert!(was_propagated)` below would not hold for it). Remove it from its aggregate
+        // — withdrawing the aggregate when its last child is gone — and skip the stock path.
+        if self.remove_subscriber_from_aggregate(ctx.reborrow(), &res) {
+            return;
+        }
+
         let was_propagated = self.res_hat(&res).router_subs.contains(&ctx.tables.zid);
 
         debug_assert!(was_propagated);
diff --git a/zenoh/src/net/routing/hat/router/queries.rs b/zenoh/src/net/routing/hat/router/queries.rs
index 5dc3508126..3b86360c32 100644
--- a/zenoh/src/net/routing/hat/router/queries.rs
+++ b/zenoh/src/net/routing/hat/router/queries.rs
@@ -277,6 +277,144 @@ impl Hat {
     }
 }
 
+impl Hat {
+    // ===== Northbound declaration aggregation (queryables) ====================================
+
+    /// See `matching_agg_sub_prefix`. North-bound HATs only; a child equal to the prefix folds too.
+    fn matching_agg_qabl_prefix(&self, tables: &TablesData, res: &Arc<Resource>) -> Option<usize> {
+        if !self.region().bound().is_north() || tables.agg_qabl_prefixes.is_empty() {
+            return None;
+        }
+        let child_ke = res.keyexpr()?;
+        tables
+            .agg_qabl_prefixes
+            .iter()
+            .position(|prefix| prefix.includes(child_ke))
+    }
+
+    /// The aggregate's advertised info. An aggregate advertises the *presence* of queryables under
+    /// the prefix, NOT a completeness claim over the whole prefix: asserting `complete` would require
+    /// a single sole authority for the entire prefix and is prone to staleness as owners come and go
+    /// (the single-remaining-owner undeclare path does not refresh folded state). It is therefore
+    /// always advertised **`complete=false`**, so `BestMatching` fans out to it (the router that
+    /// performed the fold), which routes on to the real per-key queryable, and a distinct complete
+    /// source is never shadowed. (When the prefix has a single owner this is behaviourally identical
+    /// to asserting complete, since that router is the only matching source.) `distance` is the
+    /// minimum over the folded children.
+    fn aggregate_qabl_info(&self, idx: usize) -> QueryableInfoType {
+        let distance = self
+            .north_agg_qabls
+            .get(&idx)
+            .into_iter()
+            .flat_map(|children| children.values().map(|i| i.distance))
+            .min()
+            .unwrap_or(0);
+        QueryableInfoType {
+            complete: false,
+            distance,
+        }
+    }
+
+    fn fold_queryable_upstream(
+        &mut self,
+        ctx: DispatcherContext,
+        idx: usize,
+        child: &Arc<Resource>,
+        child_info: &QueryableInfoType,
+    ) {
+        let Some(mut aggregate) = ctx.tables.agg_qabl_resources.get(idx).cloned() else {
+            return; // defensive: precreate keeps resources parallel to prefixes
+        };
+        self.north_agg_qabls
+            .entry(idx)
+            .or_default()
+            .insert(child.clone(), *child_info);
+        let agg_info = self.aggregate_qabl_info(idx);
+        if self.res_hat(&aggregate).router_qabls.get(&ctx.tables.zid) != Some(&agg_info) {
+            self.res_hat_mut(&mut aggregate)
+                .router_qabls
+                .insert(ctx.tables.zid, agg_info);
+            self.router_qabls.insert(aggregate.clone());
+            self.disable_query_routes(&mut aggregate);
+            self.propagate_sourced_queryable(
+                ctx.tables,
+                &aggregate,
+                &agg_info,
+                None,
+                &ctx.tables.zid,
+            );
+        }
+    }
+
+    /// If `res` is a child folded into some northbound aggregate, remove it; refresh the aggregate's
+    /// merged info when children remain, or withdraw it on the last child. Returns `true` iff `res`
+    /// was a folded child. The aggregate `Resource` persists (pre-created).
+    fn remove_queryable_from_aggregate(
+        &mut self,
+        ctx: DispatcherContext,
+        res: &Arc<Resource>,
+    ) -> bool {
+        let mut hit: Option<usize> = None;
+        for (idx, children) in self.north_agg_qabls.iter_mut() {
+            if children.remove(res).is_some() {
+                hit = Some(*idx);
+                break;
+            }
+        }
+        let Some(idx) = hit else {
+            return false;
+        };
+        let Some(mut aggregate) = ctx.tables.agg_qabl_resources.get(idx).cloned() else {
+            return true;
+        };
+        if self
+            .north_agg_qabls
+            .get(&idx)
+            .map_or(true, |c| c.is_empty())
+        {
+            // Last child gone: withdraw the aggregate.
+            self.north_agg_qabls.remove(&idx);
+            if self
+                .res_hat(&aggregate)
+                .router_qabls
+                .contains_key(&ctx.tables.zid)
+            {
+                self.res_hat_mut(&mut aggregate)
+                    .router_qabls
+                    .remove(&ctx.tables.zid);
+                if self.res_hat(&aggregate).router_qabls.is_empty() {
+                    self.router_qabls.retain(|r| !Arc::ptr_eq(r, &aggregate));
+                }
+                self.disable_query_routes(&mut aggregate);
+                self.propagate_forget_sourced_queryable(
+                    ctx.tables,
+                    &aggregate,
+                    None,
+                    &ctx.tables.zid,
+                );
+            }
+        } else {
+            // Children remain: refresh the aggregate's merged info if it changed.
+            let agg_info = self.aggregate_qabl_info(idx);
+            if self.res_hat(&aggregate).router_qabls.get(&ctx.tables.zid) != Some(&agg_info) {
+                self.res_hat_mut(&mut aggregate)
+                    .router_qabls
+                    .insert(ctx.tables.zid, agg_info);
+                self.router_qabls.insert(aggregate.clone());
+                self.disable_query_routes(&mut aggregate);
+                self.propagate_sourced_queryable(
+                    ctx.tables,
+                    &aggregate,
+                    &agg_info,
+                    None,
+                    &ctx.tables.zid,
+                );
+            }
+        }
+        true
+    }
+}
+
 impl HatQueriesTrait for Hat {
     #[tracing::instrument(level = "debug", skip(tables), ret)]
     fn sourced_queryables(&self, tables: &TablesData) -> HashMap<Arc<Resource>, Sources> {
@@ -366,6 +504,13 @@ impl HatQueriesTrait for Hat {
                                                         as u16,
                                                 }),
                                                 region: this.region,
+                                                // Transparent forwarder (B-A2): a non-complete entry
+                                                // whose matched resource covers the query (e.g. a
+                                                // northbound aggregate suppressing complete children).
+                                                // `complete` here = resource-includes-query; the dst
+                                                // is intrinsically a router (routers_net), so an
+                                                // AllComplete query is safely re-filtered downstream.
+                                                forwarder: complete && !qabl_info.complete,
                                             });
                                         }
                                     }
@@ -514,6 +659,16 @@ impl HatQueriesTrait for Hat {
             return;
         };
 
+        // Northbound aggregation (symmetric to subscribers): fold a downstream queryable included by
+        // a configured `aggregation.upstream.queryables` prefix into the pre-created `${prefix}`
+        // aggregate and suppress the per-key child upstream. The aggregate is advertised
+        // non-complete (a presence advertisement, not a completeness authority over the prefix), so
+        // `BestMatching` fans out to it and never shadows a distinct source sharing the prefix.
+        if let Some(idx) = self.matching_agg_qabl_prefix(ctx.tables, &res) {
+            self.fold_queryable_upstream(ctx, idx, &res, &other_info);
+            return;
+        }
+
         if self
             .res_hat(&res)
             .router_qabls
@@ -530,12 +685,19 @@ impl HatQueriesTrait for Hat {
     }
 
     #[tracing::instrument(level = "debug", skip(ctx), ret)]
-    fn unpropagate_queryable(&mut self, ctx: DispatcherContext, mut res: Arc<Resource>) {
+    fn unpropagate_queryable(&mut self, mut ctx: DispatcherContext, mut res: Arc<Resource>) {
         if self.owns(ctx.src_face) {
             // NOTE(regions): see Hat::unregister_queryable
             return;
         }
 
+        // Northbound aggregation: a folded child was never propagated per-key. Remove it from its
+        // aggregate (refreshing the aggregate's merged info, or withdrawing it on the last child)
+        // and skip the stock path (the `debug_assert!(was_propagated)` would not hold for it).
+        if self.remove_queryable_from_aggregate(ctx.reborrow(), &res) {
+            return;
+        }
+
         let was_propagated = self
             .res_hat(&res)
             .router_qabls
diff --git a/zenoh/src/net/tests/regions/declare.rs b/zenoh/src/net/tests/regions/declare.rs
index 8572f0559e..fb9e251d9f 100644
--- a/zenoh/src/net/tests/regions/declare.rs
+++ b/zenoh/src/net/tests/regions/declare.rs
@@ -176,6 +176,85 @@ fn test_multiple_gateways_r2r_token_propagation_upstream() {
     assert!(s_g1.is_bi_complete());
 }
 
+/// **Northbound subscriber aggregation (deterministic).** A router configured with
+/// `aggregation.upstream.subscribers = ["example/**"]` folds K downstream subscribers included by
+/// the prefix into a SINGLE `example/**` declaration toward its north-bound peer, suppressing the
+/// per-key children upstream. Asserts the upstream face records exactly one `DeclareSubscriber`
+/// (not K), and that undeclaring every child withdraws exactly one aggregate.
+#[test]
+fn test_northbound_subscriber_aggregation() {
+    try_init_tracing_subscriber();
+
+    let g = HarnessBuilder::new()
+        .mode(WhatAmI::Router)
+        .subregions([Region::default_south(WhatAmI::Router)])
+        .aggregation_upstream_subscribers(["example/**"])
+        .build();
+    let s = HarnessBuilder::new()
+        .mode(WhatAmI::Router)
+        .subregions([Region::Local])
+        .build();
+    let n = HarnessBuilder::new()
+        .mode(WhatAmI::Router)
+        .subregions([Region::Local])
+        .build();
+
+    let ss = s.new_session();
+
+    // s is downstream (south) of g; n is upstream (north) of g.
+    let mut s_g = Connection {
+        a: &s,
+        b: &g,
+        a2b: FaceDef::default()
+            .mode(WhatAmI::Router)
+            .remote_bound(Bound::South),
+        b2a: FaceDef::default()
+            .mode(WhatAmI::Router)
+            .region(Region::default_south(WhatAmI::Router)),
+    }
+    .establish();
+    let mut n_g = Connection {
+        a: &n,
+        b: &g,
+        a2b: FaceDef::default().mode(WhatAmI::Router),
+        b2a: FaceDef::default().mode(WhatAmI::Router),
+    }
+    .establish();
+
+    // Inline (not a closure) so the `&mut` forwards are statement-scoped and the `recorder()` reads
+    // between forwarding rounds don't conflict with the mutable borrows.
+    EstablishedConnection::bi_fwd_many_unbounded([&mut s_g, &mut n_g]);
+
+    // K downstream subscribers, all included by the configured `example/**` prefix.
+    let k = 8u32;
+    for j in 0..k {
+        ss.declare_subscriber(None, j, format!("example/sensor/{j:04}"));
+    }
+    EstablishedConnection::bi_fwd_many_unbounded([&mut s_g, &mut n_g]);
+
+    // The upstream face (g -> n) must carry exactly ONE aggregate, not K per-key children.
+    let up = n_g.b2a.recorder().subscribers();
+    assert_eq!(
+        up.len(),
+        1,
+        "northbound fold must collapse {k} children into one aggregate upstream, got {}",
+        up.len()
+    );
+
+    // Undeclaring every child withdraws exactly one aggregate upstream.
+    for j in 0..k {
+        ss.undeclare_subscriber(j);
+    }
+    EstablishedConnection::bi_fwd_many_unbounded([&mut s_g, &mut n_g]);
+    let down = n_g.b2a.recorder().undeclared_subscribers();
+    assert_eq!(
+        down.len(),
+        1,
+        "withdrawing all children must withdraw exactly one aggregate upstream, got {}",
+        down.len()
+    );
+}
+
 #[test]
 fn test_multiple_gateways_r2r_token_propagation_downstream() {
     try_init_tracing_subscriber();
diff --git a/zenoh/src/net/tests/regions/mod.rs b/zenoh/src/net/tests/regions/mod.rs
index 3072d42230..398f28283a 100644
--- a/zenoh/src/net/tests/regions/mod.rs
+++ b/zenoh/src/net/tests/regions/mod.rs
@@ -729,6 +729,8 @@ pub(crate) struct HarnessBuilder {
     subregions: Vec<Region>,
     start_runtime: bool,
     start_adminspace: bool,
+    agg_upstream_subscribers: Vec<String>,
+    agg_upstream_queryables: Vec<String>,
 }
 
 impl HarnessBuilder {
@@ -739,9 +741,29 @@ impl HarnessBuilder {
             subregions: Vec::new(),
             start_runtime: true,
             start_adminspace: false,
+            agg_upstream_subscribers: Vec::new(),
+            agg_upstream_queryables: Vec::new(),
         }
     }
 
+    /// Configure `aggregation.upstream.subscribers` prefixes (northbound fold) for this router harness.
+    pub(crate) fn aggregation_upstream_subscribers<'a>(
+        mut self,
+        prefixes: impl IntoIterator<Item = &'a str>,
+    ) -> Self {
+        self.agg_upstream_subscribers = prefixes.into_iter().map(str::to_string).collect();
+        self
+    }
+
+    /// Configure `aggregation.upstream.queryables` prefixes (northbound fold) for this router harness.
+    pub(crate) fn aggregation_upstream_queryables<'a>(
+        mut self,
+        prefixes: impl IntoIterator<Item = &'a str>,
+    ) -> Self {
+        self.agg_upstream_queryables = prefixes.into_iter().map(str::to_string).collect();
+        self
+    }
+
     /// Set the [`WhatAmI`] mode of this harness.
     pub(crate) fn mode(mut self, mode: WhatAmI) -> Self {
         self.mode = mode;
@@ -796,6 +818,32 @@ impl HarnessBuilder {
                 .unwrap();
             config.plugins_loading.set_enabled(false).unwrap();
 
+            let to_arr = |v: &[String]| {
+                format!(
+                    "[{}]",
+                    v.iter()
+                        .map(|p| format!("\"{p}\""))
+                        .collect::<Vec<_>>()
+                        .join(",")
+                )
+            };
+            if !self.agg_upstream_subscribers.is_empty() {
+                config
+                    .insert_json5(
+                        "aggregation/upstream/subscribers",
+                        &to_arr(&self.agg_upstream_subscribers),
+                    )
+                    .unwrap();
+            }
+            if !self.agg_upstream_queryables.is_empty() {
+                config
+                    .insert_json5(
+                        "aggregation/upstream/queryables",
+                        &to_arr(&self.agg_upstream_queryables),
+                    )
+                    .unwrap();
+            }
+
             let runtime = block_on(
                 RuntimeBuilder::new(crate::api::config::Config(config))
                     .subregions(self.subregions)
diff --git a/zenoh/tests/regions/main.rs b/zenoh/tests/regions/main.rs
index 478d50748d..70334777d6 100644
--- a/zenoh/tests/regions/main.rs
+++ b/zenoh/tests/regions/main.rs
@@ -32,6 +32,8 @@ mod scenario6;
 mod scenario7;
 #[cfg(feature = "unstable")]
 mod scenario8;
+#[cfg(feature = "unstable")]
+mod upstream_agg;
 
 use std::{
     sync::{Arc, RwLock},
diff --git a/zenoh/tests/regions/upstream_agg.rs b/zenoh/tests/regions/upstream_agg.rs
new file mode 100644
index 0000000000..c50a239e72
--- /dev/null
+++ b/zenoh/tests/regions/upstream_agg.rs
@@ -0,0 +1,714 @@
+//
+// Copyright (c) 2026 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
+//
+
+//! Real-transport (loopback TCP) integration tests for **northbound declaration aggregation**
+//! (`aggregation.upstream.{subscribers,queryables}`): a north-bound router HAT folds a downstream
+//! session's per-key subscribers/queryables included by a configured prefix into a single
+//! `${prefix}` declaration toward the upstream, suppressing the per-key children upstream while
+//! keeping them locally for downward routing. An upstream router thus holds one Resource per prefix
+//! instead of one per forwarded child key-expression.
+//!
+//! These exercise the full downstream→upstream routing path over real sessions (matching the
+//! `scenario*` tests); deterministic in-process coverage is in
+//! `zenoh/src/net/tests/regions/declare.rs::test_northbound_subscriber_aggregation`.
+//! (For readability the tests below name the two routers `gateway`/`edge` and the upstream client
+//! `client`, and use `branch1/...` key-expressions — all purely illustrative; the feature itself is
+//! topology- and application-agnostic.)
+
+use std::time::Duration;
+
+use zenoh_config::WhatAmI::Router;
+use zenoh_core::ztimeout;
+
+use crate::{loc, Node};
+
+const TIMEOUT: Duration = Duration::from_secs(60);
+
+async fn count_admin(loc: &str, zid: &str, kind: &str, obs_id: &str, needle: &str) -> usize {
+    use zenoh_config::WhatAmI::Client;
+    let observer = ztimeout!(Node::new(Client, obs_id).connect(&[loc]).open());
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let sel = format!("@/{zid}/router/{kind}/**");
+    let replies = ztimeout!(observer.get(sel.as_str())).unwrap();
+    let mut n = 0usize;
+    while let Ok(reply) = replies.recv_async().await {
+        if let Ok(s) = reply.result() {
+            let ke = s.key_expr().as_str().to_string();
+            let payload = s
+                .payload()
+                .try_to_string()
+                .map(|c| c.into_owned())
+                .unwrap_or_default();
+            if ke.contains(needle) || payload.contains(needle) {
+                n += 1;
+            }
+        }
+    }
+    ztimeout!(observer.close()).unwrap();
+    n
+}
+
+/// Subscribers: K per-key subscribers on a downstream session collapse to ONE `branch1/**` at the
+/// upstream router; a publish from an upstream client still reaches the downstream subscriber, and
+/// the aggregate is withdrawn on teardown.
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn verify_upstream_subscriber_aggregation() {
+    use std::sync::{Arc, Mutex};
+
+    use zenoh_config::WhatAmI::Client;
+
+    let k: usize = 20;
+
+    let gateway = ztimeout!(Node::new(Router, "1ca00001")
+        .listen("tcp/127.0.0.1:0")
+        .open());
+    let gateway_loc = loc!(gateway).to_string();
+    let gateway_zid = ztimeout!(gateway.info().zid()).to_string();
+    let edge = ztimeout!(Node::new(Router, "eda00001")
+        .listen("tcp/127.0.0.1:0")
+        .connect(&[gateway_loc.as_str()])
+        .insert("aggregation/upstream/subscribers", "[\"branch1/**\"]")
+        .open());
+    let edge_loc = loc!(edge).to_string();
+    let internal = ztimeout!(Node::new(Client, "c1a00001")
+        .connect(&[edge_loc.as_str()])
+        .open());
+
+    let received: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+    let mut subs = Vec::new();
+    for j in 0..k {
+        let received = received.clone();
+        subs.push(ztimeout!(internal
+            .declare_subscriber(format!("branch1/sensor/{j:04}"))
+            .callback(move |s| {
+                received
+                    .lock()
+                    .unwrap()
+                    .push(s.key_expr().as_str().to_string());
+            })));
+    }
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let count = count_admin(
+        &gateway_loc,
+        &gateway_zid,
+        "subscriber",
+        "2ba00001",
+        "branch1",
+    )
+    .await;
+
+    let client = ztimeout!(Node::new(Client, "aba00001")
+        .connect(&[gateway_loc.as_str()])
+        .open());
+    tokio::time::sleep(Duration::from_millis(500)).await;
+    ztimeout!(client.put("branch1/sensor/0005", "hello")).unwrap();
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let delivered = received
+        .lock()
+        .unwrap()
+        .iter()
+        .any(|kk| kk == "branch1/sensor/0005");
+
+    drop(subs);
+    ztimeout!(client.close()).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let count_after = count_admin(
+        &gateway_loc,
+        &gateway_zid,
+        "subscriber",
+        "2ba00002",
+        "branch1",
+    )
+    .await;
+
+    eprintln!(
+        "\n=== upstream subscriber aggregation (K={k}) ===\n\
+         gateway branch1 subscribers : {count} (expect 1)\n\
+         client->branch delivery    : {delivered} (expect true)\n\
+         after teardown           : {count_after} (expect 0)"
+    );
+
+    ztimeout!(internal.close()).unwrap();
+    ztimeout!(edge.close()).unwrap();
+    ztimeout!(gateway.close()).unwrap();
+
+    assert_eq!(
+        count, 1,
+        "gateway must advertise one aggregate branch1 subscriber, got {count}"
+    );
+    assert!(
+        delivered,
+        "client->branch data must reach the internal subscriber through the aggregate"
+    );
+    assert_eq!(
+        count_after, 0,
+        "aggregate must be withdrawn after all children undeclared, got {count_after}"
+    );
+}
+
+/// Queryables: K per-key queryables collapse to ONE at the upstream router; an upstream client `get`
+/// reaches the real queryable; a wildcard `get` fans out to all children; a missing key returns
+/// empty (no hang);
+/// teardown withdraws the aggregate.
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn verify_upstream_queryable_aggregation() {
+    use zenoh::Wait;
+    use zenoh_config::WhatAmI::Client;
+
+    let k: usize = 20;
+
+    let gateway = ztimeout!(Node::new(Router, "1ca00010")
+        .listen("tcp/127.0.0.1:0")
+        .open());
+    let gateway_loc = loc!(gateway).to_string();
+    let gateway_zid = ztimeout!(gateway.info().zid()).to_string();
+    let edge = ztimeout!(Node::new(Router, "eda00010")
+        .listen("tcp/127.0.0.1:0")
+        .connect(&[gateway_loc.as_str()])
+        .insert("aggregation/upstream/queryables", "[\"branch1/**\"]")
+        .open());
+    let edge_loc = loc!(edge).to_string();
+    let internal = ztimeout!(Node::new(Client, "c1a00010")
+        .connect(&[edge_loc.as_str()])
+        .open());
+
+    let mut qabls = Vec::new();
+    for j in 0..k {
+        let my_key = format!("branch1/sensor/{j:04}");
+        qabls.push(ztimeout!(internal
+            .declare_queryable(my_key.clone())
+            .callback(move |q| {
+                Wait::wait(q.reply(my_key.clone(), "pong")).unwrap();
+            })));
+    }
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let count = count_admin(
+        &gateway_loc,
+        &gateway_zid,
+        "queryable",
+        "2ba00010",
+        "branch1",
+    )
+    .await;
+
+    let client = ztimeout!(Node::new(Client, "aba00010")
+        .connect(&[gateway_loc.as_str()])
+        .open());
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let replies = ztimeout!(client.get("branch1/sensor/0005")).unwrap();
+    let mut got_reply = false;
+    while let Ok(reply) = replies.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/sensor/0005" {
+                got_reply = true;
+            }
+        }
+    }
+
+    let wild = ztimeout!(client.get("branch1/**")).unwrap();
+    let mut wild_replies = 0usize;
+    while let Ok(reply) = wild.recv_async().await {
+        if reply.result().is_ok() {
+            wild_replies += 1;
+        }
+    }
+
+    let empty = ztimeout!(client.get("branch1/sensor/9999")).unwrap();
+    let mut empty_replies = 0usize;
+    while let Ok(reply) = empty.recv_async().await {
+        if reply.result().is_ok() {
+            empty_replies += 1;
+        }
+    }
+
+    drop(qabls);
+    ztimeout!(client.close()).unwrap();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    let count_after = count_admin(
+        &gateway_loc,
+        &gateway_zid,
+        "queryable",
+        "2ba00011",
+        "branch1",
+    )
+    .await;
+
+    eprintln!(
+        "\n=== upstream queryable aggregation (K={k}) ===\n\
+         gateway branch1 queryables : {count} (expect 1)\n\
+         concrete get reply      : {got_reply} (expect true)\n\
+         wildcard get replies    : {wild_replies} (expect {k})\n\
+         missing-key get replies : {empty_replies} (expect 0)\n\
+         after teardown          : {count_after} (expect 0)"
+    );
+
+    ztimeout!(internal.close()).unwrap();
+    ztimeout!(edge.close()).unwrap();
+    ztimeout!(gateway.close()).unwrap();
+
+    assert_eq!(
+        count, 1,
+        "gateway must advertise one aggregate branch1 queryable, got {count}"
+    );
+    assert!(
+        got_reply,
+        "client get must reach the real per-key queryable through the aggregate"
+    );
+    assert_eq!(
+        wild_replies, k,
+        "wildcard get must fan out to all {k} children, got {wild_replies}"
+    );
+    assert_eq!(
+        empty_replies, 0,
+        "missing-key get must return empty, got {empty_replies}"
+    );
+    assert_eq!(
+        count_after, 0,
+        "aggregate must be withdrawn after teardown, got {count_after}"
+    );
+}
+
+/// **B-A2 regression — `target=AllComplete` must reach a complete child through the (complete=false)
+/// aggregate.** The forwarding router folds its per-key queryables into a `complete=false`
+/// `branch1/**` aggregate. An upstream client issuing `get(target=AllComplete)` for a key served by a
+/// genuinely-complete downstream queryable must be forwarded through the aggregate (the upstream
+/// router treats the non-complete wildcard covering the query, toward a router, as a transparent
+/// forwarder) and reach the child — previously the upstream router's AllComplete branch dropped the
+/// aggregate and silently returned zero replies. The negative control proves the forwarder does NOT
+/// over-broaden: AllComplete to a NON-complete child still returns empty (the forwarding router
+/// re-applies the filter and excludes it).
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn verify_upstream_queryable_allcomplete() {
+    use zenoh::{query::QueryTarget, Wait};
+    use zenoh_config::WhatAmI::Client;
+
+    let gateway = ztimeout!(Node::new(Router, "1ca00012")
+        .listen("tcp/127.0.0.1:0")
+        .open());
+    let gateway_loc = loc!(gateway).to_string();
+    let edge = ztimeout!(Node::new(Router, "eda00012")
+        .listen("tcp/127.0.0.1:0")
+        .connect(&[gateway_loc.as_str()])
+        .insert("aggregation/upstream/queryables", "[\"branch1/**\"]")
+        .open());
+    let edge_loc = loc!(edge).to_string();
+    let internal = ztimeout!(Node::new(Client, "c1a00012")
+        .connect(&[edge_loc.as_str()])
+        .open());
+
+    // A genuinely-complete child (the case AllComplete must reach) and a non-complete child
+    // (negative control: AllComplete must NOT spuriously reach it via the forwarder).
+    let q_complete = ztimeout!(internal
+        .declare_queryable("branch1/sensor/0005")
+        .complete(true)
+        .callback(|q| {
+            Wait::wait(q.reply("branch1/sensor/0005", "pong")).unwrap();
+        }));
+    let q_incomplete = ztimeout!(internal
+        .declare_queryable("branch1/diag/0001")
+        .complete(false)
+        .callback(|q| {
+            Wait::wait(q.reply("branch1/diag/0001", "pong")).unwrap();
+        }));
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let client = ztimeout!(Node::new(Client, "aba00012")
+        .connect(&[gateway_loc.as_str()])
+        .open());
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    // (1) THE FIX: AllComplete to the complete child must reach it through the aggregate.
+    let ac = ztimeout!(client
+        .get("branch1/sensor/0005")
+        .target(QueryTarget::AllComplete))
+    .unwrap();
+    let mut ac_replies = 0usize;
+    while let Ok(reply) = ac.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/sensor/0005" {
+                ac_replies += 1;
+            }
+        }
+    }
+
+    // (2) NEGATIVE CONTROL: AllComplete to the non-complete child must stay empty.
+    let acn = ztimeout!(client
+        .get("branch1/diag/0001")
+        .target(QueryTarget::AllComplete))
+    .unwrap();
+    let mut acn_replies = 0usize;
+    while let Ok(reply) = acn.recv_async().await {
+        if reply.result().is_ok() {
+            acn_replies += 1;
+        }
+    }
+
+    // (3) CONTROL: default BestMatching still reaches the complete child.
+    let bm = ztimeout!(client.get("branch1/sensor/0005")).unwrap();
+    let mut bm_replies = 0usize;
+    while let Ok(reply) = bm.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/sensor/0005" {
+                bm_replies += 1;
+            }
+        }
+    }
+
+    eprintln!(
+        "\n=== B-A2 AllComplete through aggregate ===\n\
+         AllComplete -> complete child   : {ac_replies} (expect >=1)\n\
+         AllComplete -> incomplete child : {acn_replies} (expect 0)\n\
+         BestMatching -> complete child  : {bm_replies} (expect >=1)"
+    );
+
+    drop(q_complete);
+    drop(q_incomplete);
+    ztimeout!(client.close()).unwrap();
+    ztimeout!(internal.close()).unwrap();
+    ztimeout!(edge.close()).unwrap();
+    ztimeout!(gateway.close()).unwrap();
+
+    assert!(
+        ac_replies >= 1,
+        "B-A2: AllComplete get must reach the complete child through the aggregate, got {ac_replies}"
+    );
+    assert_eq!(
+        acn_replies, 0,
+        "AllComplete must not over-broaden to a non-complete child, got {acn_replies}"
+    );
+    assert!(
+        bm_replies >= 1,
+        "BestMatching get must still reach the child, got {bm_replies}"
+    );
+}
+
+/// **No-shadow correctness.** TWO forwarding routers advertise the SAME prefix `branch1/**` (each
+/// with distinct keys). Because the aggregate is always advertised `complete=false` (it never
+/// asserts `complete=true`), the default `BestMatching` get must NOT be shadowed onto a single
+/// router — a client querying a key served by router B must still reach it even though router A also
+/// advertises the prefix. (Under an unconditional-complete design this would silently return empty.)
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn verify_multi_edge_same_prefix_no_shadow() {
+    use zenoh::Wait;
+    use zenoh_config::WhatAmI::Client;
+
+    let gateway = ztimeout!(Node::new(Router, "1ca00020")
+        .listen("tcp/127.0.0.1:0")
+        .open());
+    let gateway_loc = loc!(gateway).to_string();
+
+    let mk_edge = |zid: &'static str| {
+        let gateway_loc = gateway_loc.clone();
+        async move {
+            ztimeout!(Node::new(Router, zid)
+                .listen("tcp/127.0.0.1:0")
+                .connect(&[gateway_loc.as_str()])
+                .insert("aggregation/upstream/queryables", "[\"branch1/**\"]")
+                .open())
+        }
+    };
+    let edge_a = mk_edge("eda00020").await;
+    let edge_b = mk_edge("eda00021").await;
+    let edge_a_loc = loc!(edge_a).to_string();
+    let edge_b_loc = loc!(edge_b).to_string();
+
+    // branchA serves branch1/a/**, branchB serves branch1/b/** — same configured prefix, disjoint keys.
+    let branch_a = ztimeout!(Node::new(Client, "c1a00020")
+        .connect(&[edge_a_loc.as_str()])
+        .open());
+    let branch_b = ztimeout!(Node::new(Client, "c1a00021")
+        .connect(&[edge_b_loc.as_str()])
+        .open());
+    // Declared COMPLETE — the discriminating case: if the folded `branch1/**` aggregate inherited a
+    // child's complete=true, BestMatching would shadow one router. The aggregate is advertised
+    // non-complete (a presence advertisement), so the query fans out and both routers stay reachable.
+    let _qa = ztimeout!(branch_a
+        .declare_queryable("branch1/a/x")
+        .complete(true)
+        .callback(|q| Wait::wait(q.reply("branch1/a/x", "from-A")).unwrap()));
+    let _qb = ztimeout!(branch_b
+        .declare_queryable("branch1/b/y")
+        .complete(true)
+        .callback(|q| Wait::wait(q.reply("branch1/b/y", "from-B")).unwrap()));
+    tokio::time::sleep(Duration::from_secs(2)).await;
+
+    let client = ztimeout!(Node::new(Client, "aba00020")
+        .connect(&[gateway_loc.as_str()])
+        .open());
+    tokio::time::sleep(Duration::from_millis(500)).await;
+
+    let mut got_a = false;
+    let mut got_b = false;
+    let ra = ztimeout!(client.get("branch1/a/x")).unwrap();
+    while let Ok(reply) = ra.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/a/x" {
+                got_a = true;
+            }
+        }
+    }
+    let rb = ztimeout!(client.get("branch1/b/y")).unwrap();
+    while let Ok(reply) = rb.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/b/y" {
+                got_b = true;
+            }
+        }
+    }
+
+    eprintln!(
+        "\n=== multi-edge same-prefix (no shadow) ===\n\
+         get branch1/a/x reached edge A: {got_a} (expect true)\n\
+         get branch1/b/y reached edge B: {got_b} (expect true)"
+    );
+
+    ztimeout!(client.close()).unwrap();
+    ztimeout!(branch_a.close()).unwrap();
+    ztimeout!(branch_b.close()).unwrap();
+    ztimeout!(edge_a.close()).unwrap();
+    ztimeout!(edge_b.close()).unwrap();
+    ztimeout!(gateway.close()).unwrap();
+
+    assert!(
+        got_a,
+        "key served by edge A must be reachable (no shadowing)"
+    );
+    assert!(
+        got_b,
+        "key served by edge B must be reachable (no shadowing)"
+    );
+}
+
+/// Cross-mesh: the aggregate (sub + qabl) propagates natively across a router mesh; a client on
+/// router A reaches the downstream subscriber/queryable behind router B for both pub and get.
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn verify_upstream_aggregation_across_mesh() {
+    use std::sync::{Arc, Mutex};
+
+    use zenoh::Wait;
+    use zenoh_config::WhatAmI::Client;
+
+    let k: usize = 20;
+
+    let gateway_a = ztimeout!(Node::new(Router, "1ca00030")
+        .listen("tcp/127.0.0.1:0")
+        .open());
+    let gateway_a_loc = loc!(gateway_a).to_string();
+    let gateway_a_zid = ztimeout!(gateway_a.info().zid()).to_string();
+    let gateway_b = ztimeout!(Node::new(Router, "1ca00031")
+        .listen("tcp/127.0.0.1:0")
+        .connect(&[gateway_a_loc.as_str()])
+        .open());
+    let gateway_b_loc = loc!(gateway_b).to_string();
+    let edge = ztimeout!(Node::new(Router, "eda00030")
+        .listen("tcp/127.0.0.1:0")
+        .connect(&[gateway_b_loc.as_str()])
+        .insert("aggregation/upstream/subscribers", "[\"branch1/**\"]")
+        .insert("aggregation/upstream/queryables", "[\"branch1/**\"]")
+        .open());
+    let edge_loc = loc!(edge).to_string();
+    let internal = ztimeout!(Node::new(Client, "c1a00030")
+        .connect(&[edge_loc.as_str()])
+        .open());
+
+    let received: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+    let mut subs = Vec::new();
+    let mut qabls = Vec::new();
+    for j in 0..k {
+        let received = received.clone();
+        subs.push(ztimeout!(internal
+            .declare_subscriber(format!("branch1/sensor/{j:04}"))
+            .callback(move |s| received
+                .lock()
+                .unwrap()
+                .push(s.key_expr().as_str().to_string()))));
+        let my_key = format!("branch1/sensor/{j:04}");
+        qabls.push(ztimeout!(internal
+            .declare_queryable(my_key.clone())
+            .callback(
+                move |q| Wait::wait(q.reply(my_key.clone(), "pong")).unwrap()
+            )));
+    }
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    let subs_at_a = count_admin(
+        &gateway_a_loc,
+        &gateway_a_zid,
+        "subscriber",
+        "2ba00030",
+        "branch1",
+    )
+    .await;
+    let qabls_at_a = count_admin(
+        &gateway_a_loc,
+        &gateway_a_zid,
+        "queryable",
+        "2ba00031",
+        "branch1",
+    )
+    .await;
+
+    let client = ztimeout!(Node::new(Client, "aba00030")
+        .connect(&[gateway_a_loc.as_str()])
+        .open());
+    tokio::time::sleep(Duration::from_millis(800)).await;
+    ztimeout!(client.put("branch1/sensor/0005", "cmd")).unwrap();
+    let replies = ztimeout!(client.get("branch1/sensor/0007")).unwrap();
+    let mut got_reply = false;
+    while let Ok(reply) = replies.recv_async().await {
+        if let Ok(s) = reply.result() {
+            if s.key_expr().as_str() == "branch1/sensor/0007" {
+                got_reply = true;
+            }
+        }
+    }
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let delivered = received
+        .lock()
+        .unwrap()
+        .iter()
+        .any(|kk| kk == "branch1/sensor/0005");
+
+    eprintln!(
+        "\n=== cross-mesh aggregation (K={k}) ===\n\
+         gateway_A subs={subs_at_a} qabls={qabls_at_a} (expect 1,1)\n\
+         client(A)->branch(B) pub={delivered} get={got_reply} (expect true,true)"
+    );
+
+    drop(subs);
+    drop(qabls);
+    ztimeout!(client.close()).unwrap();
+    ztimeout!(internal.close()).unwrap();
+    ztimeout!(edge.close()).unwrap();
+    ztimeout!(gateway_b.close()).unwrap();
+    ztimeout!(gateway_a.close()).unwrap();
+
+    assert_eq!(
+        subs_at_a, 1,
+        "aggregate subscriber must propagate cross-mesh, got {subs_at_a}"
+    );
+    assert_eq!(
+        qabls_at_a, 1,
+        "aggregate queryable must propagate cross-mesh, got {qabls_at_a}"
+    );
+    assert!(delivered, "client(A)->branch(B) pub must be delivered");
+    assert!(got_reply, "client(A)->branch(B) get must reply");
+}
+
+/// Scale bench (ignored): upstream cardinality + RSS, per-key vs aggregated. Tune AGG_N / AGG_K.
+#[cfg(feature = "internal")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+#[ignore = "perf bench; run with --ignored --nocapture"]
+async fn perf_upstream_aggregation_scale() {
+    use zenoh_config::WhatAmI::Client;
+
+    fn read_rss_kb() -> u64 {
+        let Ok(status) = std::fs::read_to_string("/proc/self/status") else {
+            return 0;
+        };
+        for line in status.lines() {
+            if let Some(rest) = line.strip_prefix("VmRSS:") {
+                return rest
+                    .split_whitespace()
+                    .next()
+                    .and_then(|s| s.parse().ok())
+                    .unwrap_or(0);
+            }
+        }
+        0
+    }
+
+    let n: usize = std::env::var("AGG_N")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(100);
+    let k: usize = std::env::var("AGG_K")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(100);
+    eprintln!(
+        "\n=== upstream-agg scale: N={n} branchs, K={k} keys ({} total) ===",
+        n * k
+    );
+
+    for aggregated in [false, true] {
+        let gateway = ztimeout!(Node::new(Router, "1c000060")
+            .listen("tcp/127.0.0.1:0")
+            .open());
+        let gateway_loc = loc!(gateway).to_string();
+        let gateway_zid = ztimeout!(gateway.info().zid()).to_string();
+        let mut eb = Node::new(Router, "ed000061")
+            .listen("tcp/127.0.0.1:0")
+            .connect(&[gateway_loc.as_str()]);
+        if aggregated {
+            let prefixes = (0..n)
+                .map(|i| format!("\"branch{i}/**\""))
+                .collect::<Vec<_>>()
+                .join(",");
+            eb = eb.insert("aggregation/upstream/subscribers", &format!("[{prefixes}]"));
+        }
+        let edge = ztimeout!(eb.open());
+        let edge_loc = loc!(edge).to_string();
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        let rss_before = read_rss_kb();
+
+        let mut branchs = Vec::with_capacity(n);
+        let mut subs = Vec::new();
+        for i in 0..n {
+            let branch = ztimeout!(Node::new(Client, &format!("c1{i:06x}"))
+                .connect(&[edge_loc.as_str()])
+                .open());
+            for j in 0..k {
+                subs.push(ztimeout!(branch
+                    .declare_subscriber(format!("branch{i}/sensor{j:04}"))
+                    .callback(|_| {})));
+            }
+            branchs.push(branch);
+        }
+        tokio::time::sleep(Duration::from_secs(2)).await;
+        let rss_after = read_rss_kb();
+        let card = count_admin(
+            &gateway_loc,
+            &gateway_zid,
+            "subscriber",
+            "2b000062",
+            "branch",
+        )
+        .await;
+        eprintln!(
+            "{}: gateway-card={card} rss-delta={} kB",
+            if aggregated {
+                "AGGREGATED"
+            } else {
+                "PER-KEY   "
+            },
+            rss_after.saturating_sub(rss_before)
+        );
+        drop(subs);
+        for r in branchs {
+            let _ = r.close().await;
+        }
+        ztimeout!(edge.close()).unwrap();
+        ztimeout!(gateway.close()).unwrap();
+        tokio::time::sleep(Duration::from_secs(2)).await;
+    }
+}