diff --git a/CHANGELOG.md b/CHANGELOG.md
index 36e599a827..729502ef07 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - **GFQL polars engine natively runs `UNWIND` of a carried `collect()` list column**: `... WITH collect(x) AS xs UNWIND xs AS y RETURN ...` (exploding a list-valued column produced by `collect()`, the row-pipeline analogue of the IC6 unwind shape) previously declined on polars — `unwind_polars` only accepted a scalar literal list. Now explodes a `List`-dtype column reference (`with_columns` copy → `filter(list.len() > 0)` → `explode`), matching the pandas oracle exactly: empty-list/null cells → 0 rows, nulls *within* a list survive, source column retained; the pre-filter also makes it independent of the polars-2.0 `empty_as_null` default. Non-list columns, name collisions, unknown identifiers, and nested-list literals still decline (honest NIE, no pandas bridge). Differential fuzz vs pandas over ~3500 collect→UNWIND→{RETURN, grouped, filtered, ORDER BY, aggregate, nested-unwind} queries: 0 disagreements. (The `MATCH ... WITH collect(..) UNWIND .. MATCH` form — IC6 with a trailing re-traversal — is compile-rewritten into WITH→MATCH reentry, a separate path.) Tests in `test_engine_polars_row_pipeline.py`.
 - **GFQL polars engine natively runs the LDBC IC11/IC6 undirected variable-length bindings table (`-[*1..k]-`), unblocking the cross-alias `WHERE NOT(person=friend)` clause**: a bounded UNDIRECTED variable-length MATCH that materializes a bindings row table (`rows(binding_ops=...)`, emitted whenever a downstream clause — e.g. a cross-alias same-path `WHERE NOT a = b` / `a <> b` — references two node aliases on the path) previously declined on polars (`binding_rows_polars` hard-NIE'd every undirected multihop), while pandas handled it. This was the single residual blocking the official LDBC IC11/IC6 queries on polars (the cross-alias WHERE lowering itself was already native — directed var-length + `WHERE NOT a = b` already matched pandas). `binding_rows_polars` now supports undirected var-length with **`min_hops == 1`** via a doubled-pair join with immediate-backtrack avoidance, an exact port of the pandas oracle `_gfql_multihop_binding_rows` (`avoid_immediate_backtrack=True`): a `__prev__` marker (seeded null, dtype-matched to the id column) drops immediate backtracks each hop (Kleene mask: null prev kept), and the step-pair set reproduces pandas' edge multiplicity exactly — each non-loop edge contributes each directed orientation TWICE (so a length-1 pair appears x2, length-2 x4) while self-loops contribute `(u,u)` x2 only (not double-counted). The multiplicity rule was derived by instrumenting pandas' `step_pairs` (which flow from the var-length `edge_op.execute` hop + `orient_edges`), NOT by reading code alone. Scoped to `min_hops == 1` because that is where the raw-edge reconstruction provably matches pandas: every edge is trivially a length-1 path so the var-length hop's backward pruning removes nothing; `min_hops == 0` (zero-hop, undoubled) and `min_hops >= 2` (backward-pruned / long-walk divergence) still **decline with an honest `NotImplementedError`** rather than risk silent-wrong multiplicities. Differential fuzz vs the pandas oracle over ~2500 random graphs (self-loops, parallel + antiparallel edges, `*1..2`…`*1..5`, all WHERE/RETURN variants): 0 disagreements; the out-of-scope windows decline (never diverge). Tests in `test_engine_polars_binding_rows.py` (parity + multiplicity/backtrack/self-loop/string-id pins + `*0..2`/`*2..3`/`*2..2` decline pins).
 - **GFQL polars engine natively runs whole-entity `count(DISTINCT n)` / identity aggregation (#1709)**: `g.gfql("MATCH (a {..}) RETURN count(DISTINCT a) AS c")` and its grouped/filtered/traversal variants (`(a)-[]->(b) RETURN count(DISTINCT b)`) previously declined on polars — the Cypher aggregation lowers `count(DISTINCT b)` to the `__gfql_node_id__` identity sentinel, and the polars `lower_expr` `Identifier` branch resolved only the prefixed `alias.__gfql_node_id__` form, not the bare sentinel, so it NIE'd. Fix threads the graph node-id column through a `_NODE_ID` contextvar (published alongside the schema in `_lower_with_schema`, wired through the four callers) so the bare sentinel resolves to the id column **only if present**, else declines (never invents/mis-resolves a column). Differential fuzz vs the pandas oracle: 600 + 300 cases, 0 disagreements. Deliberately still honest-NIE (verified declines, not silent-wrong): whole-entity `collect(b)` (list-of-entities repr not ported, cf #1650) and multi-alias binding-table identity aggregation (bare id absent on connected-pattern binding tables — adjacent to #1273). Tests in `test_engine_polars_row_pipeline.py`.
+- **GFQL polars engine natively runs `WITH ... MATCH ...` re-traversal (the IC6/IC11 core)**: a MATCH that re-traverses from a preceding `WITH`'s projected/aggregated bindings — `MATCH (p {id:X})-[:KNOWS]-(friend) WITH DISTINCT friend MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post)` — previously declined on polars (`Cypher MATCH after WITH could not recover carried node identities from the prefix stage`). Two pandas-only gaps fixed: (1) the native polars projector now emits the `_cypher_entity_projection_meta` side-channel (carried alias `id`/`ids`) that the bounded-reentry executor reads to re-seed the next MATCH; (2) the reentry executor's id-handling and the seeded binding pipeline are made engine-aware (polars `is_not_null`/`filter`/order-preserving left-join; a semi-join of the first alias to the carried ids). `RETURN entity / property / count / count(*) / DISTINCT` over WITH→MATCH now run natively with pandas parity. Differential fuzz vs pandas over ~3500 random graphs × WITH→MATCH shapes (DISTINCT/filtered/ORDER-BY prefixes × fwd/rev/undirected re-MATCH × entity/property/count/distinct RETURN): **0 disagreements, 0 crashes** (declines are honest NIEs). Still honest-NIE (clean decline, was a crash): a WITH that also carries a **scalar column** into the trailing MATCH (hidden-payload threading is still pandas-only), and duplicate carried ids / node-cartesian trailing patterns under a seed (preserve pandas path-multiplicity / avoid silent-wrong). New `test_engine_polars_with_match_reentry.py`.
 - **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.<prop>`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`.
 - **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test).
 - **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test).
