diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala
index 7216d5420..63cf62945 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticClientDelegator.scala
@@ -1247,7 +1247,8 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit context: ConversionContext): ElasticResult[ElasticResponse] =
     delegate.multiSearch(
       elasticQueries,
@@ -1255,7 +1256,8 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
       aggregations,
       fields,
       nestedHits,
-      rowInvariants
+      rowInvariants,
+      legProjections
     )
 
   /** Asynchronous search for documents / aggregations matching the SQL query.
@@ -1318,7 +1320,8 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit
     ec: ExecutionContext,
     context: ConversionContext
@@ -1329,7 +1332,8 @@ trait ElasticClientDelegator extends ElasticClientApi with BulkTypes {
       aggregations,
       fields,
       nestedHits,
-      rowInvariants
+      rowInvariants,
+      legProjections
     )
 
   /** Searches and converts results into typed entities from an SQL query.
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
index b6361ade4..de89a9e4b 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/ElasticConversion.scala
@@ -123,7 +123,8 @@ trait ElasticConversion {
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
     explodeNested: Boolean = true,
     retainDocumentId: Boolean = false,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] = {
     var json = results
     if (json.has("responses")) {
@@ -139,7 +140,8 @@ trait ElasticConversion {
         nestedHits,
         explodeNested,
         retainDocumentId,
-        rowInvariants
+        rowInvariants,
+        legProjections
       )
     } else {
       // Single search response
@@ -166,7 +168,8 @@ trait ElasticConversion {
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
     explodeNested: Boolean = true,
     retainDocumentId: Boolean = false,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit context: ConversionContext): Try[Seq[ListMap[String, Any]]] =
     Try {
       val responses = jsonArray.elements().asScala.toList
@@ -181,6 +184,14 @@ trait ElasticConversion {
         s"for ${responses.size} responses)"
       )
 
+      // Same contract, same reason (issue #354): a short list would silently give the unmatched
+      // legs ANOTHER leg's projection, which is the defect the per-leg projections exist to close.
+      require(
+        legProjections.isEmpty || legProjections.size == responses.size,
+        s"legProjections must carry one entry per response (got ${legProjections.size} " +
+        s"for ${responses.size} responses)"
+      )
+
       // Collect all errors
       val errors = responses.zipWithIndex.collect {
         case (response, idx) if response.has("error") =>
@@ -200,15 +211,23 @@ trait ElasticConversion {
         // which is also why the legs being concatenated afterwards costs nothing.
         val allRows = responses.zipWithIndex.flatMap { case (response, leg) =>
           if (!response.has("error")) {
+            // 🔴 Issue #354 — a leg builds its rows from ITS OWN projection, then column `i`
+            // takes the name the FIRST branch gave column `i` (SQL-92 §7.10). Handing every leg
+            // the first branch's names and aliases was the defect: `multiple.fieldAliases` merges
+            // the branches' maps keyed by SOURCE field, so `SELECT id AS x … UNION ALL SELECT id
+            // AS y …` kept ONE of the two and BOTH branches' rows came back as `y`, with the `x`
+            // the analyst asked for null-filled — branch 1's own rows included.
+            val leg0 = legProjections.lift(leg)
             jsonToRows(
               response,
-              fieldAliases,
+              leg0.map(_.fieldAliases).getOrElse(fieldAliases),
               aggregations,
-              fields,
-              nestedHits,
+              leg0.map(_.fields).getOrElse(fields),
+              leg0.map(_.nestedHits).getOrElse(nestedHits),
               explodeNested,
               retainDocumentId,
-              rowInvariants.lift(leg).getOrElse(ListMap.empty)
+              rowInvariants.lift(leg).getOrElse(ListMap.empty),
+              outputFields = leg0.map(_ => fields).getOrElse(Seq.empty)
             )
           } else {
             Seq.empty
@@ -310,7 +329,8 @@ trait ElasticConversion {
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
     explodeNested: Boolean = true,
     retainDocumentId: Boolean = false,
-    rowInvariants: ListMap[String, Any] = ListMap.empty
+    rowInvariants: ListMap[String, Any] = ListMap.empty,
+    outputFields: Seq[String] = Seq.empty
   )(implicit context: ConversionContext): Seq[ListMap[String, Any]] = {
     val hitsNode = Option(json.path("hits").path("hits"))
       .filter(_.isArray)
@@ -360,10 +380,16 @@ trait ElasticConversion {
     }
 
     // Normalize all rows at the end, after all transformations (flattening, aggregation merging)
-    // Filter out "*" from fields — it is an artifact of COUNT(*) and not a real column
-    val effectiveFields = fields.filterNot(_ == "*")
-    if (effectiveFields.isEmpty) rows
-    else rows.map(rowNormalizer(effectiveFields))
+    // Filter out "*" from fields — it is an artifact of COUNT(*) and not a real column.
+    // `outputFields` (issue #354) renames column `i` to what the result calls it; the "*" filter
+    // runs on the PAIRS so a dropped source name takes its target with it and the two lists cannot
+    // drift out of alignment.
+    val pairs =
+      if (outputFields.size == fields.size) fields.zip(outputFields)
+      else fields.map(f => (f, f))
+    val effective = pairs.filterNot(_._1 == "*")
+    if (effective.isEmpty) rows
+    else rows.map(rowProjector(effective.map(_._1), effective.map(_._2)))
   }
 
   def findKeyValue(path: String, map: Map[String, Any]): Option[Any] = {
@@ -463,28 +489,127 @@ trait ElasticConversion {
     */
   protected def rowNormalizer(
     requestedFields: Seq[String]
+  )(implicit context: ConversionContext): ListMap[String, Any] => ListMap[String, Any] =
+    rowProjector(requestedFields, requestedFields)
+
+  /** [[rowNormalizer]] with the OUTPUT names decoupled from the names LOOKED UP in the row — the
+    * mechanism SQL-92 §7.10 positional matching needs (issue #354).
+    *
+    * `sourceFields` is the projection the rows were BUILT from (the branch's own declared SELECT
+    * list); `targetFields` is what column `i` is called in the result (the FIRST branch's names).
+    * Column `i` of the output is therefore whatever column `i` of the branch produced, whatever
+    * either side called it — which is what `SELECT a, b FROM x UNION ALL SELECT b, a FROM y` asks
+    * for, and what a by-name lookup answered with a NULL.
+    *
+    * 🔴 "Position" here means the index into the branch's DECLARED projection, NEVER an index into
+    * whatever order the row map happens to enumerate. The two are the same word and only one is
+    * correct: a re-key derived from `row.keys` put values under the wrong column names on real ES
+    * 8.18, HTTP 200, and is strictly harder to detect than the raggedness it replaced. That is why
+    * the caller passes a name list and this walks the row looking those names up.
+    *
+    * When the two lists are equal — every branch naming its columns the same way, which is the
+    * overwhelmingly common case and every shape any captured BI workload emits — this IS
+    * [[rowNormalizer]], including its "already shaped, return the same instance" fast path, and
+    * every guard below is inert by construction. Renaming disables that fast path: a row that is
+    * already in order under its own names still has to be rebuilt under the result's.
+    *
+    * Two things a row MAP cannot express, decided once per stream rather than per row (see
+    * `outFirst` / `targetNames` in the body): a result column name repeated at two positions keeps
+    * the FIRST, and an unrequested row entry that happens to carry a result column's name is
+    * dropped rather than appended over it.
+    */
+  protected def rowProjector(
+    sourceFields: Seq[String],
+    targetFields: Seq[String]
   )(implicit context: ConversionContext): ListMap[String, Any] => ListMap[String, Any] = {
-    if (requestedFields.isEmpty) identity
+    if (sourceFields.isEmpty) identity
     else {
-      val fieldArr: Array[String] = requestedFields.toArray
+      val fieldArr: Array[String] = sourceFields.toArray
       val len = fieldArr.length
+      // A target list of a different length cannot be matched positionally against this one, so
+      // the projection degrades to a plain normalization rather than guessing an alignment.
+      val outArr: Array[String] =
+        if (targetFields.size == len) targetFields.toArray else fieldArr
+      var renames = false
+      var k = 0
+      while (k < len && !renames) {
+        renames = fieldArr(k) != outArr(k)
+        k += 1
+      }
       val fieldIndex = new java.util.HashMap[String, Integer](len * 2)
       var i = 0
       while (i < len) {
         fieldIndex.putIfAbsent(fieldArr(i), i)
         i += 1
       }
+      // 🔴 TWO hazards that only exist once the two name lists differ, both decided ONCE per
+      // stream so the row loop pays nothing for them.
+      //
+      //   * `outFirst` — a row map cannot hold the SAME column name twice, so when the FIRST
+      //     branch projects one name at two positions (`SELECT amount, amount`) only the first
+      //     position can be emitted. Without this the builder wrote both and the LAST won, so
+      //     column 1 displayed column 2's value; it also diverged across cross-builds, because
+      //     2.13's `ListMap` builder replaces a duplicate key in place while 2.12's removes and
+      //     re-appends it.
+      //   * `targetNames` — an entry the row carries that is NOT one of this branch's columns
+      //     but IS the name of a result column. Before the lists could differ such a key was
+      //     always found by `fieldIndex` and could never become an "extra"; now it can, and
+      //     appending it CLOBBERED the column the projection had just filled. Reachable, and not
+      //     exotically: the parent object of a dotted path (`addr` beside `addr.city -> city`),
+      //     `_id` when the document-id column is on, and the internal aggregation keys a metric
+      //     leaves behind. The declared column wins and the stray entry is dropped — it is not
+      //     that column, it only shares its name.
+      val outFirst: Array[Boolean] = new Array[Boolean](len)
+      if (renames) {
+        val emitted = new java.util.HashSet[String](len * 2)
+        var j = 0
+        while (j < len) {
+          outFirst(j) = emitted.add(outArr(j))
+          j += 1
+        }
+      } else java.util.Arrays.fill(outFirst, true)
+      val targetNames: java.util.HashSet[String] =
+        if (renames) new java.util.HashSet[String](java.util.Arrays.asList(outArr: _*))
+        else null
+      val nullFillMissing = context match {
+        case EntityContext => false
+        case _             => true
+      }
       if (fieldIndex.size() != len) {
-        // Duplicate output names cannot hold distinct positions in a row map — keep the
-        // legacy per-row semantics for this degenerate shape, name set hoisted per stream
-        val requestedSet = requestedFields.toSet
-        row => normalizeRowOrdered(row, requestedFields, requestedSet)
-      } else {
-        val nullFillMissing = context match {
-          case EntityContext => false
-          case _             => true
+        // Duplicate SOURCE names cannot hold distinct positions in a row map.
+        //   * without a rename this is the historical degenerate shape — keep the exact legacy
+        //     per-row semantics, name set hoisted per stream;
+        //   * WITH a rename the duplicates are well defined after all: `SELECT amount, amount`
+        //     feeding a result whose columns are called `a, b` means both read `amount`. The
+        //     index cannot express that (it maps a name to its FIRST position), so this shape
+        //     walks the target list instead — `O(cols x row)` per row, on a shape nothing but a
+        //     hand-written duplicate projection reaches.
+        if (!renames) {
+          val requestedSet = sourceFields.toSet
+          row => normalizeRowOrdered(row, sourceFields, requestedSet)
+        } else {
+          val sourceSet = sourceFields.toSet
+          row => {
+            val builder = ListMap.newBuilder[String, Any]
+            var j = 0
+            while (j < len) {
+              if (outFirst(j)) {
+                row.get(fieldArr(j)) match {
+                  case Some(v) => builder += outArr(j) -> v
+                  case None    => if (nullFillMissing) builder += outArr(j) -> null
+                }
+              }
+              j += 1
+            }
+            row.foreach { entry =>
+              if (!sourceSet.contains(entry._1) && !targetNames.contains(entry._1))
+                builder += entry
+            }
+            builder.result()
+          }
         }
-        row => {
+      } else { row =>
+        {
           val values = new Array[Any](len)
           val seen = new Array[Boolean](len)
           var extras: ListBuffer[(String, Any)] = null
@@ -498,8 +623,14 @@ trait ElasticConversion {
               values(p) = entry._2
               seen(p) = true
               p += 1
-              // All requested fields matched in order: whatever the iterator still holds are
-              // extras already in their final position — the row IS its normalized form
+              // Every source field matched in order. Without a rename the row IS its normalized
+              // form and whatever the iterator still holds are extras already in their final
+              // position; WITH one the row still has to be rebuilt, and those trailing entries are
+              // drained below instead. 🔴 Leaving the loop either way is what keeps this condition
+              // — the only one on the per-ENTRY path — identical to the pre-#354 one: a `p < len`
+              // guard here MEASURED +3.6% on the commonest shape of all, a five-column projection
+              // Elasticsearch returns in SELECT order, which is a tax on every row of every query
+              // for a case only a `UNION ALL` can reach.
               if (p == len) passthrough = true
             } else {
               inOrder = false
@@ -507,20 +638,36 @@ trait ElasticConversion {
               if (idx ne null) {
                 values(idx.intValue) = entry._2
                 seen(idx.intValue) = true
+              } else if (renames && targetNames.contains(entry._1)) {
+                // a stray entry wearing a result column's name — dropped, see `targetNames`
               } else {
                 if (extras eq null) extras = new ListBuffer[(String, Any)]
                 extras += entry
               }
             }
           }
+          // Under a rename the loop may have stopped on `passthrough` with entries still to come;
+          // none of them can be a source name (a row map's keys are unique and all `len` of them
+          // were just consumed), so they are extras — subject to the same `targetNames` rule.
+          if (renames) {
+            while (it.hasNext) {
+              val entry = it.next()
+              if (!targetNames.contains(entry._1)) {
+                if (extras eq null) extras = new ListBuffer[(String, Any)]
+                extras += entry
+              }
+            }
+          }
           // An in-order strict prefix needs no rebuild either when missing fields are skipped
-          if (passthrough || (inOrder && !nullFillMissing)) row
+          if (!renames && (passthrough || (inOrder && !nullFillMissing))) row
           else {
             val builder = ListMap.newBuilder[String, Any]
             var j = 0
             while (j < len) {
-              if (seen(j)) builder += fieldArr(j) -> values(j)
-              else if (nullFillMissing) builder += fieldArr(j) -> null
+              if (outFirst(j)) {
+                if (seen(j)) builder += outArr(j) -> values(j)
+                else if (nullFillMissing) builder += outArr(j) -> null
+              }
               j += 1
             }
             if (extras ne null) extras.foreach(builder += _)
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
index 224f3e6b2..075fa397a 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/SearchApi.scala
@@ -490,36 +490,9 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
         }
       case parsed: SingleSearch =>
         // #276 -- resolve temporal literals against the mapped `date` columns BEFORE rendering
-        val single = resolveWithSchema(parsed) match {
-          case ElasticSuccess(resolved) => resolved
-          case ElasticFailure(error)    => return ElasticResult.failure(error)
-        }
-        val elasticQuery = ElasticQuery(
-          single,
-          collection.immutable.Seq(single.sources: _*),
-          sql = Some(query),
-          explodeNested = single.explodeNested
-        )
-        this match {
-          case scrollApi: ScrollApi if single.returnsRows && requiresScrollPaging(single.limit) =>
-            // A row query is data-bound, not time-bound: every page request below
-            // carries its own timeout, so the stream always terminates.
-            Await.result(
-              scrollRows(scrollApi, single, elasticQuery),
-              Duration.Inf
-            )
-          case _ =>
-            if (single.windowRowQuery)
-              searchWithWindowEnrichment(single)
-            else
-              singleSearch(
-                elasticQuery,
-                single.fieldAliases,
-                single.sqlAggregations,
-                extractOutputFieldNames(single),
-                single.nestedHitsMappings,
-                rowInvariantsOf(single)
-              )
+        resolveWithSchema(parsed) match {
+          case ElasticSuccess(single) => searchResolved(single, query)
+          case ElasticFailure(error)  => ElasticResult.failure(error)
         }
 
       case parsed: MultiSearch =>
@@ -552,12 +525,13 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
           elasticQueries,
           multiple.fieldAliases,
           multiple.sqlAggregations,
-          // ⚠️ Pre-existing simplification, untouched: the output NAMES come from the FIRST leg
-          // only. The row-invariant CONSTANTS below are per-leg, because each leg may declare its
-          // own (#253 FOLD-IN 1).
-          multiple.requests.headOption.map(extractOutputFieldNames).getOrElse(Seq.empty),
+          // The result's column NAMES come from the FIRST leg (SQL-92 §7.10). Everything a leg
+          // needs to build its OWN rows — aliases, projection, nested-hits mapping — travels
+          // per leg below, as do the row-invariant CONSTANTS (#253 FOLD-IN 1).
+          unionAllOutputFieldNames(multiple),
           multiple.requests.headOption.map(_.nestedHitsMappings).getOrElse(Map.empty),
-          multiple.requests.map(rowInvariantsOf)
+          multiple.requests.map(rowInvariantsOf),
+          unionAllLegProjections(multiple)
         )
 
       case _ =>
@@ -755,7 +729,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit context: ConversionContext): ElasticResult[ElasticResponse] = {
     elasticQueries.queries.flatMap { elasticQuery =>
       validateJson("search", elasticQuery.query).map(error =>
@@ -797,7 +772,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
             fields,
             nestedHits,
             elasticQueries.explodeNested,
-            rowInvariants = rowInvariants
+            rowInvariants = rowInvariants,
+            legProjections = legProjections
           )
         ) match {
           case success @ ElasticSuccess(_) =>
@@ -884,29 +860,9 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
 
       case parsed: SingleSearch =>
         // #276 -- resolve temporal literals against the mapped `date` columns BEFORE rendering
-        val single = resolveWithSchema(parsed) match {
-          case ElasticSuccess(resolved) => resolved
-          case ElasticFailure(error)    => return Future.successful(ElasticResult.failure(error))
-        }
-        val elasticQuery = ElasticQuery(
-          single,
-          collection.immutable.Seq(single.sources: _*)
-        )
-        this match {
-          case scrollApi: ScrollApi if single.returnsRows && requiresScrollPaging(single.limit) =>
-            scrollRows(scrollApi, single, elasticQuery)
-          case _ =>
-            if (single.windowRowQuery)
-              Future.successful(searchWithWindowEnrichment(single))
-            else
-              singleSearchAsync(
-                elasticQuery,
-                single.fieldAliases,
-                single.sqlAggregations,
-                extractOutputFieldNames(single),
-                single.nestedHitsMappings,
-                rowInvariantsOf(single)
-              )
+        resolveWithSchema(parsed) match {
+          case ElasticSuccess(single) => searchResolvedAsync(single)
+          case ElasticFailure(error)  => Future.successful(ElasticResult.failure(error))
         }
 
       case parsed: MultiSearch =>
@@ -929,11 +885,11 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
           elasticQueries,
           multiple.fieldAliases,
           multiple.sqlAggregations,
-          // ⚠️ Pre-existing simplification, untouched: the output NAMES come from the FIRST leg
-          // only. The row-invariant CONSTANTS below are per-leg (#253 FOLD-IN 1).
-          multiple.requests.headOption.map(extractOutputFieldNames).getOrElse(Seq.empty),
+          // See `search`'s arm: first leg names the columns, every leg builds its own rows.
+          unionAllOutputFieldNames(multiple),
           multiple.requests.headOption.map(_.nestedHitsMappings).getOrElse(Map.empty),
-          multiple.requests.map(rowInvariantsOf)
+          multiple.requests.map(rowInvariantsOf),
+          unionAllLegProjections(multiple)
         )
 
       case _ =>
@@ -1088,7 +1044,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit
     ec: ExecutionContext,
     context: ConversionContext
@@ -1113,7 +1070,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
               fields,
               nestedHits,
               elasticQueries.explodeNested,
-              rowInvariants = rowInvariants
+              rowInvariants = rowInvariants,
+              legProjections = legProjections
             )
           ) match {
             case success @ ElasticSuccess(_) =>
@@ -2269,7 +2227,13 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
     multiple.requests
       .foldLeft(zero) {
         case (ElasticSuccess(acc), leg) =>
-          search(leg) match {
+          // 🔴 [[searchResolved]], not `search` — the legs arrive ALREADY resolved from the seam
+          // in `search`, and `resolveWithSchema` is not a pure check: a leg carrying a WHERE
+          // subquery has its inner statement EXECUTED against Elasticsearch by
+          // `SubqueryResolver`, uncached. Re-entering `search` here resolved every leg a second
+          // time (issue #355); the licensed cap path was repaired the same way in story 22.6 and
+          // this route still had it.
+          searchResolved(leg, leg.sql) match {
             case ElasticSuccess(r)     => ElasticResult.success(acc :+ r)
             case ElasticFailure(error) => ElasticResult.failure(error)
           }
@@ -2278,6 +2242,74 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
       .map(mergeLegResponses(multiple, sql, _))
   }
 
+  /** Execute a SingleSearch whose schema and temporal literals are ALREADY resolved.
+    *
+    * The body `search`'s `SingleSearch` arm used to inline, lifted so that a caller holding a
+    * resolved statement can execute it without paying for — or re-running — the resolution (issue
+    * #355). `search` reaches it through `resolveWithSchema`; `unionAllByLeg` reaches it with the
+    * legs that seam already resolved.
+    */
+  private def searchResolved(single: SingleSearch, sql: String)(implicit
+    context: ConversionContext
+  ): ElasticResult[ElasticResponse] = {
+    implicit def timestamp: Long = System.currentTimeMillis()
+    val elasticQuery = ElasticQuery(
+      single,
+      collection.immutable.Seq(single.sources: _*),
+      sql = Some(sql),
+      explodeNested = single.explodeNested
+    )
+    this match {
+      case scrollApi: ScrollApi if single.returnsRows && requiresScrollPaging(single.limit) =>
+        // A row query is data-bound, not time-bound: every page request below
+        // carries its own timeout, so the stream always terminates.
+        Await.result(
+          scrollRows(scrollApi, single, elasticQuery),
+          Duration.Inf
+        )
+      case _ =>
+        if (single.windowRowQuery)
+          searchWithWindowEnrichment(single)
+        else
+          singleSearch(
+            elasticQuery,
+            single.fieldAliases,
+            single.sqlAggregations,
+            extractOutputFieldNames(single),
+            single.nestedHitsMappings,
+            rowInvariantsOf(single)
+          )
+    }
+  }
+
+  /** The async twin of [[searchResolved]]. */
+  private def searchResolvedAsync(single: SingleSearch)(implicit
+    ec: ExecutionContext,
+    context: ConversionContext
+  ): Future[ElasticResult[ElasticResponse]] = {
+    implicit def timestamp: Long = System.currentTimeMillis()
+    val elasticQuery = ElasticQuery(
+      single,
+      collection.immutable.Seq(single.sources: _*)
+    )
+    this match {
+      case scrollApi: ScrollApi if single.returnsRows && requiresScrollPaging(single.limit) =>
+        scrollRows(scrollApi, single, elasticQuery)
+      case _ =>
+        if (single.windowRowQuery)
+          Future.successful(searchWithWindowEnrichment(single))
+        else
+          singleSearchAsync(
+            elasticQuery,
+            single.fieldAliases,
+            single.sqlAggregations,
+            extractOutputFieldNames(single),
+            single.nestedHitsMappings,
+            rowInvariantsOf(single)
+          )
+    }
+  }
+
   /** The async twin. Sequential futures — same shape, same memory argument. */
   private def unionAllByLegAsync(multiple: MultiSearch, sql: String)(implicit
     ec: ExecutionContext,
@@ -2289,7 +2321,8 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
       .foldLeft(zero) { (accF, leg) =>
         accF.flatMap {
           case ElasticSuccess(acc) =>
-            searchAsync(leg).map {
+            // Already resolved by the seam — see `unionAllByLeg` (issue #355).
+            searchResolvedAsync(leg).map {
               case ElasticSuccess(r)     => ElasticResult.success(acc :+ r)
               case ElasticFailure(error) => ElasticResult.failure(error)
             }
@@ -2301,54 +2334,110 @@ trait SearchApi extends ElasticConversion with ElasticClientHelpers with SchemaC
 
   /** THE row contract of a `UNION ALL`, in one place, for every route that concatenates legs.
     *
-    * A `UNION ALL` has three execution routes — the one-shot `_msearch`, the per-leg route below,
+    * A `UNION ALL` has three execution routes — the one-shot `_msearch`, the per-leg route above,
     * and `CoreDqlExtension`'s licensed cap fold — and a caller must not be able to tell them apart
-    * from the rows. The contract is the one the ONE-SHOT path already implements, because that is
-    * the route every bounded statement has always taken: [[ElasticConversion.rowNormalizer]] over
-    * the FIRST branch's output names.
+    * from the rows.
+    *
+    * 🔴 SQL-92 §7.10 matches set-operation branches BY ORDINAL POSITION, never by name: the
+    * branches must agree on DEGREE and on the type of the i-th column, and the result takes the
+    * FIRST branch's names. `CORRESPONDING` — the optional clause that asks for name-based matching
+    * — is the proof, because nobody adds an opt-in for the behaviour they already have (almost
+    * nobody implements it; DuckDB went the other way and added a non-standard `UNION BY NAME`).
     *
-    * 🔴 BY NAME, not by position, and the difference is a wrong answer rather than a cosmetic one.
-    * An earlier draft of this story re-keyed positionally "because that is what the one-shot path
-    * does" — it is not: `multiSearch` hands `requests.head`'s names to `parseResponseTree`, which
-    * applies `rowNormalizer`, a name-keyed lookup that null-fills a miss and appends an extra.
-    * MEASURED by the independent review on real ES 8.18, `SELECT category, tag FROM l UNION ALL
-    * SELECT tag, category FROM r` (legal — same arity, same type) came back from the positional
-    * re-key with `category` holding the TAG and `tag` holding the CATEGORY, HTTP 200, while the
-    * one-shot route bound each value to its own name. A `SELECT *` leg was worse: its own names are
-    * unknown to `extractOutputFieldNames`, so `zip` put the id under `category` and DROPPED the
-    * columns past the first branch's width.
+    * Every route used to look the first branch's names up IN THE LEG'S ROW, which is the path of
+    * least resistance from a wire format that is a name → value map — and a wrong answer with HTTP
+    * 200 (issue #354). MEASURED on real ES 8.18:
     *
-    * The returned function hoists every stream-constant decision (the name array, the index, the
-    * context) ONCE per statement; it is applied per row on the un-LIMITed extraction path, where
-    * "per row" can mean millions (`feedback_no_per_row_hot_path_work`).
+    *   - `SELECT a, b FROM x UNION ALL SELECT a, a FROM y` — both parse-time guards admit it (same
+    *     degree, same types) and column 2 of branch 2 came back NULL;
+    *   - `SELECT id AS x FROM l UNION ALL SELECT id AS y FROM r` — `MultiSearch.fieldAliases`
+    *     merges the branches' maps keyed by SOURCE field, so one of `x`/`y` survived and EVERY row,
+    *     branch 1's own included, answered `{x -> null, y -> …}`.
     *
-    * A first branch of `SELECT *` yields NO names — `rowNormalizer` is then `identity` and every
-    * leg keeps what Elasticsearch returned, which is the only honest answer for an opaque
-    * projection.
+    * 🔴 The trap in the fix, and the reason this takes a NAME LIST rather than a row: "position"
+    * means the index into the branch's DECLARED projection, never an index into whatever order the
+    * row map happens to enumerate. A story-22.6 draft derived it from `row.keys` and put values
+    * under the wrong column names — HTTP 200 again, and strictly harder to detect than the
+    * raggedness it replaced, because the column names looked right.
     *
-    * ⚠️ One shape where the routes still differ, measured on real ES 8.18 rather than assumed:
-    * `SELECT id AS x FROM l UNION ALL SELECT id AS y FROM r` — the one-shot route answers `{x ->
-    * null, y -> …}` for EVERY row including branch 1's own, because it never applies a leg's own
-    * alias mapping; the per-leg routes answer `{x -> …}` for branch 1. That is a PRE-EXISTING
-    * defect of the one-shot path, and the better of the two answers is the one the routes here give
-    * — propagating it to make the three agree would be aligning to a bug.
+    * Two shapes are therefore left matched BY NAME on purpose, not by omission:
+    *
+    *   - a FIRST branch of `SELECT *` yields no names, so there is nothing to match against and
+    *     every leg keeps its OWN projection (its own names, its own aliases — never the merged map,
+    *     see [[unionAllLegProjections]]), which is what that leg would answer on its own;
+    *   - an OPAQUE leg (`SELECT *` past the first branch) declares no projection of its own —
+    *     `MultiSearch.declared` is `None` for it and the arity/type guards exempt it — so its rows
+    *     arrive in `_source` order and a positional `zip` would put the id under the first branch's
+    *     first column and DROP everything past its width. It keeps the first branch's names, looked
+    *     up by name, which is the only matching an opaque projection admits.
+    *
+    * The returned functions hoist every stream-constant decision (the name arrays, the index, the
+    * context) ONCE per statement; one is applied per row on the un-LIMITed extraction path, where
+    * "per row" can mean millions (`feedback_no_per_row_hot_path_work`). Where the branches agree on
+    * their column names — the overwhelmingly common case — every mapper is `rowNormalizer` itself,
+    * fast path and all, so a `UNION ALL` of homogeneous branches pays exactly what it paid before.
+    *
+    * @return
+    *   ONE mapper per leg, in leg order.
     */
-  private[client] def unionAllRowNormalizer(
+  private[client] def unionAllRowMappers(
     multiple: MultiSearch
-  )(implicit context: ConversionContext): ListMap[String, Any] => ListMap[String, Any] =
-    rowNormalizer(multiple.requests.headOption.map(extractOutputFieldNames).getOrElse(Seq.empty))
+  )(implicit context: ConversionContext): Seq[ListMap[String, Any] => ListMap[String, Any]] = {
+    val outputFields = unionAllOutputFieldNames(multiple)
+    if (outputFields.isEmpty) multiple.requests.map(_ => identity[ListMap[String, Any]] _)
+    else
+      multiple.requests.map { leg =>
+        rowProjector(unionAllLegFieldNames(leg, outputFields), outputFields)
+      }
+  }
+
+  /** The names the result's columns take: the FIRST branch's, SQL-92 §7.10. */
+  private def unionAllOutputFieldNames(multiple: MultiSearch): Seq[String] =
+    multiple.requests.headOption.map(extractOutputFieldNames).getOrElse(Seq.empty)
+
+  /** The projection a leg's rows are BUILT from — its own, or the result's names when the leg is an
+    * opaque `SELECT *` that declares none (see [[unionAllRowMappers]]).
+    */
+  private def unionAllLegFieldNames(leg: SingleSearch, outputFields: Seq[String]): Seq[String] = {
+    val own = extractOutputFieldNames(leg)
+    if (own.isEmpty) outputFields else own
+  }
+
+  /** Everything each leg of a one-shot `_msearch` needs to build ITS OWN rows (issue #354).
+    *
+    * 🔴 Emitted even when the FIRST branch declares no projection, and that is not cosmetic: with
+    * no projections `parseMultiSearchResponse` falls back to `multiple.fieldAliases`, which merges
+    * the branches' maps keyed by the SOURCE field — so `SELECT * FROM t UNION ALL SELECT id AS x
+    * FROM l UNION ALL SELECT id AS y FROM r` kept ONE of `x`/`y` and renamed BOTH later legs with
+    * it. That is #354 case 2's exact mechanism, surviving behind an opaque first branch. There is
+    * still nothing to RENAME to (the result has no declared names), so each leg simply keeps its
+    * own — which is precisely what `search(leg)` gives the per-leg route, so the two agree.
+    */
+  private def unionAllLegProjections(multiple: MultiSearch): Seq[LegProjection] = {
+    val outputFields = unionAllOutputFieldNames(multiple)
+    multiple.requests.map { leg =>
+      LegProjection(
+        leg.fieldAliases,
+        unionAllLegFieldNames(leg, outputFields),
+        leg.nestedHitsMappings
+      )
+    }
+  }
 
-  /** ONE merge, two callers; the row contract is [[unionAllRowNormalizer]]'s. */
+  /** ONE merge, two callers; the row contract is [[unionAllRowMappers]]'s. */
   private def mergeLegResponses(
     multiple: MultiSearch,
     sql: String,
     responses: Seq[ElasticResponse]
   )(implicit context: ConversionContext): ElasticResponse = {
-    val normalise = unionAllRowNormalizer(multiple)
+    val mappers = unionAllRowMappers(multiple)
     ElasticResponse(
       Some(sql),
       responses.map(_.query).mkString("\n"),
-      responses.flatMap(_.results.map(normalise)),
+      responses.zipWithIndex.flatMap { case (response, leg) =>
+        val normalise = mappers.applyOrElse(leg, (_: Int) => identity[ListMap[String, Any]] _)
+        response.results.map(normalise)
+      },
       multiple.fieldAliases,
       toClientAggregations(multiple.sqlAggregations)
     )
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtension.scala b/core/src/main/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtension.scala
index 990c4bf20..53e6df578 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtension.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtension.scala
@@ -357,6 +357,27 @@ class CoreDqlExtension extends ExtensionSpi {
     * row leg, and an aggregation-shaped leg, which scrolling would page hits a bucket query never
     * returns for — is executed as itself and its rows are taken from the result, bounded by the
     * budget.
+    *
+    * 🔴 Once the budget is SPENT the same arithmetic keeps working, and that is issue #355's fix:
+    * `remaining` is 0, so the leg is asked for `0 + 1` rows and contributes `take(0)` of them — an
+    * EXISTENCE PROBE. A leg that answers a row is one the result was cut before; a leg that answers
+    * nothing was never going to contribute, and the statement is not reported as capped for
+    * dropping it. The previous shape treated "a leg is being dropped" as truncation outright, which
+    * over-reported whenever every remaining leg was EMPTY — `truncated = true`, a non-empty warning
+    * and a cap-hit, byte-identical in outcome to a genuine cut, on a statement nothing had cut.
+    * Probing costs at most ONE bounded request per remaining leg and stops at the first leg that
+    * answers, because from there truncation is an established fact.
+    *
+    * The probe never SCROLLS — see [[existenceProbeIfSpent]], which is gated on the same predicate
+    * `SearchExecutor` routes on, so the shape that would otherwise be paged is the shape that gets
+    * `LIMIT 1`. It is not always FREE: a leg with its own `LIMIT` and a grouped leg are executed as
+    * themselves, because neither has an existence test cheaper than running it — and both are
+    * reached only when the budget ran out precisely at the preceding leg's boundary.
+    *
+    * ⚠️ A dropped leg that FAILS now fails the statement, where before it was never executed and
+    * the statement answered HTTP 200 with `truncated = true`. That is the same rule every other leg
+    * of a capped `UNION ALL` already follows, and the alternative is reporting a cap from a leg
+    * nobody could read — but it is a user-visible change.
     */
   private def cappedUnionAllRows(
     multi: MultiSearch,
@@ -376,50 +397,40 @@ class CoreDqlExtension extends ExtensionSpi {
     // `acc` would reach `max + 1`, so a later leg would still have one row of budget left and
     // would be executed — losing the property that legs past a spent budget never run at all.
     val capBit = new java.util.concurrent.atomic.AtomicBoolean(false)
-    // The row contract, hoisted ONCE for the whole statement — the same function the per-leg and
-    // one-shot routes apply, so all three routes answer with one row shape. Per-row work on this
-    // path is per-row over an UN-LIMITED extraction (`feedback_no_per_row_hot_path_work`).
-    val normalise = client.unionAllRowNormalizer(multi)
+    // The row contract, hoisted ONCE for the whole statement — ONE mapper per leg, the same
+    // functions the per-leg and one-shot routes apply, so all three routes answer with one row
+    // shape. Per-row work on this path is per-row over an UN-LIMITED extraction
+    // (`feedback_no_per_row_hot_path_work`).
+    val mappers = client.unionAllRowMappers(multi)
     val zero: Future[ElasticResult[Seq[ListMap[String, Any]]]] =
       Future.successful(ElasticResult.success(Seq.empty[ListMap[String, Any]]))
-    multi.requests
-      .foldLeft(zero) { (accF, leg) =>
+    multi.requests.zipWithIndex
+      .foldLeft(zero) { case (accF, (leg, legIndex)) =>
         accF.flatMap {
           case failure @ ElasticFailure(_) => Future.successful(failure)
           case ElasticSuccess(acc) =>
+            val normalise =
+              mappers.applyOrElse(legIndex, (_: Int) => identity[ListMap[String, Any]] _)
             val remaining = max.toLong - acc.size.toLong
-            // what the leg is ASKED for; what it may CONTRIBUTE is `remaining`
+            // what the leg is ASKED for; what it may CONTRIBUTE is `remaining`. When the budget
+            // is spent that is `1` and `0` — the existence probe (issue #355).
             val probe = remaining + 1L
             def keep(rows: Seq[ListMap[String, Any]]): Seq[ListMap[String, Any]] = {
               if (rows.size.toLong > remaining) capBit.set(true)
               acc ++ rows.take(remaining.toInt).map(normalise)
             }
-            if (remaining <= 0L) {
-              // 🔴 THE BUDGET IS GLOBAL, SO THE PROBE MUST BE. A per-leg probe row cannot see the
-              // truncation that happens AT A LEG BOUNDARY: when leg `i` returns exactly
-              // `remaining` rows there is no probe row in it, and legs `i+1…n` are then skipped
-              // here without anyone asking whether they had rows — so a genuinely truncated
-              // statement reported `truncated = false`, an empty warning and ZERO cap-hits.
-              // MEASURED on real ES 8.18 over 3-shard 5-document indices: `A UNION ALL B` at quota
-              // 5 returned 5 rows where SQL says 10, silently; for a Community user
-              // (`maxQueryResults = 10000`) that is any `UNION ALL` whose first leg holds 10,000
-              // documents. The flag and the meter agreed with each other AND BOTH LIED, which is
-              // worse than the contradiction it replaced because nothing detects it.
-              //
-              // Reaching this arm means the budget is spent and a leg is being dropped: that IS
-              // the truncation. It over-reports only when every remaining leg happens to be
-              // EMPTY — accepted by the lead as far narrower than the corner it closes, and not
-              // worth a global probe row that would cost an extra leg execution.
-              capBit.set(true)
+            if (remaining <= 0L && capBit.get()) {
+              // Truncation is already an established FACT — a probe could only confirm it, so the
+              // remaining legs are dropped without costing a request.
               Future.successful(ElasticResult.success(acc))
-            } else if (leg.returnsRows && leg.limit.isEmpty)
+            } else if (remaining > 0L && leg.returnsRows && leg.limit.isEmpty)
               client
                 .scroll(leg, client.defaultScrollConfig.copy(maxDocuments = Some(probe)))
                 .map(_._1)
                 .runWith(Sink.seq)
                 .map(rows => ElasticResult.success(keep(rows)))
             else
-              client.dqlExecutor.execute(leg).flatMap {
+              client.dqlExecutor.execute(existenceProbeIfSpent(leg, remaining)).flatMap {
                 case ElasticSuccess(q: QueryStructured) =>
                   Future.successful(ElasticResult.success(keep(q.response.results)))
                 case ElasticSuccess(q: QueryRows) =>
@@ -467,6 +478,41 @@ class CoreDqlExtension extends ExtensionSpi {
       .map(_.map(rows => (rows, capBit.get())))
   }
 
+  /** Issue #355 — a leg the budget cannot pay for, bounded so that "is this statement TRUNCATED?"
+    * is answered by a fact rather than by the assumption that a dropped leg had rows.
+    *
+    * 🔴 The rewrite is gated on the SAME predicate `SearchExecutor` routes on (`limit.isDefined ||
+    * fields.isEmpty`), not on a proxy for it, because its whole job is to keep the probe off the
+    * scroll path — a leg the budget cannot pay for must never be SCROLLED, which is what "never
+    * paged, never materialised beyond a row" means here. So exactly the shape that would otherwise
+    * scroll — no `LIMIT` of its own, fields projected — is given `LIMIT 1` and becomes a one-shot
+    * `"size": 1` request. That covers a plain row leg AND the aggregation-BEARING-but-not-grouped
+    * shape (`SELECT amount, MAX(amount) AS m FROM y`, whose `fields` is `List(amount)` because
+    * `fields` drops the aggregates), which `SearchExecutor` answers as `QueryStream(api.scroll(…))`
+    * — MEASURED: a `returnsRows` gate left it SCROLLED.
+    *
+    * Everything else is executed AS ITSELF, and neither case reaches a scroll:
+    *
+    *   - a leg with its OWN `LIMIT` — rewriting it would DISCARD that bound and make the probe lie.
+    *     `LIMIT 0` contributes nothing by construction, yet a `LIMIT 1` probe finds a row in it and
+    *     reports a truncation nothing truncated — the very over-report this method exists to close,
+    *     needing no empty index at all; an `OFFSET` past the matching documents does the same. The
+    *     leg is bounded by what the analyst wrote, so running it costs what it always would.
+    *   - an un-`LIMIT`ed GROUP BY / windowed leg (`fields.isEmpty`) — `searchAsync` answers it
+    *     one-shot because `returnsRows` is false for a grouped statement. Its rows are BUCKETS,
+    *     `hits.total` says nothing about how many it has, and the only test for "does it produce a
+    *     row" is running it. A `LIMIT` would be wrong as well as useless: on an aggregation the
+    *     limit is the `terms` bucket size, fixed up by `Bucket.update` at parse time and not by a
+    *     `copy` after it.
+    *
+    * `copy` without `update()` is safe for the shape it applies to: the only derived state reading
+    * `limit` is that bucket size (excluded above), the inner-hits `size` (a `def`) and the `sql`
+    * renders (lazy vals on the new instance).
+    */
+  private def existenceProbeIfSpent(leg: SingleSearch, remaining: Long): SingleSearch =
+    if (remaining > 0L || leg.limit.isDefined || leg.fields.isEmpty) leg
+    else leg.copy(limit = Some(Limit(1, None)))
+
   /** Apply the single-index result-boundary rule (ADR D4) at the (licensed) quota.
     *   - explicit LIMIT > finite quota → 402 reject (intentional asymmetry)
     *   - no LIMIT, finite quota, NOT a join leg → cap the scroll + flag truncated
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/metrics/MetricsElasticClient.scala b/core/src/main/scala/app/softnetwork/elastic/client/metrics/MetricsElasticClient.scala
index 9b636a3d2..420e2d76f 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/metrics/MetricsElasticClient.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/metrics/MetricsElasticClient.scala
@@ -26,6 +26,7 @@ import app.softnetwork.elastic.client.{
   ElasticQueries,
   ElasticQuery,
   ElasticResponse,
+  LegProjection,
   SingleValueAggregateResult
 }
 import app.softnetwork.elastic.client.bulk._
@@ -945,7 +946,8 @@ class MetricsElasticClient(
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit context: ConversionContext): ElasticResult[ElasticResponse] = {
     measureResult("multisearch") {
       delegate.multiSearch(
@@ -954,7 +956,8 @@ class MetricsElasticClient(
         aggregations,
         fields,
         nestedHits,
-        rowInvariants
+        rowInvariants,
+        legProjections
       )
     }
   }
@@ -1014,7 +1017,8 @@ class MetricsElasticClient(
     aggregations: ListMap[String, SQLAggregation],
     fields: Seq[String] = Seq.empty,
     nestedHits: Map[String, Seq[(String, String)]] = Map.empty,
-    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty
+    rowInvariants: Seq[ListMap[String, Any]] = Seq.empty,
+    legProjections: Seq[LegProjection] = Seq.empty
   )(implicit
     ec: ExecutionContext,
     context: ConversionContext
@@ -1028,7 +1032,8 @@ class MetricsElasticClient(
           aggregations,
           fields,
           nestedHits,
-          rowInvariants
+          rowInvariants,
+          legProjections
         )
         .asInstanceOf[Future[ElasticResult[ElasticResponse]]]
     }
diff --git a/core/src/main/scala/app/softnetwork/elastic/client/package.scala b/core/src/main/scala/app/softnetwork/elastic/client/package.scala
index da4b1c0a5..52cae02d4 100644
--- a/core/src/main/scala/app/softnetwork/elastic/client/package.scala
+++ b/core/src/main/scala/app/softnetwork/elastic/client/package.scala
@@ -290,6 +290,32 @@ package object client extends SerializationApi {
         |""".stripMargin
   }
 
+  /** What ONE branch of a `UNION ALL` needs to build ITS OWN rows, before those rows take the
+    * result's column names (issue #354).
+    *
+    * `_msearch` answers with one response per branch, and every one of them used to be parsed with
+    * the FIRST branch's aliases, projection and nested-hits mapping — so a branch that named its
+    * columns differently lost its values to a by-name lookup, and a branch with its own nested
+    * mapping was flattened by another branch's. The per-leg route never had that problem: it
+    * executes each branch through `search(leg)`, which uses that branch's own everything by
+    * construction. This carries the same three inputs to the one-shot route, so the two routes
+    * agree because they do the same thing, not because one was taught to imitate the other.
+    *
+    * @param fieldAliases
+    *   the BRANCH's `source field -> alias` map
+    * @param fields
+    *   the BRANCH's declared output names, in SELECT order — or, for an opaque `SELECT *` branch
+    *   that declares none, the FIRST branch's names, which keeps the only matching an opaque
+    *   projection admits (by name)
+    * @param nestedHits
+    *   the BRANCH's `JOIN UNNEST` mappings
+    */
+  case class LegProjection(
+    fieldAliases: ListMap[String, String],
+    fields: Seq[String],
+    nestedHits: Map[String, Seq[(String, String)]]
+  )
+
   /** Retry configuration
     */
   case class RetryConfig(
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala
index 485614d7a..373ea6ffb 100644
--- a/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala
+++ b/core/src/test/scala/app/softnetwork/elastic/client/RowNormalizerSpec.scala
@@ -150,4 +150,115 @@ class RowNormalizerSpec extends AnyFlatSpec with Matchers with ElasticConversion
       legacy(row, duplicated)(EntityContext).toList
     }
   }
+
+  // ── rowProjector: the same walk with the OUTPUT names decoupled (issue #354) ───────────────
+
+  private def project(
+    row: ListMap[String, Any],
+    source: Seq[String],
+    target: Seq[String]
+  )(implicit ctx: ConversionContext): ListMap[String, Any] =
+    rowProjector(source, target)(ctx)(row)
+
+  "rowProjector" should "be rowNormalizer exactly when the two name lists are equal" in {
+    val row = ListMap[String, Any]("a" -> 1, "b" -> 2, "c" -> 3, "_id" -> "42")
+    // including the fast path: an already-shaped row comes back as the SAME instance
+    project(row, fields, fields)(NativeContext) should be theSameInstanceAs row
+    project(ListMap[String, Any]("c" -> 3), fields, fields)(NativeContext).toList shouldBe
+    normalize(ListMap[String, Any]("c" -> 3))(NativeContext).toList
+  }
+
+  it should "rename column i to target i, whatever either side calls it" in {
+    // the reordered-branch shape: the row is keyed by the BRANCH's names, in the BRANCH's order
+    val row = ListMap[String, Any]("tag" -> "t", "category" -> "c")
+    project(row, Seq("tag", "category"), Seq("category", "tag"))(NativeContext).toList shouldBe
+    List("category" -> "t", "tag" -> "c")
+  }
+
+  it should "rebuild even when the row is already in source order" in {
+    // 🔴 the fast path MUST be off under a rename: the row IS in order under its own names, and
+    // returning it unchanged would answer with the branch's names instead of the result's.
+    val row = ListMap[String, Any]("y" -> 9)
+    project(row, Seq("y"), Seq("x"))(NativeContext).toList shouldBe List("x" -> 9)
+  }
+
+  it should "null-fill a target whose source the row does not carry, and append extras" in {
+    val row = ListMap[String, Any]("q" -> 1, "extra" -> true)
+    project(row, Seq("p", "q"), Seq("a", "b"))(NativeContext).toList shouldBe
+    List("a" -> null, "b" -> 1, "extra" -> true)
+    // …and EntityContext skips the missing one rather than null-filling it, as it always has
+    project(row, Seq("p", "q"), Seq("a", "b"))(EntityContext).toList shouldBe
+    List("b" -> 1, "extra" -> true)
+  }
+
+  /** 🔴 A hazard that exists ONLY once the two lists differ: before, a row key equal to a requested
+    * name was always found by the name index and could never become an "extra". Now it can, and
+    * appending it CLOBBERED the column the projection had just filled — the declared column read
+    * back as a raw nested object, or as Elasticsearch's `_id` instead of the branch's own.
+    *
+    * Reachable shapes, all measured on this engine: the PARENT of a dotted path (`parseSimpleHits`
+    * re-adds `addr.city` as `city` while `addr` survives from `_source`), `_id` when the
+    * document-id column is on, and the internal aggregation key a metric leaves behind.
+    */
+  it should "not let a stray row entry clobber a result column that shares its name" in {
+    val row = ListMap[String, Any]("profileId" -> "P", "profiles" -> ListMap("city" -> "Paris"))
+    project(row, Seq("profileId", "profiles.city"), Seq("profileId", "profiles"))(
+      NativeContext
+    ).toList shouldBe List("profileId" -> "P", "profiles" -> null)
+
+    val withCity = ListMap[String, Any](
+      "profileId" -> "P",
+      "profiles"  -> ListMap("city" -> "Paris"),
+      "city"      -> "Paris"
+    )
+    project(withCity, Seq("profileId", "city"), Seq("profileId", "profiles"))(
+      NativeContext
+    ).toList shouldBe List("profileId" -> "P", "profiles" -> "Paris")
+
+    // the `_id` shape: the branch's own `id` is what column 1 holds, not the document id
+    val withDocId = ListMap[String, Any]("id" -> "biz-7", "_id" -> "esdoc-123")
+    project(withDocId, Seq("id"), Seq("_id"))(NativeContext).toList shouldBe
+    List("_id" -> "biz-7")
+  }
+
+  /** …and the same guard in the duplicate-source arm, which builds by walking the TARGET list. */
+  it should "not let a stray row entry clobber a result column under a duplicate projection" in {
+    val row = ListMap[String, Any]("a" -> 7, "e" -> 1)
+    project(row, Seq("a", "a"), Seq("x", "e"))(NativeContext).toList shouldBe
+    List("x" -> 7, "e" -> 7)
+  }
+
+  /** 🔴 A result column name repeated at two positions: a row MAP cannot hold it twice, so the
+    * FIRST position is the one that survives. Emitting both let the SECOND win, so column 1
+    * displayed column 2's value — and it diverged across cross-builds, because 2.13's `ListMap`
+    * builder replaces a duplicate key in place while 2.12's removes and re-appends it.
+    */
+  it should "keep the FIRST position when the target names repeat" in {
+    val row = ListMap[String, Any]("amount" -> "A", "m" -> "M")
+    project(row, Seq("amount", "m"), Seq("amount", "amount"))(NativeContext).toList shouldBe
+    List("amount" -> "A")
+    // …including when the first position has no value to supply
+    project(ListMap[String, Any]("m" -> "M"), Seq("amount", "m"), Seq("amount", "amount"))(
+      NativeContext
+    ).toList shouldBe List("amount" -> null)
+  }
+
+  it should "feed every target that names the same source under a duplicate projection" in {
+    // `SELECT a, a` renamed onto `(x, y)`: a row map holds ONE `a`, and both columns read it
+    val row = ListMap[String, Any]("a" -> 7, "extra" -> true)
+    project(row, Seq("a", "a"), Seq("x", "y"))(NativeContext).toList shouldBe
+    List("x" -> 7, "y" -> 7, "extra" -> true)
+  }
+
+  it should "degrade to a plain normalization when the target list has a different length" in {
+    // no positional alignment exists, so nothing is guessed
+    val row = ListMap[String, Any]("b" -> 2)
+    project(row, fields, Seq("x"))(NativeContext).toList shouldBe
+    normalize(row)(NativeContext).toList
+  }
+
+  it should "be identity when there is no source projection to match" in {
+    val row = ListMap[String, Any]("whatever" -> 1)
+    project(row, Seq.empty, Seq("a"))(NativeContext) should be theSameInstanceAs row
+  }
 }
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/UnionAllRoutingSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/UnionAllRoutingSpec.scala
index a602a8024..e1a006ba0 100644
--- a/core/src/test/scala/app/softnetwork/elastic/client/UnionAllRoutingSpec.scala
+++ b/core/src/test/scala/app/softnetwork/elastic/client/UnionAllRoutingSpec.scala
@@ -109,12 +109,47 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     )(implicit ec: ExecutionContext): scala.concurrent.Future[ElasticResult[Option[JsonNode]]] =
       scala.concurrent.Future.successful(emptyResponse)
 
+    /** The `_msearch` answer: ONE response per leg, each carrying a single hit whose `_source` is
+      * keyed by THAT leg's SOURCE field names and valued `"<source field>@<leg>"`.
+      *
+      * 🔴 `_source`, not output names, and that is what makes the alias mapping load-bearing: the
+      * rename from `id` to `x` is `jsonNodeToMap`'s, driven by the leg's `fieldAliases`. A stub
+      * that answered `{"x": …}` directly would be green whichever alias map the parse used, which
+      * is exactly the defect issue #354 case 2 reports. The leg's SQL is recovered from the body
+      * the stub itself rendered above, so this needs no out-of-band channel.
+      */
+    private def legHitsResponse(elasticQueries: ElasticQueries): ElasticResult[Option[JsonNode]] = {
+      val mapper = new com.fasterxml.jackson.databind.ObjectMapper()
+      val root = mapper.createObjectNode()
+      val responses = root.putArray("responses")
+      elasticQueries.queries.zipWithIndex.foreach { case (q, i) =>
+        val leg = i + 1
+        val sql = mapper.readTree(q.query).path("sql").asText()
+        val sourceNames: Seq[String] =
+          app.softnetwork.elastic.sql.parser.Parser(sql) match {
+            case Right(sel: SingleSearch) =>
+              val fields = sel.select.fieldsWithComputedAliases
+              if (fields.size == 1 && fields.head.identifier.identifierName == "*")
+                Seq("id", "category", "tag", "amount")
+              else fields.map(_.identifier.identifierName)
+            case _ => Seq(s"leg$leg")
+          }
+        val response = responses.addObject()
+        val hits = response.putObject("hits")
+        hits.putObject("total").put("value", sourceNames.size)
+        val hitsArray = hits.putArray("hits")
+        val source = hitsArray.addObject().putObject("_source")
+        sourceNames.foreach(n => source.put(n, s"$n@$leg"))
+      }
+      ElasticResult.success(Some(root))
+    }
+
     override private[client] def executeMultiSearch(
       elasticQueries: ElasticQueries
     ): ElasticResult[Option[JsonNode]] = {
       msearchCalls.incrementAndGet()
       lastMultiQuery = Some(elasticQueries)
-      emptyResponse
+      legHitsResponse(elasticQueries)
     }
 
     // `searchAsync` reaches the ASYNC seam, never the sync one — a counter on `executeMultiSearch`
@@ -125,7 +160,7 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
       scala.concurrent.Future.successful {
         msearchCalls.incrementAndGet()
         lastMultiQuery = Some(elasticQueries)
-        emptyResponse
+        legHitsResponse(elasticQueries)
       }
 
     override def scroll(statement: SearchStatement, config: ScrollConfig)(implicit
@@ -158,6 +193,21 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     }
   }
 
+  /** Counts how many times the per-statement `SingleSearch` seam is entered (issue #355). The
+    * `MultiSearch` seam resolves every leg once; anything beyond that is the per-leg fold resolving
+    * what it was already handed.
+    */
+  private class ResolveCountingClient extends RecordingClient {
+    val singleResolutions = new AtomicInteger(0)
+
+    override private[client] def resolveWithSchema(
+      single: SingleSearch
+    ): ElasticResult[SingleSearch] = {
+      singleResolutions.incrementAndGet()
+      super.resolveWithSchema(single)
+    }
+  }
+
   /** Every routing row asserts the statement actually EXECUTED. Without this a parse rejection or a
     * seam refusal would leave both counters at zero and satisfy half the assertions below for
     * entirely the wrong reason.
@@ -248,17 +298,24 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
   }
 
   /** 🔴 THE ROW CONTRACT. The routes must agree, or the presence of a `LIMIT` decides what a caller
-    * gets — and the first attempt at making them agree was itself a wrong answer.
+    * gets — and they must agree on what SQL says, which is where issue #354 comes in.
     *
-    * The contract is [[ElasticConversion.rowNormalizer]] over the FIRST branch's output names: a
-    * BY-NAME lookup that null-fills a miss and appends an extra. It is the function the one-shot
-    * `_msearch` path already applies, and the lead's ruling for this story.
+    * SQL-92 §7.10 matches set-operation branches BY ORDINAL POSITION: same degree, i-th column
+    * type-compatible with i-th column, and the result takes the FIRST branch's names. Names play no
+    * part in the matching. `CORRESPONDING` — the optional clause that asks for name-based matching
+    * — is the proof, because nobody adds an opt-in for the behaviour they already have.
     *
-    * An earlier draft re-keyed POSITIONALLY. The two rows below are the shapes that exposed it,
-    * both measured on real ES 8.18 by the independent review — and note that neither is exotic: one
-    * is the same two columns written in a different order, the other is `SELECT *`.
+    * Story 22.6 shipped one BY-NAME contract for all three routes, which made them agree with each
+    * other but not with SQL; #354 is the remaining half, and the rows below are its three measured
+    * shapes. Every value NAMES ITS OWN COLUMN (`tag@2`), so the oracle is independent of the
+    * mapping under test: a value that lands under the wrong name says so.
+    *
+    * 🔴 The trap the fix had to avoid, recorded because a 22.6 draft fell into it: "position" means
+    * the index into the branch's DECLARED projection, never an index into whatever order the row
+    * map enumerates. Deriving it from `row.keys` put values under the wrong column names on real ES
+    * 8.18 — HTTP 200, and harder to detect than the raggedness it replaced.
     */
-  "The per-leg route" should "bind every value to its own column when a branch REORDERS the projection" in {
+  "The per-leg route" should "match a REORDERED branch POSITIONALLY, as SQL-92 does" in {
     val client = new RecordingClient
     val response = client.search(
       SelectStatement("SELECT category, tag FROM l UNION ALL SELECT tag, category FROM r")
@@ -270,10 +327,11 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     response.results should have size 2
     // Every row keyed by the first branch's names, in its order …
     response.results.map(_.keys.toSeq) shouldBe Seq(Seq("category", "tag"), Seq("category", "tag"))
-    // 🔴 … and each VALUE under the column it names. A positional re-key reads
-    // `category -> tag@2` here: same keys, same order, opposite binding, HTTP 200.
-    response.results.map(_("category")) shouldBe Seq("category@1", "category@2")
-    response.results.map(_("tag")) shouldBe Seq("tag@1", "tag@2")
+    // … and column i holds column i OF ITS OWN BRANCH. Branch 2 declared `tag` first, so `tag@2`
+    // is what the result's FIRST column holds — reordering the projection is how an analyst aligns
+    // two differently-named schemas, and a by-name lookup silently undid it.
+    response.results.map(_("category")) shouldBe Seq("category@1", "tag@2")
+    response.results.map(_("tag")) shouldBe Seq("tag@1", "category@2")
   }
 
   it should "not mis-key or DROP columns when a branch is an opaque SELECT *" in {
@@ -288,9 +346,10 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     response.results should have size 2
     val starRow = response.results(1)
     // 🔴 `MultiSearch.declared` is `None` for `SELECT *` — arity and type checks exempt it on
-    // purpose — so the leg's own names are unknown to `extractOutputFieldNames` and its row
-    // arrives in `_source` order. A positional re-key put the ID under `category` and TRUNCATED
-    // the row to the first branch's width, losing two columns. HTTP 200 both ways.
+    // purpose — so the leg DECLARES no projection and there is no position to match. Its row
+    // arrives in `_source` order; a positional `zip` over that order put the ID under `category`
+    // and TRUNCATED the row to the first branch's width, losing two columns. HTTP 200 both ways.
+    // An opaque branch therefore keeps the by-name match, which is the only one it admits.
     starRow("category") shouldBe "category@2"
     starRow("tag") shouldBe "tag@2"
     starRow("id") shouldBe "id@2"
@@ -299,7 +358,11 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     starRow.keys.toSeq.take(2) shouldBe Seq("category", "tag")
   }
 
-  it should "null-fill a column a branch does not declare" in {
+  /** Issue #354, the shape the analyst sees most: an alias per branch. The first branch names the
+    * result, so branch 2's single column becomes `x` — where a by-name lookup answered `{x -> null,
+    * y -> …}` and, on the one-shot route, did it to branch 1's OWN rows too.
+    */
+  it should "give a branch that aliases its column differently the FIRST branch's name" in {
     val client = new RecordingClient
     val response = client.search(
       SelectStatement("SELECT a AS x FROM t UNION ALL SELECT b AS y FROM u")
@@ -311,21 +374,125 @@ class UnionAllRoutingSpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
     client.scrollCalls.get() shouldBe 2
     response.results should have size 2
     response.results.head shouldBe ListMap[String, Any]("x" -> "x@1")
-    // 🔴 Recorded rather than asserted-away, and MEASURED on real ES 8.18 rather than assumed:
-    // where the branches AGREE on a column name every route gives the same answer (the two rows
-    // above), and where they disagree — an alias per branch — this route null-fills the missing
-    // name and keeps the branch's own column as an extra:
-    //
-    //   SELECT id AS x FROM l UNION ALL SELECT id AS y FROM r
-    //     one-shot : {x -> null, y -> L_id_1} … {x -> null, y -> R_id_2}
-    //     per-leg  : {x -> L_id_1}            … {x -> null, y -> R_id_2}
-    //
-    // The routes therefore still differ on BRANCH 1 of this one shape, because the one-shot
-    // `_msearch` route never applies a leg's own alias mapping and loses branch 1's value under
-    // its own declared name. That is PRE-EXISTING and outside this story (the review recorded it
-    // as an observation); the per-leg answer is the better of the two, and propagating the
-    // one-shot defect to make the two agree would be aligning to a bug.
-    response.results(1) shouldBe ListMap[String, Any]("x" -> null, "y" -> "y@2")
+    // 🔴 not `{x -> null, y -> "y@2"}`: the column the analyst asked for holds branch 2's value,
+    // and the branch's own alias does not leak into the result as a second column.
+    response.results(1) shouldBe ListMap[String, Any]("x" -> "y@2")
+  }
+
+  /** Issue #354 case 1 — the shape BOTH parse-time guards admit (same degree, same types) and that
+    * still came back with a NULL: a branch may project the same column twice, and positionally that
+    * is perfectly well defined. A row map cannot hold two `a` keys, so the leg's row carries one —
+    * and both result columns read it.
+    */
+  it should "feed BOTH result columns from a branch that projects one column twice" in {
+    val client = new RecordingClient
+    val response = client.search(
+      SelectStatement("SELECT a, b FROM x UNION ALL SELECT a, a FROM y")
+    ) match {
+      case ElasticSuccess(r)     => r
+      case ElasticFailure(error) => fail(s"refused: ${error.message}")
+    }
+    client.scrollCalls.get() shouldBe 2
+    response.results should have size 2
+    response.results.head shouldBe ListMap[String, Any]("a" -> "a@1", "b" -> "b@1")
+    // column 2 was NULL before #354 — branch 2 never names a column `b`, and nothing looked at
+    // what it DID name column 2.
+    response.results(1) shouldBe ListMap[String, Any]("a" -> "a@2", "b" -> "a@2")
+  }
+
+  /** 🔴 THE OTHER ROUTE, on the shape that exposed it. A bounded `UNION ALL` is ONE `_msearch`, and
+    * every leg of its response used to be parsed with the FIRST branch's alias map — but
+    * `MultiSearch.fieldAliases` merges the branches' maps keyed by SOURCE field, so `id AS x` and
+    * `id AS y` collapse to ONE entry and the survivor renames BOTH legs' `id`. MEASURED on real ES
+    * 8.18: every row, branch 1's own included, came back `{x -> null, y -> …}`.
+    *
+    * The fixture is the mechanism: the stub answers each leg with a `_source` keyed by that leg's
+    * SOURCE field, valued `"<source>@<leg>"`, which is exactly what Elasticsearch returns and what
+    * makes the alias mapping load-bearing. A stub that answered output names would have been green
+    * against the defect.
+    */
+  "The one-shot _msearch route" should "apply each leg's OWN alias map, then the first branch's names" in {
+    val client = new RecordingClient
+    val response = client.search(
+      SelectStatement("SELECT id AS x FROM l LIMIT 5 UNION ALL SELECT id AS y FROM r LIMIT 5")
+    ) match {
+      case ElasticSuccess(r)     => r
+      case ElasticFailure(error) => fail(s"refused: ${error.message}")
+    }
+    client.msearchCalls.get() shouldBe 1
+    client.scrollCalls.get() shouldBe 0
+    response.results shouldBe Seq(
+      ListMap[String, Any]("x" -> "id@1"),
+      ListMap[String, Any]("x" -> "id@2")
+    )
+  }
+
+  it should "match a REORDERED branch positionally too — the routes agree" in {
+    val client = new RecordingClient
+    val response = client.search(
+      SelectStatement(
+        "SELECT category, tag FROM l LIMIT 5 UNION ALL SELECT tag, category FROM r LIMIT 5"
+      )
+    ) match {
+      case ElasticSuccess(r)     => r
+      case ElasticFailure(error) => fail(s"refused: ${error.message}")
+    }
+    client.msearchCalls.get() shouldBe 1
+    response.results shouldBe Seq(
+      ListMap[String, Any]("category" -> "category@1", "tag" -> "tag@1"),
+      ListMap[String, Any]("category" -> "tag@2", "tag"      -> "category@2")
+    )
+  }
+
+  /** 🔴 #354 case 2 SURVIVING BEHIND AN OPAQUE FIRST BRANCH. With no names to match against there
+    * is nothing to RENAME to — but that is not a licence to hand every leg the MERGED alias map,
+    * which is keyed by the SOURCE field and therefore keeps ONE of `x`/`y` and renames both later
+    * legs with it. Each leg keeps its OWN projection, which is exactly what it would answer alone,
+    * so the one-shot route and the per-leg route agree here too.
+    */
+  it should "still give each leg its OWN alias map when the FIRST branch is SELECT *" in {
+    val client = new RecordingClient
+    val response = client.search(
+      SelectStatement(
+        "SELECT * FROM t LIMIT 5 UNION ALL SELECT id AS x FROM l LIMIT 5 " +
+        "UNION ALL SELECT id AS y FROM r LIMIT 5"
+      )
+    ) match {
+      case ElasticSuccess(r)     => r
+      case ElasticFailure(error) => fail(s"refused: ${error.message}")
+    }
+    client.msearchCalls.get() shouldBe 1
+    response.results should have size 3
+    // the opaque branch keeps what Elasticsearch returned …
+    response.results.head.keys.toSeq shouldBe Seq("id", "category", "tag", "amount")
+    // … and each aliasing branch keeps ITS OWN column, not the other's. Before the fix both read
+    // `{y -> …}` (or both `{x -> …}`, whichever alias the merge happened to keep).
+    response.results(1) shouldBe ListMap[String, Any]("x" -> "id@2")
+    response.results(2) shouldBe ListMap[String, Any]("y" -> "id@3")
+  }
+
+  /** Issue #355 — the per-leg route re-entered `search(leg)`, which resolves the leg a SECOND time.
+    * `resolveWithSchema` is not a pure check: a leg carrying a WHERE subquery has its inner
+    * statement EXECUTED by `SubqueryResolver`, uncached.
+    *
+    * 🔴 Counting inner searches would be VACUOUS here: phase one rewrites the subquery into
+    * literals, so the second pass finds nothing left to execute and the count stays at 1 either
+    * way. The assertion is therefore on the mechanism — how many times the per-statement seam was
+    * entered at all (measured: reinstating `search(leg)` makes it 5).
+    */
+  "A UNION ALL executed per leg" should "resolve each leg ONCE — the seam's resolution is reused" in {
+    val client = new ResolveCountingClient
+    client.search(
+      SelectStatement("SELECT a FROM x WHERE a IN (SELECT b FROM y) UNION ALL SELECT a FROM z")
+    ) match {
+      case ElasticSuccess(_)     => ()
+      case ElasticFailure(error) => fail(s"refused: ${error.message}")
+    }
+    client.scrollCalls.get() shouldBe 2
+    // TWO from the `MultiSearch` seam (one per leg) plus ONE for the inner statement
+    // `SubqueryResolver` executes — and NONE from the fold. Re-entering `search(leg)` makes it 5
+    // (measured), which is the whole of issue #355's second half.
+    client.singleResolutions.get() shouldBe 3
   }
 
   // ── the seam guard runs FIRST ──────────────────────────────────────────────────────────────
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtensionSpec.scala b/core/src/test/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtensionSpec.scala
index a6e694db7..e9d7ddb42 100644
--- a/core/src/test/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtensionSpec.scala
+++ b/core/src/test/scala/app/softnetwork/elastic/client/extensions/CoreDqlExtensionSpec.scala
@@ -204,6 +204,66 @@ class CoreDqlExtensionSpec extends AnyFlatSpec with Matchers {
     * one of the three the executor can answer for a `SingleSearch`, so it exercises the arm without
     * pretending the executor does something it does not.
     */
+  /** Legs with DIFFERENT row counts, so that "the budget was spent at a leg boundary" can be told
+    * apart from "…and a leg that was dropped actually had rows" (issue #355).
+    *
+    * `BudgetClient` serves the same `rowsPerLeg` to every leg, so with it an EMPTY remaining leg is
+    * unrepresentable and the over-report is invisible. Legs are consumed in order, and a leg past
+    * the end of the list serves nothing — `scroll` and `searchAsync` share the counter because the
+    * fold executes legs sequentially and takes exactly one of the two routes per leg.
+    */
+  private class PerLegRowsClient(rowsPerLeg: Seq[Int]) extends NopeClientApi {
+    override protected def logger: Logger = testLogger
+
+    @volatile var scrolls: Seq[(SearchStatement, ScrollConfig)] = Seq.empty
+    @volatile var executed: Seq[SearchStatement] = Seq.empty
+    private val leg = new java.util.concurrent.atomic.AtomicInteger(0)
+
+    private def nextLegRows(): Int = {
+      val i = leg.getAndIncrement()
+      rowsPerLeg.applyOrElse(i, (_: Int) => 0)
+    }
+
+    override def scroll(
+      statement: SearchStatement,
+      config: ScrollConfig = ScrollConfig()
+    )(implicit
+      system: ActorSystem,
+      context: ConversionContext
+    ): Source[(ListMap[String, Any], ScrollMetrics), NotUsed] = {
+      synchronized { scrolls = scrolls :+ ((statement, config)) }
+      val available = nextLegRows()
+      // as the real one does: honours `config.maxDocuments`, IGNORES the statement's own LIMIT
+      val served = config.maxDocuments.map(_.toInt.min(available)).getOrElse(available)
+      Source((1 to served).map(i => (ListMap[String, Any]("a" -> i), ScrollMetrics())).toList)
+    }
+
+    override def searchAsync(
+      statement: SearchStatement
+    )(implicit
+      ec: scala.concurrent.ExecutionContext,
+      context: ConversionContext
+    ): scala.concurrent.Future[ElasticResult[ElasticResponse]] = {
+      synchronized { executed = executed :+ statement }
+      val available = nextLegRows()
+      val bound = statement match {
+        case sel: SingleSearch => sel.limit.map(_.limit).getOrElse(available)
+        case _                 => available
+      }
+      scala.concurrent.Future.successful(
+        ElasticSuccess(
+          ElasticResponse(
+            sql = None,
+            query = "{}",
+            results = (1 to bound.min(available)).map(i => ListMap[String, Any]("a" -> i)),
+            fieldAliases = ListMap.empty,
+            aggregations = ListMap.empty
+          )
+        )
+      )
+    }
+  }
+
   private class UnexpectedVariantClient extends NopeClientApi {
     override protected def logger: Logger = testLogger
 
@@ -891,6 +951,184 @@ class CoreDqlExtensionSpec extends AnyFlatSpec with Matchers {
     }
   }
 
+  /** 🔴 ISSUE #355 — truncation is a FACT about rows, not about legs.
+    *
+    * `cappedUnionAllRows` detects truncation in two places: inside a leg (each leg is asked for one
+    * row more than it may contribute) and AT a leg boundary. The boundary bit used to be a GUESS —
+    * dropping a leg was treated as truncation without asking whether that leg had any rows — so a
+    * statement whose remaining legs were all EMPTY reported `truncated = true`, a non-empty warning
+    * advising a `LIMIT` the analyst had often already written, and a cap-hit. Byte-identical in
+    * outcome to a genuine cut, so no consumer could tell them apart.
+    *
+    * The trigger is not exotic: a first branch whose `LIMIT` equals the quota, or legs summing to
+    * exactly the quota, followed by a branch that matches nothing.
+    *
+    * Both halves are asserted, because a fix that simply stopped reporting at a boundary would
+    * reinstate the UNDER-report this replaced — a genuinely truncated statement answering
+    * `truncated = false` with zero cap-hits, the flag and the meter agreeing and both wrong.
+    */
+  it should "not report truncation when every leg the budget dropped is EMPTY (#355)" in {
+    val matrix = Seq(
+      // rows per leg, quota, expected rows, expected truncated, clue
+      (Seq(10, 0), 10, 10, false, "budget spent at the boundary, remaining leg empty"),
+      (Seq(5, 5, 0), 10, 10, false, "three legs, the dropped one empty"),
+      (Seq(5, 0), 5, 5, false, "two legs, the dropped one empty"),
+      (Seq(5, 0, 0), 5, 5, false, "every dropped leg empty"),
+      // …and the control: a dropped leg that DOES have rows is a real cut and must still say so
+      (Seq(5, 0, 5), 5, 5, true, "an empty leg BEFORE a non-empty one is still a cut"),
+      (Seq(10, 3), 10, 10, true, "budget spent at the boundary, remaining leg non-empty"),
+      (Seq(0, 5), 5, 5, false, "a leading EMPTY leg costs no budget")
+    )
+    matrix.foreach { case (rows, quota, expectedRows, expectedTruncated, clue) =>
+      val collector = new TelemetryCollector
+      val client = new PerLegRowsClient(rows)
+      val ext = new CoreDqlExtension()
+      ext.initialize(
+        ConfigFactory.empty(),
+        strategy(
+          managerWithQuota(
+            Quota.Community.copy(maxQueryResults = Some(quota)),
+            LicenseType.Community
+          ),
+          collector
+        )
+      )
+      val sql = rows.indices.map(i => s"SELECT a FROM t${i + 1}").mkString(" UNION ALL ")
+      val parsed = Parser(sql) match {
+        case Right(st) => st
+        case Left(e)   => fail(s"parse failed: ${e.msg}")
+      }
+      val result = Await.result(ext.execute(parsed, client), 10.seconds) match {
+        case ElasticSuccess(q: QueryRows) => q
+        case other                        => fail(s"[$sql @ $quota] expected QueryRows, got $other")
+      }
+      withClue(s"[$clue: legs $rows, quota $quota] ") {
+        result.rows should have size expectedRows.toLong
+        result.truncation.map(_.truncated) shouldBe Some(expectedTruncated)
+        // the warning and the meter follow the SAME fact — they may not disagree with the flag
+        result.truncation.map(_.warning.nonEmpty) shouldBe Some(expectedTruncated)
+        capHits(collector)("max_query_results") shouldBe (if (expectedTruncated) 1L else 0L)
+      }
+    }
+  }
+
+  /** 🔴 THE MECHANISM behind the row above, which a flag alone cannot see: what the probe COSTS.
+    *
+    * The property "a leg the budget cannot pay for is never SCROLLED" is preserved exactly — the
+    * probe is a ONE-SHOT request bounded to a single row, which `SearchExecutor` routes through
+    * `searchAsync`. And it stops at the first leg that answers, because from there truncation is an
+    * established fact: a three-leg statement whose second leg has rows never touches the third.
+    */
+  it should "probe a dropped leg with a ONE-SHOT single row, and stop at the first that answers" in {
+    val client = new PerLegRowsClient(Seq(5, 3, 7))
+    val res = runWith(
+      "SELECT a FROM x UNION ALL SELECT a FROM y UNION ALL SELECT a FROM z",
+      Quota.Community.copy(maxQueryResults = Some(5)),
+      client
+    )
+    res shouldBe a[ElasticSuccess[_]]
+    // leg 1 alone spends the whole budget …
+    client.scrolls.map(_._1.sql) shouldBe Seq("SELECT a FROM x")
+    // … leg 2 is PROBED, bounded to one row and never scrolled …
+    client.executed.map(_.sql) shouldBe Seq("SELECT a FROM y LIMIT 1")
+    // … and leg 3 is not touched at all: leg 2 already established the truncation.
+    val rows = res.asInstanceOf[ElasticSuccess[QueryResult]].value match {
+      case q: QueryRows => q
+      case other        => fail(s"expected QueryRows, got $other")
+    }
+    rows.rows should have size 5L
+    rows.truncation.map(_.truncated) shouldBe Some(true)
+  }
+
+  /** …and when the dropped legs are empty the probe walks them ALL before concluding — the cost the
+    * exactness buys, stated rather than assumed. Still one bounded request per leg, no scroll.
+    */
+  it should "probe every dropped leg when each answers nothing" in {
+    val client = new PerLegRowsClient(Seq(5, 0, 0))
+    runWith(
+      "SELECT a FROM x UNION ALL SELECT a FROM y UNION ALL SELECT a FROM z",
+      Quota.Community.copy(maxQueryResults = Some(5)),
+      client
+    )
+    client.scrolls.map(_._1.sql) shouldBe Seq("SELECT a FROM x")
+    client.executed.map(_.sql) shouldBe Seq("SELECT a FROM y LIMIT 1", "SELECT a FROM z LIMIT 1")
+  }
+
+  /** 🔴 The probe must respect the DROPPED leg's own bound, or it reinstates the over-report it
+    * exists to close. `LIMIT 0` contributes nothing by construction, and a `LIMIT 1` rewrite would
+    * find a row in it and report a truncation nothing had truncated — deterministic, needing no
+    * empty index at all.
+    */
+  it should "honour a dropped leg's own LIMIT when probing it (#355)" in {
+    val collector = new TelemetryCollector
+    val client = new PerLegRowsClient(Seq(10, 8))
+    val ext = new CoreDqlExtension()
+    ext.initialize(
+      ConfigFactory.empty(),
+      strategy(
+        managerWithQuota(
+          Quota.Community.copy(maxQueryResults = Some(10)),
+          LicenseType.Community
+        ),
+        collector
+      )
+    )
+    val parsed = Parser("SELECT a FROM x UNION ALL SELECT a FROM y LIMIT 0") match {
+      case Right(st) => st
+      case Left(e)   => fail(s"parse failed: ${e.msg}")
+    }
+    val result = Await.result(ext.execute(parsed, client), 10.seconds) match {
+      case ElasticSuccess(q: QueryRows) => q
+      case other                        => fail(s"expected QueryRows, got $other")
+    }
+    result.rows should have size 10L
+    result.truncation.map(_.truncated) shouldBe Some(false)
+    capHits(collector)("max_query_results") shouldBe 0L
+    // the mechanism: the leg was executed AS ITSELF, its `LIMIT 0` intact
+    client.executed.map(_.sql) shouldBe Seq("SELECT a FROM y LIMIT 0")
+  }
+
+  /** 🔴 "A leg the budget cannot pay for is never SCROLLED", on the shape a `returnsRows` test
+    * cannot see. `SELECT amount, MAX(amount) AS m FROM y` is NOT row-shaped, carries no LIMIT and
+    * projects fields, so `SearchExecutor` answers it as `QueryStream(api.scroll(single))` — the
+    * probe has to bound it or it opens a scroll on a leg the budget already refused. The gate is
+    * `SearchExecutor`'s own predicate for exactly that reason.
+    */
+  it should "bound an aggregation-BEARING dropped leg too, never scrolling it" in {
+    val client = new PerLegRowsClient(Seq(5, 3))
+    val res = runWith(
+      "SELECT amount, category FROM x UNION ALL SELECT amount, MAX(amount) AS m FROM y",
+      Quota.Community.copy(maxQueryResults = Some(5)),
+      client
+    )
+    res shouldBe a[ElasticSuccess[_]]
+    // leg 1 spent the budget and is the ONLY scroll; leg 2 was probed one-shot, bounded to a row
+    client.scrolls.map(_._1.sql) shouldBe Seq("SELECT amount, category FROM x")
+    client.executed.map(_.sql) shouldBe Seq("SELECT amount, MAX(amount) AS m FROM y LIMIT 1")
+  }
+
+  /** …and a GROUPED dropped leg is executed as itself: its rows are BUCKETS, a `LIMIT` on it would
+    * mean the `terms` size (fixed at parse time, not by a `copy`), and `searchAsync` answers it
+    * one-shot because `returnsRows` is false — so it still never reaches a scroll.
+    */
+  it should "execute a GROUPED dropped leg unchanged, and still not scroll it" in {
+    val client = new PerLegRowsClient(Seq(5, 0))
+    val res = runWith(
+      "SELECT a, b FROM x UNION ALL SELECT category, COUNT(*) AS n FROM y GROUP BY category",
+      Quota.Community.copy(maxQueryResults = Some(5)),
+      client
+    )
+    val rows = res.asInstanceOf[ElasticSuccess[QueryResult]].value match {
+      case q: QueryRows => q
+      case other        => fail(s"expected QueryRows, got $other")
+    }
+    client.scrolls.map(_._1.sql) shouldBe Seq("SELECT a, b FROM x")
+    client.executed.map(_.sql) shouldBe
+    Seq("SELECT category, COUNT(*) AS n FROM y GROUP BY category")
+    // it produced no buckets, so nothing was cut
+    rows.truncation.map(_.truncated) shouldBe Some(false)
+  }
+
   /** 🔴 The loud backstop, exercised. Until this row existed, reinstating `case _ => success(acc)`
     * before the catch-all left the entire core suite green — the mutation the prior review used,
     * still surviving after the round that claimed to have re-falsified it.
@@ -945,8 +1183,9 @@ class CoreDqlExtensionSpec extends AnyFlatSpec with Matchers {
     *
     * The comparison below is against the route the same client takes WITHOUT the extension, on the
     * same statement and the same rows — an oracle that is a real execution rather than a
-    * transcribed expectation. The one-shot route shares the mechanism by construction (all three
-    * call `SearchApi.unionAllRowNormalizer`) and is pinned in `UnionAllRoutingSpec`.
+    * transcribed expectation. The one-shot route shares the mechanism by construction (all three go
+    * through `SearchApi.unionAllRowMappers`, one mapper per leg) and is pinned in
+    * `UnionAllRoutingSpec`.
     */
   it should "produce the same row shape as the un-capped route for the same statement" in {
     implicit val ctx: ConversionContext = NativeContext
@@ -972,10 +1211,13 @@ class CoreDqlExtensionSpec extends AnyFlatSpec with Matchers {
     }
 
     capped shouldBe plain
-    // …and not vacuously: both routes really produced the re-keyed heterogeneous rows
+    // …and not vacuously: both routes really produced the re-keyed heterogeneous rows, and both
+    // matched branch 2 POSITIONALLY (issue #354) — branch 2 declared `tag` first, so `tag@2` is
+    // what the result's first column holds.
     capped should have size 2
     capped.map(_.keys.toSeq).distinct shouldBe Seq(Seq("category", "tag"))
-    capped.map(_("category")) shouldBe Seq("category@1", "category@2")
+    capped.map(_("category")) shouldBe Seq("category@1", "tag@2")
+    capped.map(_("tag")) shouldBe Seq("tag@1", "category@2")
   }
 
   /** 🔴 The seam resolves the statement ONCE, and the fold executes what it resolved.
@@ -1157,12 +1399,13 @@ class CoreDqlExtensionSpec extends AnyFlatSpec with Matchers {
     // both legs contributed — the stream leg was consumed, not refused (HTTP 500 before the fix)
     // and not dropped (HTTP 200 with half the rows)
     rows should have size 2
-    // …and the row contract reached the STREAM leg's rows too: its `amount` lands under `amount`,
-    // the first branch's `category` it does not declare is null-filled, and its own `m` follows as
-    // an extra. A fold that skipped `normalise` for this arm reddens here.
+    // …and the row contract reached the STREAM leg's rows too: matched POSITIONALLY against the
+    // first branch (issue #354), so its column 1 (`amount`) lands under `amount` and its column 2
+    // (`m`) under `category` — the name the first branch gave column 2. A fold that skipped
+    // `normalise` for this arm reddens here, and so does a by-name lookup (which answered
+    // `category -> null` with the branch's own `m` trailing as an extra).
     rows.head shouldBe ListMap[String, Any]("amount" -> "amount@1", "category" -> "category@1")
-    rows(1) shouldBe
-    ListMap[String, Any]("amount" -> "amount@2", "category" -> null, "m" -> "m@2")
+    rows(1) shouldBe ListMap[String, Any]("amount" -> "amount@2", "category" -> "m@2")
   }
 
   /** …and the fold handles that variant end to end, with the branch LIMIT honoured. */
diff --git a/core/src/test/scala/app/softnetwork/elastic/client/perf/RowCostProbe.scala b/core/src/test/scala/app/softnetwork/elastic/client/perf/RowCostProbe.scala
new file mode 100644
index 000000000..0b41ad02c
--- /dev/null
+++ b/core/src/test/scala/app/softnetwork/elastic/client/perf/RowCostProbe.scala
@@ -0,0 +1,325 @@
+/*
+ * Copyright 2025 SOFTNETWORK
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package app.softnetwork.elastic.client.perf
+
+import app.softnetwork.elastic.client.{
+  ConversionContext,
+  ElasticConversion,
+  NativeContext,
+  NopeClientApi
+}
+import app.softnetwork.elastic.sql.parser.Parser
+import app.softnetwork.elastic.sql.query.SingleSearch
+
+import scala.collection.immutable.ListMap
+import scala.collection.mutable.ListBuffer
+
+/** Row-shaping cost probe for issue #354's `rowNormalizer` -> `rowProjector` generalisation.
+  *
+  * Row shaping is THE per-row hot path: every row of every query passes through it, and an
+  * un-LIMITed extraction can mean millions of them, so a regression here is a regression in
+  * end-to-end query time. The parse path is untouched by #354/#355 (no grammar change), which is
+  * why this probe measures rows rather than statements.
+  *
+  * ==The control is in this file, and the two run interleaved==
+  * `baseline` is the PRE-#354 `rowNormalizer`, transcribed verbatim from `git show
+  * <merge-base>:core/.../ElasticConversion.scala`. `current` is the shipped one. They are measured
+  * A/B/A/B in ONE JVM on the SAME rows, because a difference measured against a remembered number
+  * from another process is noise, not attribution.
+  *
+  * Run it with:
+  *
+  * {{{
+  * sbt "core/Test/runMain app.softnetwork.elastic.client.perf.RowCostProbe"
+  * sbt "core/Test/runMain app.softnetwork.elastic.client.perf.RowCostProbe 20 200000"
+  * }}}
+  *
+  * Arguments, optional and positional: `blocks` (interleaved A/B pairs, default [[DefaultBlocks]])
+  * and `rowsPerBlock` (default [[DefaultRowsPerBlock]]).
+  *
+  * ==This probe NEVER asserts==
+  * It prints and exits, like `ParseCostProbe`. A timing assertion in the suite is a CI-flake
+  * liability (issues #269/#270). Nothing here is collected by `sbt test`: this is a `main`-bearing
+  * object, not a ScalaTest suite.
+  */
+object RowCostProbe extends ElasticConversion {
+
+  val DefaultBlocks: Int = 31
+  val DefaultRowsPerBlock: Int = 200000
+
+  private implicit val ctx: ConversionContext = NativeContext
+
+  // ── the control: the pre-#354 implementation, verbatim ────────────────────────────────────
+  //
+  // Its duplicate-requested-name arm delegated to a trait-private helper; it is replaced here by
+  // the equivalent public `normalizeRow`, and no measured shape reaches that arm.
+  private def baseline(
+    requestedFields: Seq[String]
+  ): ListMap[String, Any] => ListMap[String, Any] = {
+    if (requestedFields.isEmpty) identity
+    else {
+      val fieldArr: Array[String] = requestedFields.toArray
+      val len = fieldArr.length
+      val fieldIndex = new java.util.HashMap[String, Integer](len * 2)
+      var i = 0
+      while (i < len) {
+        fieldIndex.putIfAbsent(fieldArr(i), i)
+        i += 1
+      }
+      if (fieldIndex.size() != len) { row => normalizeRow(row, requestedFields) }
+      else {
+        val nullFillMissing = true
+        row => {
+          val values = new Array[Any](len)
+          val seen = new Array[Boolean](len)
+          var extras: ListBuffer[(String, Any)] = null
+          var inOrder = true
+          var passthrough = false
+          var p = 0
+          val it = row.iterator
+          while (!passthrough && it.hasNext) {
+            val entry = it.next()
+            if (inOrder && fieldArr(p) == entry._1) {
+              values(p) = entry._2
+              seen(p) = true
+              p += 1
+              if (p == len) passthrough = true
+            } else {
+              inOrder = false
+              val idx = fieldIndex.get(entry._1)
+              if (idx ne null) {
+                values(idx.intValue) = entry._2
+                seen(idx.intValue) = true
+              } else {
+                if (extras eq null) extras = new ListBuffer[(String, Any)]
+                extras += entry
+              }
+            }
+          }
+          if (passthrough || (inOrder && !nullFillMissing)) row
+          else {
+            val builder = ListMap.newBuilder[String, Any]
+            var j = 0
+            while (j < len) {
+              if (seen(j)) builder += fieldArr(j) -> values(j)
+              else if (nullFillMissing) builder += fieldArr(j) -> null
+              j += 1
+            }
+            if (extras ne null) extras.foreach(builder += _)
+            builder.result()
+          }
+        }
+      }
+    }
+  }
+
+  // ── row shapes, each one a real thing Elasticsearch hands back ────────────────────────────
+  private def cols(n: Int): Seq[String] = (1 to n).map(i => s"col$i")
+
+  private def row(names: Seq[String]): ListMap[String, Any] =
+    ListMap(names.map(k => k -> (s"$k-value": Any)): _*)
+
+  private case class Shape(
+    label: String,
+    fields: Seq[String],
+    rows: Array[ListMap[String, Any]]
+  )
+
+  private def shapes: Seq[Shape] = {
+    val f5 = cols(5)
+    val f20 = cols(20)
+    Seq(
+      // the dominant case by far: `SELECT a, b, c, d, e` over a flat index — `_source` order
+      // already IS the SELECT order, so the row is returned as the same instance
+      Shape("5 cols, already shaped (fast path)", f5, Array(row(f5))),
+      Shape("20 cols, already shaped (fast path)", f20, Array(row(f20))),
+      // a projection Elasticsearch answers out of order — the full rebuild
+      Shape("5 cols, reordered (rebuild)", f5, Array(row(f5.reverse))),
+      // the document-id column / an inner-hits leftover trailing the projection
+      Shape(
+        "5 cols + 2 extras (fast path)",
+        f5,
+        Array(row(f5) ++ ListMap[String, Any]("_id" -> "42", "_score" -> 1.0))
+      ),
+      // a column the document does not carry: null-filled, so a rebuild
+      Shape("5 cols, one missing (rebuild)", f5, Array(row(f5.drop(1))))
+    )
+  }
+
+  private def median(xs: Array[Long]): Long = {
+    val sorted = xs.sorted
+    sorted(sorted.length / 2)
+  }
+
+  /** The MIN of many interleaved blocks, not the mean: a microbenchmark's noise is one-sided
+    * (scheduling, GC, another core) and the fastest observed block is the one least polluted by it.
+    * Both sides get the same treatment, and the median is printed beside it so a disagreement
+    * between the two is visible rather than hidden.
+    */
+  private def best(xs: Array[Long]): Long = xs.min
+
+  private def timeRows(
+    f: ListMap[String, Any] => ListMap[String, Any],
+    rows: Array[ListMap[String, Any]],
+    iterations: Int
+  ): Long = {
+    var sink = 0
+    val n = rows.length
+    val start = System.nanoTime()
+    var i = 0
+    while (i < iterations) {
+      sink += f(rows(i % n)).size
+      i += 1
+    }
+    val elapsed = System.nanoTime() - start
+    if (sink == Int.MinValue) println("")
+    elapsed
+  }
+
+  def main(args: Array[String]): Unit = {
+    val blocks = args.lift(0).map(_.toInt).getOrElse(DefaultBlocks)
+    val rows = args.lift(1).map(_.toInt).getOrElse(DefaultRowsPerBlock)
+
+    println(s"RowCostProbe — $blocks interleaved A/B blocks x $rows rows each")
+    println(
+      s"JVM ${System.getProperty("java.version")} · Scala ${util.Properties.versionNumberString}"
+    )
+
+    // global warm-up: both implementations, every shape, before any timing
+    shapes.foreach { s =>
+      timeRows(baseline(s.fields), s.rows, 50000)
+      timeRows(rowNormalizer(s.fields), s.rows, 50000)
+    }
+
+    println()
+    println(
+      f"${"shape"}%-38s ${"base"}%9s ${"current"}%9s ${"delta"}%8s   (ns/row, best of blocks)"
+    )
+    shapes.foreach { s =>
+      val b = Array.ofDim[Long](blocks)
+      val c = Array.ofDim[Long](blocks)
+      var k = 0
+      while (k < blocks) {
+        // INTERLEAVED, same rows, same JVM, alternating order so drift cannot favour one side
+        if (k % 2 == 0) {
+          b(k) = timeRows(baseline(s.fields), s.rows, rows)
+          c(k) = timeRows(rowNormalizer(s.fields), s.rows, rows)
+        } else {
+          c(k) = timeRows(rowNormalizer(s.fields), s.rows, rows)
+          b(k) = timeRows(baseline(s.fields), s.rows, rows)
+        }
+        k += 1
+      }
+      val bm = best(b).toDouble / rows
+      val cm = best(c).toDouble / rows
+      val bmed = median(b).toDouble / rows
+      val cmed = median(c).toDouble / rows
+      println(
+        f"${s.label}%-38s $bm%9.2f $cm%9.2f ${(cm - bm) / bm * 100}%+7.1f%%" +
+        f"   (median $bmed%6.2f / $cmed%6.2f)"
+      )
+    }
+
+    // ── what the FEATURE itself costs: a renaming leg has no pre-#354 equivalent ─────────────
+    println()
+    println("UNION ALL positional rename (new work — no baseline equivalent):")
+    println(f"${"shape"}%-38s ${"no rename"}%9s ${"renaming"}%9s ${"delta"}%8s   (ns/row)")
+    val f5 = cols(5)
+    val renamed = Seq("a", "b", "c", "d", "e")
+    Seq(
+      ("5 cols, already shaped", Array(row(f5))),
+      ("5 cols, reordered", Array(row(f5.reverse)))
+    ).foreach { case (label, rs) =>
+      val n = Array.ofDim[Long](blocks)
+      val r = Array.ofDim[Long](blocks)
+      var k = 0
+      while (k < blocks) {
+        if (k % 2 == 0) {
+          n(k) = timeRows(rowProjector(f5, f5), rs, rows)
+          r(k) = timeRows(rowProjector(f5, renamed), rs, rows)
+        } else {
+          r(k) = timeRows(rowProjector(f5, renamed), rs, rows)
+          n(k) = timeRows(rowProjector(f5, f5), rs, rows)
+        }
+        k += 1
+      }
+      val nm = best(n).toDouble / rows
+      val rm = best(r).toDouble / rows
+      println(f"$label%-38s $nm%9.2f $rm%9.2f ${(rm - nm) / nm * 100}%+7.1f%%")
+    }
+
+    // ── issue #355's OTHER half: the resolution the per-leg route no longer repeats ─────────
+    //
+    // `unionAllByLeg` used to re-enter `search(leg)`, which re-ran `resolveWithSchema` on a leg the
+    // `MultiSearch` seam had already resolved — so an N-leg `UNION ALL` paid 2N resolutions where
+    // N is enough. This is what ONE of them costs on a client whose schema lookup is a miss (the
+    // shape every `NopeClientApi`-style client and every un-mapped index takes), so the saving per
+    // dropped call is at least this.
+    val resolver = new NopeClientApi {
+      override protected def logger: org.slf4j.Logger =
+        org.slf4j.LoggerFactory.getLogger("RowCostProbe")
+    }
+    val leg = Parser("SELECT category, amount FROM union_left WHERE amount > 10") match {
+      case Right(single: SingleSearch) => single
+      case other => throw new IllegalStateException(s"probe statement: $other")
+    }
+    var w = 0
+    while (w < 20000) { resolver.resolveWithSchema(leg); w += 1 }
+    val resolveBlocks = Array.ofDim[Long](blocks)
+    var rb = 0
+    while (rb < blocks) {
+      val n = 50000
+      val t = System.nanoTime()
+      var i = 0
+      while (i < n) { resolver.resolveWithSchema(leg); i += 1 }
+      resolveBlocks(rb) = (System.nanoTime() - t) / n
+      rb += 1
+    }
+    println()
+    println(
+      f"per-leg resolveWithSchema, DROPPED once per leg by #355: ${best(resolveBlocks)}%d ns" +
+      f" (median ${median(resolveBlocks)}%d ns)"
+    )
+
+    // ── the per-STREAM half: what building one normalizer costs ──────────────────────────────
+    println()
+    println("per-stream construction (paid ONCE per statement/leg, not per row):")
+    Seq(5, 20).foreach { width =>
+      val f = cols(width)
+      val n = 200000
+      val b = Array.ofDim[Long](blocks)
+      val c = Array.ofDim[Long](blocks)
+      var k = 0
+      while (k < blocks) {
+        var t = System.nanoTime()
+        var i = 0
+        while (i < n) { baseline(f); i += 1 }
+        b(k) = System.nanoTime() - t
+        t = System.nanoTime()
+        i = 0
+        while (i < n) { rowNormalizer(f); i += 1 }
+        c(k) = System.nanoTime() - t
+        k += 1
+      }
+      val bm = best(b).toDouble / n
+      val cm = best(c).toDouble / n
+      println(
+        f"  $width%2d columns: baseline $bm%7.1f ns   current $cm%7.1f ns   ${(cm - bm)}%+7.1f ns"
+      )
+    }
+  }
+}
diff --git a/testkit/src/main/scala/app/softnetwork/elastic/client/UnionAllCompletenessSpec.scala b/testkit/src/main/scala/app/softnetwork/elastic/client/UnionAllCompletenessSpec.scala
index 2d0c18d92..a6b845790 100644
--- a/testkit/src/main/scala/app/softnetwork/elastic/client/UnionAllCompletenessSpec.scala
+++ b/testkit/src/main/scala/app/softnetwork/elastic/client/UnionAllCompletenessSpec.scala
@@ -29,6 +29,7 @@ import org.scalatest.flatspec.AnyFlatSpecLike
 import org.scalatest.matchers.should.Matchers
 import org.slf4j.{Logger, LoggerFactory}
 
+import scala.collection.immutable.ListMap
 import scala.language.implicitConversions
 
 /** Story 22.6 AD-6 — issue #209's family, one venue over.
@@ -115,10 +116,12 @@ trait UnionAllCompletenessSpec extends AnyFlatSpecLike with ElasticDockerTestKit
   /** `client.search(SelectStatement(...))` and NOT `searchAs`: the macro types a `UNION ALL` from
     * the FIRST leg, which is a separate concern from row completeness.
     */
-  private def rowCountOf(sql: String): Int = {
+  private def rowCountOf(sql: String): Int = rowsOf(sql).size
+
+  private def rowsOf(sql: String): Seq[ListMap[String, Any]] = {
     implicit val ctx: ConversionContext = NativeContext
     client.search(SelectStatement(sql)) match {
-      case ElasticSuccess(response) => response.results.size
+      case ElasticSuccess(response) => response.results
       case ElasticFailure(error)    => fail(s"[$sql] failed: ${error.message}")
     }
   }
@@ -151,6 +154,68 @@ trait UnionAllCompletenessSpec extends AnyFlatSpecLike with ElasticDockerTestKit
     ) shouldBe (leftCategories + rightCategories)
   }
 
+  // ── issue #354: SQL-92 §7.10 matches branches BY POSITION, on EVERY route ─────────────────
+
+  /** The result's column NAMES come from the first branch and the branches are matched by ORDINAL
+    * POSITION — names play no part in it (`CORRESPONDING` is the opt-in for name matching, and its
+    * existence is the proof). The engine used to look the first branch's names up IN THE LEG'S ROW,
+    * which answered NULL for a column the branch names differently.
+    *
+    * `cat_01` exists only in `union_left` and `cat_13` only in `union_right`, so each branch
+    * contributes a set of `id`s that names its own index — an oracle independent of the mapping
+    * under test. Row ORDER within a leg is not guaranteed on a 3-shard index, so the assertions are
+    * on SETS.
+    */
+  "UNION ALL with an alias per branch" should "take the FIRST branch's column name, one-shot" in {
+    val rows = rowsOf(
+      s"SELECT id AS x FROM $leftIndex WHERE category = 'cat_01' LIMIT 5 " +
+      s"UNION ALL SELECT id AS y FROM $rightIndex WHERE category = 'cat_13' LIMIT 5"
+    )
+    rows should have size (2 * docsPerCategory).toLong
+    // 🔴 ONE column, called `x`, on EVERY row — branch 1's own included. Before the fix
+    // `MultiSearch.fieldAliases` merged the two aliases keyed by the SOURCE field `id`, one
+    // survived, and every row came back `{x -> null, y -> …}`.
+    rows.map(_.keys.toSeq).distinct shouldBe Seq(Seq("x"))
+    rows.map(_("x")).toSet shouldBe
+    ((1 to docsPerCategory).map(d => s"${leftIndex}_cat_01_$d") ++
+    (1 to docsPerCategory).map(d => s"${rightIndex}_cat_13_$d")).toSet
+  }
+
+  it should "answer identically when the legs are paged per leg" in {
+    val rows = rowsOf(
+      s"SELECT id AS x FROM $leftIndex WHERE category = 'cat_01' " +
+      s"UNION ALL SELECT id AS y FROM $rightIndex WHERE category = 'cat_13'"
+    )
+    rows should have size (2 * docsPerCategory).toLong
+    rows.map(_.keys.toSeq).distinct shouldBe Seq(Seq("x"))
+    rows.map(_("x")).toSet shouldBe
+    ((1 to docsPerCategory).map(d => s"${leftIndex}_cat_01_$d") ++
+    (1 to docsPerCategory).map(d => s"${rightIndex}_cat_13_$d")).toSet
+  }
+
+  /** Reordering the projection is how an analyst aligns two differently-shaped sources, and a
+    * by-name lookup silently undid it: column 1 held branch 2's `category` because that is what
+    * branch 1 called column 1.
+    */
+  "UNION ALL with a REORDERED second branch" should "bind column i to column i of that branch" in {
+    val rows = rowsOf(
+      s"SELECT category, id FROM $leftIndex WHERE category = 'cat_01' LIMIT 5 " +
+      s"UNION ALL SELECT id, category FROM $rightIndex WHERE category = 'cat_13' LIMIT 5"
+    )
+    rows should have size (2 * docsPerCategory).toLong
+    rows.map(_.keys.toSeq).distinct shouldBe Seq(Seq("category", "id"))
+    val (left, right) = rows.splitAt(docsPerCategory)
+    left.map(_("category")).toSet shouldBe Set("cat_01")
+    left.map(_("id")).toSet shouldBe (1 to docsPerCategory)
+      .map(d => s"${leftIndex}_cat_01_$d")
+      .toSet
+    // 🔴 branch 2 declared `id` FIRST, so the result's first column — the one branch 1 called
+    // `category` — holds branch 2's ids, and `id` holds its category.
+    right.map(_("category")).toSet shouldBe
+    (1 to docsPerCategory).map(d => s"${rightIndex}_cat_13_$d").toSet
+    right.map(_("id")).toSet shouldBe Set("cat_13")
+  }
+
   "A distinct UNION on the plain client" should
   "be refused with 400 naming the extension, never executed as UNION ALL" in {
     implicit val ctx: ConversionContext = NativeContext