@@ -6,7 +6,7 @@
 """
 
 from datetime import timedelta
-from typing import Any, Callable, Dict, List, Tuple, Union
+from typing import Any, Callable, Dict, List, Set, Tuple, Union
 
 import pandas as pd
 
@@ -61,6 +61,9 @@ def apply_sawtooth_window_tiling(
     # Step 2: Group by entity keys + hop interval and aggregate
     agg_dict: Dict[str, Tuple[str, Union[str, Callable[[Any], Any]]]] = {}
     ir_metadata_dict = {}
+    # max/min cannot be undone by subtraction, so these columns keep per-hop
+    # values instead of cumulative ones (reduced over the window later).
+    per_hop_columns: Set[str] = set()
 
     for agg in aggregations:
         feature_name = agg.resolved_name(window_size)
@@ -76,8 +79,10 @@ def apply_sawtooth_window_tiling(
                 agg_dict[f"_tail_{feature_name}"] = (agg.column, "count")
             elif agg.function == "max":
                 agg_dict[f"_tail_{feature_name}"] = (agg.column, "max")
+                per_hop_columns.add(f"_tail_{feature_name}")
             elif agg.function == "min":
                 agg_dict[f"_tail_{feature_name}"] = (agg.column, "min")
+                per_hop_columns.add(f"_tail_{feature_name}")
 
         elif metadata.type in ("holistic", "avg", "std", "var"):
             # Holistic aggregations: compute IRs
@@ -141,15 +146,11 @@ def apply_sawtooth_window_tiling(
                 group_df, on=group_by_keys + ["_hop_interval"], how="left"
             )
 
-            # Fill NaN with 0 for IR columns
+            # Fill NaN with 0 and compute cumulative sum for IR columns
+            # (empty hops stay NaN for per-hop max/min columns)
             for col in ir_columns:
-                if col in merged.columns:
-                    merged[col] = merged[col].fillna(0)
-
-            # Compute cumulative sum for IR columns
-            for col in ir_columns:
-                if col in merged.columns:
-                    merged[col] = merged[col].cumsum()
+                if col in merged.columns and col not in per_hop_columns:
+                    merged[col] = merged[col].fillna(0).cumsum()
 
             cumulative_results.append(merged)
 
 
@@ -34,6 +34,9 @@ def convert_cumulative_to_windowed(
     1. Subtracts intermediate representation (IR) components
     2. Recomputes the final value from windowed IRs
 
+    max and min cannot be undone by subtraction, so their tiles hold per-hop
+    values and the windowed value is the max/min over the hops in the window.
+
     Args:
         tiles_df: DataFrame with cumulative tiles from orchestrator.
                   Must contain: entity_keys, _tile_end, IR columns, feature columns
@@ -88,11 +91,17 @@ def convert_cumulative_to_windowed(
                         current_val = float(row[feature_name])
 
                         if agg.function.lower() in ("max", "min"):
-                            if has_prev_tile and feature_name in prev_tile.index:
-                                prev_val = float(prev_tile[feature_name])
-                                windowed_row[feature_name] = current_val
+                            # max/min cannot be subtracted: the tiles hold
+                            # per-hop values, so reduce the hops in the window
+                            in_window = group_df.loc[
+                                (group_df["_tile_end"] > window_start)
+                                & (group_df["_tile_end"] <= tile_end),
+                                feature_name,
+                            ]
+                            if agg.function.lower() == "max":
+                                windowed_row[feature_name] = float(in_window.max())
                             else:
-                                windowed_row[feature_name] = current_val
+                                windowed_row[feature_name] = float(in_window.min())
                         else:
                             # For sum and count: subtract previous from current
                             if has_prev_tile and feature_name in prev_tile.index:
 
@@ -190,3 +190,52 @@ def test_tiled_holistic_aggregation_matches_pandas(function, expected_fn):
             & (df["event_timestamp"] < tile_end)
         ]["trips"]
         assert row[feature_name] == pytest.approx(expected_fn(in_window), nan_ok=True)
+
+
+@pytest.mark.parametrize("function", ["max", "min"])
+def test_tiled_max_min_aggregation_matches_pandas(function):
+    """Tiled max/min only consider the hops inside the sliding window."""
+    from feast.aggregation.tiling.orchestrator import apply_sawtooth_window_tiling
+    from feast.aggregation.tiling.tile_subtraction import (
+        convert_cumulative_to_windowed,
+    )
+
+    start = pd.Timestamp("2024-01-01")
+    # No events between minutes 20 and 35, so two hops are empty.
+    minutes = [0, 5, 10, 15, 20, 35, 40, 45]
+    values = [-1.0, -2.0, 30.0, -10.0, 4.0, -7.0, 5.0, -8.0]
+    df = pd.DataFrame(
+        {
+            "driver_id": [1] * len(values),
+            "event_timestamp": [start + timedelta(minutes=m) for m in minutes],
+            "trips": values,
+        }
+    )
+    window = timedelta(minutes=20)
+    agg = Aggregation(column="trips", function=function, time_window=window)
+
+    tiles = apply_sawtooth_window_tiling(
+        df=df.copy(),
+        aggregations=[agg],
+        group_by_keys=["driver_id"],
+        timestamp_col="event_timestamp",
+        window_size=window,
+        hop_size=timedelta(minutes=5),
+    )
+    result = convert_cumulative_to_windowed(
+        tiles_df=tiles,
+        entity_keys=["driver_id"],
+        timestamp_col="event_timestamp",
+        window_size=window,
+        aggregations=[agg],
+    )
+
+    feature_name = agg.resolved_name(window)
+    assert len(result) == 10
+    for _, row in result.iterrows():
+        tile_end = row["event_timestamp"]
+        in_window = df[
+            (df["event_timestamp"] >= tile_end - window)
+            & (df["event_timestamp"] < tile_end)
+        ]["trips"]
+        assert row[feature_name] == getattr(in_window, function)()