diff --git a/bin/test-polars.sh b/bin/test-polars.sh
index 4b98a1fb27..457ce82155 100755
--- a/bin/test-polars.sh
+++ b/bin/test-polars.sh
@@ -18,6 +18,7 @@ POLARS_TEST_FILES=(
     graphistry/tests/compute/gfql/test_engine_polars_chain.py
     graphistry/tests/compute/gfql/test_engine_polars_row_pipeline.py
     graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
+    graphistry/tests/compute/gfql/test_engine_polars_with_match_reentry.py
     graphistry/tests/compute/gfql/test_engine_polars_cypher_conformance.py
     graphistry/tests/compute/gfql/test_engine_polars_conformance_matrix.py
     graphistry/tests/compute/gfql/test_polars_string_predicate_nonstring.py
@@ -29,11 +30,14 @@ POLARS_TEST_FILES=(
     # index tests exercise the seeded-index hook in the polars hop entry (hop.py) — without
     # them the hook dominates the now-thin file and trips its per-file coverage floor
     graphistry/tests/compute/gfql/index/test_index.py
+    # engine-agnostic frame/series primitives (graphistry/Engine.py) — the polars branches of
+    # these dispatch helpers are only measured when this lane covers graphistry (see cov widen below)
+    graphistry/tests/test_engine_frame_helpers.py
 )
 
 COV_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_ARGS=(--cov=graphistry/compute --cov-report=)
+    COV_ARGS=(--cov=graphistry --cov-report=)
 fi
 
 python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
@@ -42,7 +46,7 @@ python -B -m pytest -vv "${COV_ARGS[@]}" "${POLARS_TEST_FILES[@]}" "$@"
 # appended into the same coverage data file when POLARS_COV=1 (CI audit reads it)
 COV_APPEND_ARGS=()
 if [ -n "${POLARS_COV:-}" ]; then
-    COV_APPEND_ARGS=(--cov=graphistry/compute --cov-report= --cov-append)
+    COV_APPEND_ARGS=(--cov=graphistry --cov-report= --cov-append)
 fi
 python -B -m pytest -vv "${COV_APPEND_ARGS[@]}" \
     graphistry/tests/compute/gfql/cypher/test_lowering.py -k polars
diff --git a/graphistry/Engine.py b/graphistry/Engine.py
index 6cbafa653d..6650229f80 100644
--- a/graphistry/Engine.py
+++ b/graphistry/Engine.py
@@ -3,12 +3,18 @@
 import numpy as np
 import pandas as pd
 import pyarrow as pa
-from typing import Any, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Union
 from typing_extensions import Literal
 from enum import Enum
 
 from graphistry.models.types import ValidationParam
 
+if TYPE_CHECKING:
+    # Frame aliases (pandas types for mypy; ``Any`` at runtime). Imported under TYPE_CHECKING
+    # and referenced via string annotations below so Engine.py — imported very early — never
+    # triggers ``graphistry.compute`` package init at runtime (would be circular).
+    from graphistry.compute.typing import DataFrameT, SeriesT
+
 
 class Engine(Enum):
     PANDAS = 'pandas'
@@ -850,3 +856,103 @@ def safe_merge(
         raise ValueError("Must specify either 'on' or both 'left_on' and 'right_on'")
 
     return result
+
+
+# ---------------------------------------------------------------------------
+# Engine-agnostic series / frame primitives (pandas / cuDF / polars dispatch).
+#
+# Pure per-row/per-column dispatch helpers with no domain knowledge — the polars
+# branches genuinely return polars objects mypy can't narrow to the pandas frame
+# aliases, so the localized ``# type: ignore`` lives here at the dispatch point and
+# callers get a clean ``SeriesT`` / ``DataFrameT`` contract with no ``cast()``.
+# Annotations are strings so the TYPE_CHECKING-only alias import stays runtime-free.
+# ---------------------------------------------------------------------------
+
+
+def is_series_like(s: object) -> bool:
+    """True for a pandas/cuDF Series (``.dropna``) or a polars Series (module check).
+
+    Some engine-agnostic callers accept an ``ids`` Series that is pandas under
+    ``engine='pandas'`` and polars under ``engine='polars'``; both are valid."""
+    return hasattr(s, "dropna") or is_polars_df(s)
+
+
+def series_not_null_mask(s: "SeriesT") -> "SeriesT":
+    """Non-null boolean mask, engine-aware (polars ``is_not_null`` vs pandas ``notna``)."""
+    if is_polars_df(s):
+        return s.is_not_null()  # type: ignore[attr-defined,no-any-return]
+    return s.notna()
+
+
+def series_filter(s: "SeriesT", mask: "SeriesT") -> "SeriesT":
+    """Filter a Series by a boolean mask, engine-aware, dropping the old index (pandas)."""
+    if is_polars_df(s):
+        return s.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return s[mask].reset_index(drop=True)
+
+
+def frame_filter(df: "DataFrameT", mask: "SeriesT") -> "DataFrameT":
+    """Filter a DataFrame's rows by a boolean mask, engine-aware, dropping the old index."""
+    if is_polars_df(df):
+        return df.filter(mask)  # type: ignore[attr-defined,no-any-return]
+    return df.loc[mask].reset_index(drop=True)
+
+
+def ordered_left_join(left: "DataFrameT", right: "DataFrameT", *, on: str) -> "DataFrameT":
+    """Left join preserving ``left`` row order, engine-aware. Polars ``.merge`` does not exist;
+    ``safe_merge`` (pandas/cuDF ``.merge``) cannot run on polars frames, so branch to
+    ``.join(..., maintain_order='left')`` which pins the left-row ordering the caller needs.
+
+    ``right`` may arrive on a different engine than ``left`` (e.g. natively-projected polars
+    ``left`` against a still-pandas base table), so align ``right`` onto ``left``'s engine
+    before the polars join."""
+    if is_polars_df(left):
+        if not is_polars_df(right):
+            right = df_to_engine(right, Engine.POLARS)
+        return left.join(right, on=on, how="left", maintain_order="left")  # type: ignore[call-arg,no-any-return]
+    return safe_merge(left, right, on=on, how="left")
+
+
+def row_as_mapping(rows: "DataFrameT", row_index: int) -> Mapping[str, Any]:
+    """One frame row as a col->scalar mapping, engine-aware (``row[col]`` works for
+    both the pandas Series and the polars named-row dict)."""
+    if is_polars_df(rows):
+        return rows.row(row_index, named=True)  # type: ignore[attr-defined,no-any-return]
+    return rows.iloc[row_index]
+
+
+def assign_constant_columns(df: "DataFrameT", values: Dict[str, Any]) -> "DataFrameT":
+    """Broadcast scalar ``values`` as constant columns, engine-aware."""
+    if not values:
+        return df
+    if is_polars_df(df):
+        import polars as pl
+        return df.with_columns([pl.lit(v).alias(k) for k, v in values.items()])  # type: ignore[attr-defined,no-any-return]
+    return df.assign(**values)
+
+
+def drop_columns(df: "DataFrameT", cols: Sequence[str]) -> "DataFrameT":
+    """Drop columns by name, engine-aware (polars ``drop(list)`` vs pandas ``drop(columns=)``)."""
+    if is_polars_df(df):
+        return df.drop(list(cols))  # type: ignore[no-any-return]
+    return df.drop(columns=list(cols))
+
+
+def series_to_pylist(values: "SeriesT") -> List[Any]:
+    """Series -> python list, engine-aware, with defensive arrow/pandas fallbacks."""
+    if hasattr(values, "to_arrow"):
+        try:
+            return list(values.to_arrow().to_pylist())
+        except Exception:
+            pass
+    if hasattr(values, "to_pandas"):
+        try:
+            return list(values.to_pandas().tolist())
+        except Exception:
+            pass
+    if hasattr(values, "tolist"):
+        try:
+            return list(values.tolist())
+        except Exception:
+            pass
+    return list(values)
diff --git a/graphistry/compute/gfql/cypher/reentry/execution.py b/graphistry/compute/gfql/cypher/reentry/execution.py
index 6028a60d21..0b8c82d71b 100644
--- a/graphistry/compute/gfql/cypher/reentry/execution.py
+++ b/graphistry/compute/gfql/cypher/reentry/execution.py
@@ -2,9 +2,24 @@
 # ruff: noqa: E501
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional, Sequence, Tuple, Union, cast
-
-from graphistry.Engine import EngineAbstract, df_concat, df_cons, resolve_engine, safe_merge
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union, cast
+
+from graphistry.Engine import (
+    EngineAbstract,
+    assign_constant_columns,
+    df_concat,
+    df_cons,
+    drop_columns,
+    frame_filter,
+    is_series_like,
+    ordered_left_join,
+    resolve_engine,
+    row_as_mapping,
+    safe_merge,
+    series_filter,
+    series_not_null_mask,
+    series_to_pylist,
+)
 from graphistry.Plottable import Plottable
 from graphistry.compute.exceptions import GFQLValidationError, ErrorCode
 from graphistry.compute.gfql.cypher.reentry.naming import _reentry_hidden_column_name
@@ -23,30 +38,6 @@
 from graphistry.Engine import is_polars_df as _is_polars_df
 
 
-def _reentry_row(prefix_rows: Any, row_index: int) -> Any:
-    """One prefix row as a col->scalar mapping, engine-aware (``row[col]`` works for
-    both the pandas Series and the polars named-row dict)."""
-    if _is_polars_df(prefix_rows):
-        return prefix_rows.row(row_index, named=True)
-    return prefix_rows.iloc[row_index]
-
-
-def _assign_constant_columns(df: Any, values: Dict[str, Any]) -> Any:
-    """Broadcast scalar ``values`` as constant columns, engine-aware."""
-    if not values:
-        return df
-    if _is_polars_df(df):
-        import polars as pl
-        return df.with_columns([pl.lit(v).alias(k) for k, v in values.items()])
-    return df.assign(**values)
-
-
-def _drop_columns(df: Any, cols: Sequence[str]) -> Any:
-    if _is_polars_df(df):
-        return df.drop(list(cols))
-    return df.drop(columns=list(cols))
-
-
 def _bind_reentry_graph(graph: Plottable, node_rows: Optional[DataFrameT], *, empty_edges: bool = False) -> Plottable:
     out = graph.bind()
     out._nodes = node_rows
@@ -183,7 +174,7 @@ def _optional_reentry_key(record: Dict[str, Any], columns: Tuple[str, ...]) -> T
 
 
 def _records_for_columns(df: DataFrameT, columns: Tuple[str, ...]) -> List[Dict[str, Any]]:
-    values_by_column = {column: _series_to_pylist(cast(SeriesT, df[column])) for column in columns}
+    values_by_column = {column: series_to_pylist(cast(SeriesT, df[column])) for column in columns}
     row_count = len(df)
     return [
         {column: values_by_column[column][row_index] for column in columns}
@@ -191,14 +182,6 @@ def _records_for_columns(df: DataFrameT, columns: Tuple[str, ...]) -> List[Dict[
     ]
 
 
-def _series_to_pylist(values: SeriesT) -> List[Any]:
-    if hasattr(values, "to_arrow"):
-        return cast(List[Any], values.to_arrow().to_pylist())
-    if hasattr(values, "tolist"):
-        return cast(List[Any], values.tolist())
-    return list(values)
-
-
 def _optional_reentry_key_value(value: Any) -> Any:
     try:
         if value != value:
@@ -274,7 +257,7 @@ def compiled_query_reentry_state(
         )
     ids = meta["ids"]
     id_column = meta["id_column"]
-    if not hasattr(ids, "dropna"):
+    if not is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not recover carried node identities from the prefix stage",
             value=output_name,
@@ -306,6 +289,17 @@ def compiled_query_reentry_state(
             value=output_name,
             suggestion=REENTRY_SCALAR_SUGGESTION,
         )
+    if _is_polars_df(carried_ids) or _is_polars_df(aligned_prefix_rows) or _is_polars_df(base_nodes):
+        # Whole-row re-entry that ALSO carries scalar WITH columns (e.g. `WITH a, a.val AS av
+        # MATCH (a)-...`) threads hidden ``__cypher_reentry_*`` payload columns through the
+        # binding pipeline; that carry path is pandas-only so far. The seed-only whole-row case
+        # (no carried scalars) returned above. Decline honestly rather than run the pandas-only
+        # payload merge on polars frames (parity-or-NIE; no silent bridge).
+        raise NotImplementedError(
+            "polars engine does not yet natively support Cypher WITH -> MATCH re-entry that carries "
+            "scalar WITH columns into the trailing MATCH; use engine='pandas' for this query "
+            "(no pandas fallback; parity-or-error by design)"
+        )
     duplicate_mask = carried_ids.duplicated()
     if bool(duplicate_mask.any()) if hasattr(duplicate_mask, "any") else False:
         raise reentry_validation_error(
@@ -397,10 +391,10 @@ def compiled_query_scalar_reentry_state(
             value=missing_column,
             suggestion="Project the scalar column explicitly before MATCH re-entry.",
         )
-    row = _reentry_row(prefix_rows, row_index)
+    row = row_as_mapping(prefix_rows, row_index)
     node_rows = cast(
         DataFrameT,
-        _assign_constant_columns(
+        assign_constant_columns(
             base_nodes,
             {
                 _reentry_hidden_column_name(output_name): row[output_name]
@@ -420,7 +414,7 @@ def freeform_broadcast_row_to_nodes(
     row_index: int,
 ) -> Plottable:
     """Broadcast one free-form prefix row's hidden carries onto the base nodes."""
-    row = _reentry_row(prefix_rows, row_index)
+    row = row_as_mapping(prefix_rows, row_index)
     broadcast_values: Dict[str, Any] = {
         _reentry_hidden_column_name(col): row[col]
         for col in plan.scalar_columns
@@ -435,11 +429,11 @@ def freeform_broadcast_row_to_nodes(
     if broadcast_values:
         existing_hidden = [c for c in base_nodes.columns if isinstance(c, str) and c.startswith("__cypher_reentry_")]
         node_rows = (
-            cast(DataFrameT, _drop_columns(base_nodes, existing_hidden))
+            cast(DataFrameT, drop_columns(base_nodes, existing_hidden))
             if existing_hidden
             else base_nodes
         )
-        node_rows = cast(DataFrameT, _assign_constant_columns(node_rows, broadcast_values))
+        node_rows = cast(DataFrameT, assign_constant_columns(node_rows, broadcast_values))
     else:
         node_rows = cast(DataFrameT, base_nodes)
 
@@ -488,18 +482,18 @@ def aligned_reentry_rows(
             value=output_name,
             suggestion="Retry with a direct whole-row carry through WITH or inspect intermediate row-shaping before MATCH re-entry.",
         )
-    if not hasattr(ids, "notna"):
+    if not is_series_like(ids):
         raise reentry_validation_error(
             "Cypher MATCH after WITH could not align carried node identities from the prefix stage",
             value=output_name,
             suggestion=REENTRY_WHOLE_ROW_SUGGESTION,
         )
 
-    non_null_mask = cast(SeriesT, ids.notna())
-    carried_ids = cast(SeriesT, ids[non_null_mask].reset_index(drop=True))
+    non_null_mask = series_not_null_mask(ids)
+    carried_ids = series_filter(ids, non_null_mask)
     if prefix_rows is None:
         return carried_ids, None
-    return carried_ids, cast(DataFrameT, prefix_rows.loc[non_null_mask].reset_index(drop=True))
+    return carried_ids, frame_filter(prefix_rows, non_null_mask)
 
 
 def reentry_carry_payload(
@@ -533,4 +527,4 @@ def ordered_reentry_start_nodes(
     id_column: str,
 ) -> DataFrameT:
     # MATCH re-entry must preserve the WITH row order, not the base node-table order.
-    return cast(DataFrameT, safe_merge(carried_node_ids, node_rows, on=id_column, how="left"))
+    return ordered_left_join(carried_node_ids, node_rows, on=id_column)
diff --git a/graphistry/compute/gfql/lazy/engine/polars/projection.py b/graphistry/compute/gfql/lazy/engine/polars/projection.py
index 26c93817ef..4a58716436 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/projection.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/projection.py
@@ -10,7 +10,7 @@
 """
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from graphistry.Plottable import Plottable
 
@@ -132,12 +132,43 @@ def _flat_entity_exprs_polars(rows_df: pl.DataFrame, projection: ResultProjectio
     return out
 
 
+def _record_entity_meta(
+    entity_meta: Dict[str, Dict[str, Any]],
+    rows_df: pl.DataFrame,
+    projection: ResultProjectionPlan,
+    source_alias: str,
+    output_name: str,
+    id_column: Optional[str],
+) -> None:
+    """Record whole-entity projection metadata for one column, mirroring the pandas projector.
+
+    ``_try_native_projection`` reaches this only in the single-entity branch (flat exprs and the
+    entity-text path both decline multi-entity prefixed columns), so ``rows_df`` is the aligned
+    source frame and ``rows_df[id_column]`` is the carried alias's id column, row-aligned with the
+    projected output. Snapshot (``.clone()``) the id column so downstream reentry recovery never
+    aliases a later-mutated working frame (see #1356)."""
+    if id_column is None or id_column not in rows_df.columns:  # pragma: no cover - defensive: node re-entry always carries the id column
+        return
+    entity_meta[output_name] = {
+        "table": projection.table,
+        "alias": source_alias,
+        "id_column": id_column,
+        "ids": rows_df.get_column(id_column).clone(),
+    }
+
+
 def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection: ResultProjectionPlan, structured: bool) -> Optional[Plottable]:
     """Native projection for property/expr columns already in the polars row table + structured-
     flat or entity-text whole-entity returns; None → caller raises NIE."""
     import polars as pl
 
     exprs = []
+    # Whole-entity projection metadata side-channel (#1273 WITH->MATCH re-entry): mirror the
+    # pandas projector (result_postprocess._apply_result_projection_pandas), which records the
+    # carried alias's id column so the bounded-reentry executor can recover carried node
+    # identities. Without it a WITH-projected node alias feeding a trailing MATCH declines.
+    entity_meta: Dict[str, Dict[str, Any]] = {}
+    id_column = result._node
     for column in projection.columns:
         if column.kind == "whole_row":
             if projection.table != "nodes":
@@ -146,15 +177,16 @@ def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection:
             if structured:
                 # #1650 default: flatten to {output}.{field} (near-free, any dtype);
                 # text fallback only for synthesized-absent rows.
-                id_column = result._node
                 flat = _flat_entity_exprs_polars(rows_df, projection, source_alias, column.output_name, id_column)
                 if flat is not None:
                     exprs.extend(flat)
+                    _record_entity_meta(entity_meta, rows_df, projection, source_alias, column.output_name, id_column)
                     continue
             ent = _native_node_entity_text_expr(rows_df, source_alias, projection.exclude_columns)
             if ent is None:
                 return None
             exprs.append(ent.alias(column.output_name))
+            _record_entity_meta(entity_meta, rows_df, projection, source_alias, column.output_name, id_column)
             continue
         src = column.source_name
         if src is None or src not in rows_df.columns:
@@ -172,6 +204,8 @@ def _try_native_projection(result: Plottable, rows_df: pl.DataFrame, projection:
         return None
     out = result.bind()
     out._nodes = rows_df.select(exprs)
+    if entity_meta:
+        setattr(out, "_cypher_entity_projection_meta", entity_meta)
     edges_df = result._edges
     if edges_df is not None:
         out._edges = edges_df.clear() if _is_polars_frame(edges_df) else edges_df[:0]
diff --git a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
index 0a59a5c91e..bd98c02379 100644
--- a/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
+++ b/graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
@@ -1050,15 +1050,37 @@ def _names(lf: pl.LazyFrame) -> List[str]:
     dst = g._destination
     if nodes is None or edges is None or node_id is None or src is None or dst is None:
         return None
-    if getattr(g, "_gfql_start_nodes", None) is not None:
-        # Bounded re-entry seeds the first alias from carried rows — pandas-only.
-        return None
+    seed_ids_lf: Optional[Any] = None  # LazyFrame; Any avoids the union-typed seed_nodes.join mismatch
+    start_nodes = getattr(g, "_gfql_start_nodes", None)
+    if start_nodes is not None:
+        # Bounded WITH->MATCH re-entry (#1273): the carried WITH rows seed the first
+        # alias. Constrain the first node alias to the carried ids via a semi-join —
+        # the native twin of the pandas wavefront seed. Support only UNIQUE carried
+        # ids: then the semi-join contributes each seed node exactly once, matching the
+        # pandas seed row-for-row; duplicate carried ids could change path multiplicity,
+        # so decline (return None -> honest NIE), never risk a silent-wrong count.
+        from graphistry.Engine import Engine as _Engine, df_to_engine as _df_to_engine, is_polars_df as _is_polars
+        sn = start_nodes.collect() if isinstance(start_nodes, pl.LazyFrame) else start_nodes
+        if not _is_polars(sn):
+            sn = _df_to_engine(sn, _Engine.POLARS)
+        if node_id not in sn.columns:
+            return None
+        seed_ids = sn.select(pl.col(node_id)).drop_nulls()
+        if seed_ids.height != seed_ids.unique().height:
+            return None
+        seed_ids_lf = seed_ids.lazy()
 
     ops: List[ASTObject] = [ast_from_json(op_json, validate=False) for op_json in binding_ops]
     # Shared validation (engine-agnostic): raises the canonical GFQLValidationError
     # for malformed op sequences / duplicate aliases — same error as pandas.
     RowPipelineMixin._gfql_validate_binding_ops(ops)
     if RowPipelineMixin._gfql_binding_ops_mode(ops) == "node_cartesian":
+        if seed_ids_lf is not None:
+            # A WITH->MATCH seed constrains the FIRST alias, but the cartesian builder
+            # below does not thread it; running it would silently ignore the seed
+            # (wrong cross-product). Decline honestly (the alternating-path seed is
+            # applied at seed_nodes; node-cartesian re-entry stays pandas-only).
+            return None
         # MATCH (a), (b), ... disconnected node aliases: native cross-product (#1273).
         return _cartesian_node_bindings_polars(g, ops, node_id)
     if RowPipelineMixin._gfql_is_shortest_path_scalar_binding_ops(ops):
@@ -1132,6 +1154,9 @@ def _names(lf: pl.LazyFrame) -> List[str]:
         if not isinstance(first_op, ASTNode):
             return None
         seed_nodes = filter_by_dict_polars(nodes_lf, first_op.filter_dict)
+        if seed_ids_lf is not None:
+            # WITH->MATCH re-entry seed: constrain the first alias to the carried ids.
+            seed_nodes = seed_nodes.join(seed_ids_lf, on=node_id, how="semi")
         state = seed_nodes.select(pl.col(node_id).alias("__current__"))
         alias_frames: Dict[str, pl.LazyFrame] = {}
         node_aliases: List[str] = []
diff --git a/graphistry/compute/gfql_unified.py b/graphistry/compute/gfql_unified.py
index d0a6f0603e..d67b5ec46f 100644
--- a/graphistry/compute/gfql_unified.py
+++ b/graphistry/compute/gfql_unified.py
@@ -6,7 +6,7 @@
 from types import MappingProxyType
 from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union, cast
 from graphistry.Plottable import Plottable
-from graphistry.Engine import Engine, EngineAbstract, POLARS_ENGINES, df_concat, df_cons, df_to_engine, df_unique, is_polars_df, resolve_engine
+from graphistry.Engine import Engine, EngineAbstract, POLARS_ENGINES, df_concat, df_cons, df_to_engine, df_unique, is_polars_df, resolve_engine, series_to_pylist
 from graphistry.util import setup_logger
 from .ast import ASTObject, ASTLet, ASTNode, ASTEdge, ASTCall
 from .chain import Chain, chain as chain_impl
@@ -83,25 +83,6 @@
 logger = setup_logger(__name__)
 
 
-def _series_to_pylist(values: Any) -> List[Any]:
-    if hasattr(values, "to_arrow"):
-        try:
-            return list(values.to_arrow().to_pylist())
-        except Exception:
-            pass
-    if hasattr(values, "to_pandas"):
-        try:
-            return list(values.to_pandas().tolist())
-        except Exception:
-            pass
-    if hasattr(values, "tolist"):
-        try:
-            return list(values.tolist())
-        except Exception:
-            pass
-    return list(values)
-
-
 def _is_duplicate_carried_rows_reentry_error(exc: GFQLValidationError) -> bool:
     context = getattr(exc, "context", None)
     if exc.code != ErrorCode.E108 or not isinstance(context, dict):
@@ -217,8 +198,8 @@ def _apply_optional_null_fill(
             language="cypher",
         )
 
-    base_ids = _series_to_pylist(base_rows_df[node_col])
-    matched_id_list = _series_to_pylist(matched_ids)
+    base_ids = series_to_pylist(base_rows_df[node_col])
+    matched_id_list = series_to_pylist(matched_ids)
     if len(base_ids) == actual_rows and base_ids == matched_id_list:
         return result
 
diff --git a/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json b/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json
index a47db2309c..e82f5ad084 100644
--- a/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json
+++ b/graphistry/tests/compute/gfql/coverage_baselines/ci-pandas-py3.12.json
@@ -35,7 +35,7 @@
     "graphistry/compute/gfql/cypher/reentry/__init__.py": 100.0,
     "graphistry/compute/gfql/cypher/reentry/carry.py": 82.05,
     "graphistry/compute/gfql/cypher/reentry/compiletime.py": 85.48,
-    "graphistry/compute/gfql/cypher/reentry/execution.py": 84.75,
+    "graphistry/compute/gfql/cypher/reentry/execution.py": 83.5,
     "graphistry/compute/gfql/cypher/reentry/flatten.py": 94.05,
     "graphistry/compute/gfql/cypher/reentry/lowering_support.py": 88.38,
     "graphistry/compute/gfql/cypher/reentry/naming.py": 100.0,
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
index cb2ee0ca76..0d6474c045 100644
--- a/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
+++ b/graphistry/tests/compute/gfql/test_engine_polars_binding_rows.py
@@ -386,9 +386,20 @@ def test_polars_binding_rows_decline_branches_direct():
     no_edges = graphistry.nodes(pl.from_pandas(NODES), "id")
     assert binding_rows_polars(no_edges, bo) is None
 
+    # WITH->MATCH re-entry seed (#1273): a UNIQUE carried-id seed now lowers natively — the
+    # first alias is semi-joined to the seed ids (native twin of the pandas wavefront).
     seeded = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
     setattr(seeded, "_gfql_start_nodes", pl.DataFrame({"id": [0]}))
-    assert binding_rows_polars(seeded, bo) is None
+    seeded_out = binding_rows_polars(seeded, bo)
+    assert seeded_out is not None
+    # seed {0} -[F]-> b: EDGES forward from 0 with type F -> b in {1}; the bindings 'a' column
+    # is pinned to the seed and every row's 'a' id is 0.
+    seeded_rows = seeded_out._nodes
+    assert set(seeded_rows["a"].to_list()) == {0}
+    # A DUPLICATE-id seed still declines (semi-join would drop path multiplicity vs pandas).
+    dup = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
+    setattr(dup, "_gfql_start_nodes", pl.DataFrame({"id": [0, 0]}))
+    assert binding_rows_polars(dup, bo) is None
 
     g = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
     # node-only cartesian (#1273) is now natively supported for <=3 named aliases;
diff --git a/graphistry/tests/compute/gfql/test_engine_polars_with_match_reentry.py b/graphistry/tests/compute/gfql/test_engine_polars_with_match_reentry.py
new file mode 100644
index 0000000000..f3a8734ccd
--- /dev/null
+++ b/graphistry/tests/compute/gfql/test_engine_polars_with_match_reentry.py
@@ -0,0 +1,327 @@
+"""Differential parity: native polars ``WITH ... MATCH ...`` re-entry == pandas (oracle) (#1273).
+
+A ``WITH`` clause that projects a whole-row node alias (optionally ``DISTINCT`` / filtered /
+ordered+bounded) feeds a SUBSEQUENT ``MATCH`` that re-traverses from those carried nodes. Before
+this change the polars engine declined the WITH->MATCH boundary ("could not recover carried node
+identities") because the native projector never emitted the whole-entity id side-channel and the
+bounded-reentry executor + seeded binding pipeline were pandas-only.
+
+Pandas is the oracle: every supported query returns an identical result table, polars-typed (no
+pandas bridge). Shapes still out of subset (scalar WITH columns carried alongside the whole-row
+alias) must raise NotImplementedError, never silently diverge. Companion of the seeded
+``rows(binding_ops)`` work in test_engine_polars_binding_rows.
+"""
+import random
+
+import pandas as pd
+import pytest
+
+import graphistry
+
+pl = pytest.importorskip("polars")
+
+
+# ---------------------------------------------------------------------------
+# Fixtures / helpers
+# ---------------------------------------------------------------------------
+
+NODES = pd.DataFrame({
+    "id": [0, 1, 2, 3, 4, 5],
+    "kind": ["person", "person", "post", "post", "post", "comment"],
+    "val": [10, 20, 30, 40, 50, 60],
+})
+EDGES = pd.DataFrame({
+    "s": [0, 0, 1, 1, 2, 3, 0, 4],
+    "d": [1, 2, 2, 3, 4, 4, 3, 5],
+    "type": ["KNOWS", "HAS_CREATOR", "HAS_CREATOR", "KNOWS", "LINK", "LINK", "KNOWS", "HAS_CREATOR"],
+})
+BASE = graphistry.nodes(NODES, "id").edges(EDGES, "s", "d")
+
+
+def _to_pandas(df):
+    if df is not None and "polars" in type(df).__module__:
+        return df.to_pandas()
+    return df
+
+
+def _round_floats(df):
+    out = df.copy()
+    for col in out.columns:
+        s = out[col]
+        if pd.api.types.is_bool_dtype(s):
+            continue
+        if pd.api.types.is_numeric_dtype(s):
+            out[col] = s.astype("float64").round(6)
+    return out
+
+
+def _norm(df):
+    df = _round_floats(df).where(lambda d: d.notna(), "∅")
+    return df.astype(str).sort_values(list(df.columns)).reset_index(drop=True)
+
+
+def _assert_parity(g, query, *, order_sensitive=False):
+    rpd = g.gfql(query, engine="pandas")._nodes
+    rpl = g.gfql(query, engine="polars")._nodes
+    assert "polars" in type(rpl).__module__, f"expected polars frame for {query!r}"
+    a = _to_pandas(rpd).reset_index(drop=True)
+    b = _to_pandas(rpl).reset_index(drop=True)
+    assert list(a.columns) == list(b.columns), (
+        f"columns differ for {query!r}: {list(a.columns)} vs {list(b.columns)}"
+    )
+    assert len(a) == len(b), f"row count differs for {query!r}: {len(a)} vs {len(b)}"
+    if len(a) == 0:
+        return
+    an, bn = _norm(a), _norm(b)
+    if order_sensitive:
+        an = _round_floats(a).where(lambda d: d.notna(), "∅").astype(str)
+        bn = _round_floats(b).where(lambda d: d.notna(), "∅").astype(str)
+    pd.testing.assert_frame_equal(an, bn, check_dtype=False)
+
+
+# ---------------------------------------------------------------------------
+# Curated parity corpus (whole-row WITH -> MATCH re-entry, native on both engines)
+# ---------------------------------------------------------------------------
+
+CORPUS = [
+    # DISTINCT whole-row carry -> forward re-entry, various RETURN shapes
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post) AS c",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(*) AS c",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post.id AS pid",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post.id AS pid, post.kind AS pk",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN DISTINCT post.kind AS pk",
+    # non-DISTINCT whole-row carry (prefix may yield duplicate friend rows)
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post) AS c",
+    # filtered WITH prefix
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WHERE friend.val > 15 WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post) AS c",
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WHERE friend.kind = 'person' WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post.id AS pid",
+    # ORDER BY + bounded LIMIT prefix (preserves WITH row order into re-entry)
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH friend ORDER BY friend.val LIMIT 2 "
+    "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post.id AS pid",
+    # directed re-entry (forward / reverse) and undirected prefix
+    "MATCH (p {id:0})<-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]->(post) RETURN count(post) AS c",
+    "MATCH (p {id:0})-[:KNOWS]->(friend) WITH DISTINCT friend "
+    "MATCH (friend)<-[:HAS_CREATOR]-(post) RETURN count(post) AS c",
+    # suffix endpoint property filter
+    "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+    "MATCH (friend)-[:HAS_CREATOR]-(post {kind:'post'}) RETURN count(post) AS c",
+]
+
+
+@pytest.mark.parametrize("query", CORPUS)
+def test_with_match_reentry_parity(query):
+    _assert_parity(BASE, query)
+
+
+def test_reentry_count_pin():
+    """Exact-value pin for the canonical IC6/IC11-shaped re-entry (regression guard)."""
+    q = (
+        "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+        "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN count(post) AS c"
+    )
+    rpl = _to_pandas(BASE.gfql(q, engine="polars")._nodes)
+    rpd = _to_pandas(BASE.gfql(q, engine="pandas")._nodes)
+    assert rpl["c"].tolist() == rpd["c"].tolist()
+    assert rpl["c"].tolist() == [1]
+
+
+def test_projector_emits_entity_projection_meta_polars():
+    """The native polars projector must emit the whole-entity id side-channel the bounded
+    reentry executor reads to recover carried node identities (the root-cause gap). A terminal
+    ``RETURN friend`` exercises the same whole-entity projection branch."""
+    prefix = "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend RETURN friend"
+    res = BASE.gfql(prefix, engine="polars")
+    meta = getattr(res, "_cypher_entity_projection_meta", None)
+    assert isinstance(meta, dict) and "friend" in meta, meta
+    entry = meta["friend"]
+    assert entry["table"] == "nodes"
+    assert entry["id_column"] == "id"
+    # ids Series is polars-typed and row-aligned with the (post-DISTINCT) prefix rows.
+    ids = entry["ids"]
+    assert "polars" in type(ids).__module__
+    assert ids.to_list() == [1, 3]
+
+
+def test_projector_text_path_emits_meta_polars():
+    """The entity-TEXT projection branch (structured=False, single int/str/bool node) must also
+    emit the id side-channel — mirrors the pandas projector, which records meta for whole-entity
+    columns regardless of structured/text rendering."""
+    from graphistry.compute.gfql.cypher.lowering import ResultProjectionColumn, ResultProjectionPlan
+    from graphistry.compute.gfql.lazy.engine.polars.projection import apply_result_projection_polars
+
+    rows = pl.DataFrame({"id": [7, 9], "val": [1, 2], "n": [True, True]})
+    g = graphistry.nodes(rows, "id")
+    plan = ResultProjectionPlan(
+        alias="n",
+        table="nodes",
+        columns=(ResultProjectionColumn(output_name="n", kind="whole_row", source_name="n"),),
+        exclude_columns=(),
+    )
+    out = apply_result_projection_polars(g, plan, structured=False)
+    meta = getattr(out, "_cypher_entity_projection_meta", None)
+    assert isinstance(meta, dict) and "n" in meta
+    assert meta["n"]["id_column"] == "id"
+    assert meta["n"]["ids"].to_list() == [7, 9]
+
+
+def test_binding_rows_seed_pandas_and_missing_id_col():
+    """Seed-frame normalization branches in binding_rows_polars: a pandas seed is converted to
+    polars; a seed lacking the node-id column declines (None)."""
+    from graphistry.compute.ast import n as _n, e_forward as _ef, serialize_binding_ops as _ser
+    from graphistry.compute.gfql.lazy.engine.polars.row_pipeline import binding_rows_polars
+
+    bo = _ser([_n(name="a"), _ef(), _n(name="b")])
+    g = graphistry.nodes(pl.from_pandas(NODES), "id").edges(pl.from_pandas(EDGES), "s", "d")
+
+    # pandas seed frame -> df_to_engine conversion path
+    seeded_pd = g.bind()
+    setattr(seeded_pd, "_gfql_start_nodes", pd.DataFrame({"id": [0]}))
+    out = binding_rows_polars(seeded_pd, bo)
+    assert out is not None
+    assert set(out._nodes["a"].to_list()) == {0}
+
+    # seed frame missing the node-id column -> decline
+    seeded_bad = g.bind()
+    setattr(seeded_bad, "_gfql_start_nodes", pl.DataFrame({"other": [0]}))
+    assert binding_rows_polars(seeded_bad, bo) is None
+
+    # seeded node-cartesian (disconnected trailing aliases) -> decline: the cartesian builder
+    # doesn't thread the seed, so running it would silently ignore the WITH constraint.
+    from graphistry.compute.ast import n as _n2, serialize_binding_ops as _ser2
+    cart = _ser2([_n2(name="a"), _n2(name="b")])
+    seeded_cart = g.bind()
+    setattr(seeded_cart, "_gfql_start_nodes", pl.DataFrame({"id": [0]}))
+    assert binding_rows_polars(seeded_cart, cart) is None
+
+
+def test_with_match_reentry_binding_table_parity(monkeypatch):
+    """End-to-end parity for the case that routes the WITH re-entry seed through the multi-alias
+    BINDING-ROWS path (`binding_rows_polars`), not the single-alias hop path: a trailing MATCH that
+    binds two fresh aliases and RETURNs both forces a binding table, seeded from the carried WITH ids.
+    The direct-call tests above pin the seed branches in isolation; this proves the whole cypher
+    stack (lower -> reentry -> seeded binding_rows -> project) matches pandas on real output columns.
+    Guards that the seeded binding path is actually exercised, so it can't silently stop covering it."""
+    from graphistry.compute.gfql.lazy.engine.polars import row_pipeline as _rp
+
+    seeded_calls = {"n": 0}
+    _orig = _rp.binding_rows_polars
+
+    def _counting(g, binding_ops, *a, **k):
+        if getattr(g, "_gfql_start_nodes", None) is not None:
+            seeded_calls["n"] += 1
+        return _orig(g, binding_ops, *a, **k)
+
+    monkeypatch.setattr(_rp, "binding_rows_polars", _counting)
+
+    # (friend)-[:HAS_CREATOR]-(a)-[:LINK]-(b): two fresh aliases + multi-alias RETURN -> binding table,
+    # with the first alias (friend) constrained to the carried WITH ids.
+    q = (
+        "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend "
+        "MATCH (friend)-[:HAS_CREATOR]-(a)-[:LINK]-(b) RETURN a.id AS aid, b.id AS bid"
+    )
+    _assert_parity(BASE, q)
+    assert seeded_calls["n"] > 0, "query did not route through the seeded binding_rows path"
+
+
+def test_scalar_carry_declines_cleanly_polars():
+    """WITH that carries a scalar column ALONGSIDE the whole-row alias into the trailing MATCH is
+    pandas-only so far; polars must DECLINE (NotImplementedError), never crash or silently diverge.
+    pandas still answers it (oracle), so this documents an honest capability gap."""
+    q = (
+        "MATCH (p {id:0})-[:KNOWS]-(friend) WITH DISTINCT friend, friend.val AS fv "
+        "MATCH (friend)-[:HAS_CREATOR]-(post) RETURN post.id AS pid, fv"
+    )
+    # pandas oracle succeeds
+    _to_pandas(BASE.gfql(q, engine="pandas")._nodes)
+    # polars declines honestly
+    with pytest.raises(NotImplementedError):
+        BASE.gfql(q, engine="polars")
+
+
+# ---------------------------------------------------------------------------
+# Seeded differential fuzz (bounded, deterministic) vs the pandas oracle
+# ---------------------------------------------------------------------------
+
+_RELS = ["KNOWS", "HAS_CREATOR", "LIKES", "LINK"]
+_KINDS = ["person", "post", "comment", "forum"]
+
+
+def _fuzz_graph(seed, n):
+    rng = random.Random(seed)
+    nodes = pd.DataFrame({
+        "id": list(range(n)),
+        "kind": [rng.choice(_KINDS) for _ in range(n)],
+        "val": [rng.randint(0, 20) for _ in range(n)],
+    })
+    m = rng.randint(n, n * 3)
+    edges = pd.DataFrame({
+        "s": [rng.randint(0, n - 1) for _ in range(m)],
+        "d": [rng.randint(0, n - 1) for _ in range(m)],
+        "type": [rng.choice(_RELS) for _ in range(m)],
+    })
+    return graphistry.nodes(nodes, "id").edges(edges, "s", "d")
+
+
+def _fuzz_query(rng, n):
+    x = rng.randint(0, n - 1)
+    r1, r2 = rng.choice(_RELS), rng.choice(_RELS)
+
+    def arrow(d, rel):
+        return {"->": f"-[:{rel}]->", "<-": f"<-[:{rel}]-"}.get(d, f"-[:{rel}]-")
+
+    d1, d2 = rng.choice(["-", "->", "<-"]), rng.choice(["-", "->", "<-"])
+    distinct = rng.choice(["DISTINCT ", ""])
+    where = rng.choice(["", f" WHERE a.val > {rng.randint(0, 20)}", f" WHERE a.kind = '{rng.choice(_KINDS)}'"])
+    tail = rng.choice(["", "", f" ORDER BY a.val LIMIT {rng.randint(1, 4)}"])
+    prefix = f"MATCH (p {{id:{x}}}){arrow(d1, r1)}(a){where} WITH {distinct}a{tail}"
+    b_filt = rng.choice(["", "", f" {{kind:'{rng.choice(_KINDS)}'}}"])
+    suffix_ret = rng.choice([
+        "RETURN b.id AS bid",
+        "RETURN b.id AS bid, b.kind AS bk",
+        "RETURN b",
+        "RETURN count(b) AS c",
+        "RETURN count(*) AS c",
+        "RETURN DISTINCT b.kind AS bk",
+    ])
+    return f"{prefix} MATCH (a){arrow(d2, r2)}(b{b_filt}) {suffix_ret}"
+
+
+def test_with_match_reentry_differential_fuzz():
+    """Bounded seeded fuzz: for every supported WITH->MATCH shape, polars must match the pandas
+    oracle exactly; unsupported shapes must decline (NotImplementedError). No silent-wrong."""
+    rng = random.Random(20260718)
+    agree = decline = 0
+    checked = 0
+    for _ in range(120):
+        g = _fuzz_graph(rng.randint(0, 10_000), rng.randint(4, 12))
+        q = _fuzz_query(rng, len(g._nodes))
+        try:
+            rpd = _to_pandas(g.gfql(q, engine="pandas")._nodes)
+        except Exception:
+            # pandas oracle itself declined at parse/lower — polars must not succeed here.
+            with pytest.raises(Exception):
+                g.gfql(q, engine="polars")
+            continue
+        try:
+            rpl = _to_pandas(g.gfql(q, engine="polars")._nodes)
+        except NotImplementedError:
+            decline += 1
+            continue
+        checked += 1
+        assert list(rpd.columns) == list(rpl.columns), f"cols differ for {q!r}"
+        if len(rpd) or len(rpl):
+            pd.testing.assert_frame_equal(_norm(rpd), _norm(rpl), check_dtype=False)
+        agree += 1
+    # Guard: the sweep must actually exercise the native re-entry path, not just declines.
+    assert checked > 20, f"fuzz did not exercise enough native shapes (checked={checked})"
diff --git a/graphistry/tests/test_engine_frame_helpers.py b/graphistry/tests/test_engine_frame_helpers.py
new file mode 100644
index 0000000000..5b7ea0aaf7
--- /dev/null
+++ b/graphistry/tests/test_engine_frame_helpers.py
@@ -0,0 +1,181 @@
+"""Unit tests for the engine-agnostic frame/series primitives in ``graphistry.Engine``.
+
+These helpers (moved out of the cypher reentry executor) dispatch across
+pandas/cuDF/polars, so both branches of each are exercised here. Polars cases are
+skipped where polars is not installed (the pandas-only CI lanes); the polars lane
+(``bin/test-polars.sh``) runs this file with polars present.
+"""
+import pandas as pd
+import pytest
+
+from graphistry.Engine import (
+    assign_constant_columns,
+    drop_columns,
+    frame_filter,
+    is_series_like,
+    ordered_left_join,
+    row_as_mapping,
+    series_filter,
+    series_not_null_mask,
+    series_to_pylist,
+)
+
+try:
+    import polars as pl
+    _HAS_POLARS = True
+except ImportError:
+    _HAS_POLARS = False
+
+polars_only = pytest.mark.skipif(not _HAS_POLARS, reason="polars not installed")
+
+
+# --- pandas branches (run on every lane) ---------------------------------
+
+def test_is_series_like_pandas() -> None:
+    assert is_series_like(pd.Series([1, 2])) is True
+    assert is_series_like(object()) is False
+
+
+def test_series_not_null_mask_pandas() -> None:
+    assert list(series_not_null_mask(pd.Series([1, None, 3]))) == [True, False, True]
+
+
+def test_series_filter_pandas_drops_index() -> None:
+    s = pd.Series([10, 20, 30])
+    out = series_filter(s, pd.Series([True, False, True]))
+    assert list(out) == [10, 30]
+    assert list(out.index) == [0, 1]  # index reset
+
+
+def test_frame_filter_pandas_drops_index() -> None:
+    df = pd.DataFrame({"a": [1, 2, 3]})
+    out = frame_filter(df, pd.Series([False, True, True]))
+    assert out["a"].tolist() == [2, 3]
+    assert list(out.index) == [0, 1]
+
+
+def test_ordered_left_join_pandas_preserves_left_order() -> None:
+    left = pd.DataFrame({"k": [3, 1, 2]})
+    right = pd.DataFrame({"k": [1, 2, 3], "v": [10, 20, 30]})
+    out = ordered_left_join(left, right, on="k")
+    assert out["v"].tolist() == [30, 10, 20]
+
+
+def test_row_as_mapping_pandas() -> None:
+    df = pd.DataFrame({"a": [1, 2], "b": [3, 4]})
+    assert dict(row_as_mapping(df, 1)) == {"a": 2, "b": 4}
+
+
+def test_assign_constant_columns_pandas_and_empty() -> None:
+    df = pd.DataFrame({"a": [1, 2]})
+    assert assign_constant_columns(df, {"c": 9})["c"].tolist() == [9, 9]
+    # empty values short-circuits and returns the frame unchanged
+    same = assign_constant_columns(df, {})
+    assert same is df
+
+
+def test_drop_columns_pandas() -> None:
+    df = pd.DataFrame({"a": [1], "b": [2], "c": [3]})
+    assert list(drop_columns(df, ["b", "c"]).columns) == ["a"]
+
+
+def test_series_to_pylist_pandas_and_fallbacks() -> None:
+    # pandas Series has no to_arrow -> tolist branch
+    assert series_to_pylist(pd.Series([1, 2, 3])) == [1, 2, 3]
+
+    # object exposing to_pandas but not to_arrow (cuDF-shaped) -> to_pandas branch
+    class _ToPandasOnly:
+        def to_pandas(self) -> pd.Series:
+            return pd.Series([7, 8])
+
+    assert series_to_pylist(_ToPandasOnly()) == [7, 8]
+
+    # to_arrow that raises -> falls through the except to the next branch
+    class _ArrowRaises:
+        def to_arrow(self):  # type: ignore[no-untyped-def]
+            raise RuntimeError("boom")
+
+        def tolist(self):  # type: ignore[no-untyped-def]
+            return [5, 6]
+
+    assert series_to_pylist(_ArrowRaises()) == [5, 6]
+
+    # to_pandas that raises -> falls through its except to the tolist branch
+    class _PandasRaises:
+        def to_pandas(self):  # type: ignore[no-untyped-def]
+            raise RuntimeError("boom")
+
+        def tolist(self):  # type: ignore[no-untyped-def]
+            return [3, 4]
+
+    assert series_to_pylist(_PandasRaises()) == [3, 4]
+
+    # tolist that raises -> falls through its except to the final list() fallback
+    class _TolistRaisesIterable:
+        def tolist(self):  # type: ignore[no-untyped-def]
+            raise RuntimeError("boom")
+
+        def __iter__(self):  # type: ignore[no-untyped-def]
+            return iter([9, 10])
+
+    assert series_to_pylist(_TolistRaisesIterable()) == [9, 10]
+
+    # no arrow/pandas/tolist -> final list() fallback over an iterable
+    assert series_to_pylist([1, 2]) == [1, 2]
+
+
+# --- polars branches (polars lane only) ----------------------------------
+
+@polars_only
+def test_is_series_like_polars() -> None:
+    assert is_series_like(pl.Series([1, 2])) is True
+
+
+@polars_only
+def test_series_not_null_mask_polars() -> None:
+    assert series_not_null_mask(pl.Series([1, None, 3])).to_list() == [True, False, True]
+
+
+@polars_only
+def test_series_filter_polars() -> None:
+    out = series_filter(pl.Series([10, 20, 30]), pl.Series([True, False, True]))
+    assert out.to_list() == [10, 30]
+
+
+@polars_only
+def test_frame_filter_polars() -> None:
+    df = pl.DataFrame({"a": [1, 2, 3]})
+    out = frame_filter(df, pl.Series([False, True, True]))
+    assert out["a"].to_list() == [2, 3]
+
+
+@polars_only
+def test_ordered_left_join_polars_preserves_order_and_coerces_right() -> None:
+    left = pl.DataFrame({"k": [3, 1, 2]})
+    # right is pandas -> must be coerced to polars before the join
+    right = pd.DataFrame({"k": [1, 2, 3], "v": [10, 20, 30]})
+    out = ordered_left_join(left, right, on="k")
+    assert out["v"].to_list() == [30, 10, 20]
+
+
+@polars_only
+def test_row_as_mapping_polars() -> None:
+    df = pl.DataFrame({"a": [1, 2], "b": [3, 4]})
+    assert dict(row_as_mapping(df, 1)) == {"a": 2, "b": 4}
+
+
+@polars_only
+def test_assign_constant_columns_polars() -> None:
+    df = pl.DataFrame({"a": [1, 2]})
+    assert assign_constant_columns(df, {"c": 9})["c"].to_list() == [9, 9]
+
+
+@polars_only
+def test_drop_columns_polars() -> None:
+    df = pl.DataFrame({"a": [1], "b": [2], "c": [3]})
+    assert list(drop_columns(df, ["b", "c"]).columns) == ["a"]
+
+
+@polars_only
+def test_series_to_pylist_polars_to_arrow() -> None:
+    assert series_to_pylist(pl.Series([1, 2, 3])) == [1, 2, 3]