diff --git a/test/restart/mzcompose.py b/test/restart/mzcompose.py
index c8e4d4ab17..872619992d 100644
--- a/test/restart/mzcompose.py
+++ b/test/restart/mzcompose.py
@@ -976,6 +976,107 @@ def workflow_user_id_no_reuse_after_restart(c: Composition) -> None:
     c.sql("DROP TABLE idreuse_t1")


+def workflow_arrangement_sizes_hydration_gate(c: Composition) -> None:
+    """Regression test: arrangement-sizes hydration gate must hold across restart.
+
+    The race between the introspection-subscribe install and the
+    arrangement-size collection timer is narrow, so we repeat the
+    kill/up/observe cycle several times to keep the test reliable.
+    """
+
+    num_replicas = 2
+    arranged_names = tuple(f"mv{i}" for i in range(1, 11)) + tuple(
+        f"idx{i}" for i in range(1, 11)
+    )
+    expected_count = len(arranged_names) * num_replicas
+    name_filter = "(" + ", ".join(f"'{n}'" for n in arranged_names) + ")"
+    restart_rounds = 5
+
+    c.down(destroy_volumes=True)
+
+    with c.override(
+        Materialized(
+            additional_system_parameter_defaults={
+                "arrangement_size_collection_interval": "500ms",
+            },
+            sanity_restart=False,
+        )
+    ):
+        c.up("materialized")
+
+        mv_ddls = "\n".join(
+            f"CREATE MATERIALIZED VIEW mv{i} IN CLUSTER test AS "
+            f"SELECT a % {i + 1} AS k, SUM(b) AS s FROM t GROUP BY a % {i + 1};"
+            f"CREATE INDEX idx{i} IN CLUSTER test ON mv{i} (k);"
+            for i in range(1, 11)
+        )
+        c.sql(
+            dedent(
+                f"""
+                CREATE CLUSTER test SIZE 'scale=1,workers=1', REPLICATION FACTOR {num_replicas};
+                CREATE TABLE t (a int, b int);
+                INSERT INTO t SELECT g, g % 100 FROM generate_series(1, 100000) g;
+                {mv_ddls}
+                """
+            )
+        )
+
+        def wait_for_full_sample() -> str:
+            deadline = time.time() + 120
+            while time.time() < deadline:
+                rows = c.sql_query(
+                    f"""
+                    SELECT max(collection_timestamp)::text FROM (
+                        SELECT collection_timestamp
+                          FROM mz_internal.mz_object_arrangement_size_history h
+                          JOIN mz_objects o ON o.id = h.object_id
+                         WHERE o.name IN {name_filter}
+                         GROUP BY collection_timestamp
+                        HAVING count(*) = {expected_count}
+                    ) AS s
+                    """
+                )
+                if rows and rows[0][0] is not None:
+                    return rows[0][0]
+                time.sleep(0.5)
+            raise UIError(
+                f"timed out waiting for a sample with {expected_count} rows"
+            )
+
+        all_partials: list[tuple[int, str, int]] = []
+        for round_idx in range(restart_rounds):
+            pre_
restart_ts = wait_for_full_sample()
+
+            c.kill("materialized")
+            c.up("materialized")
+
+            time.sleep(8)
+
+            samples = c.sql_query(
+                f"""
+                SELECT collection_timestamp::text, count(*)
+                  FROM mz_internal.mz_object_arrangement_size_history h
+                  JOIN mz_objects o ON o.id = h.object_id
+                 WHERE o.name IN {name_filter}
+                   AND collection_timestamp > '{pre_restart_ts}'::timestamptz
+                 GROUP BY collection_timestamp
+                 ORDER BY collection_timestamp
+                """
+            )
+            assert (
+                samples
+            ), f"round {round_idx}: no post-restart collection_timestamps found"
+
+            for ts, cnt in samples:
+                if cnt != expected_count:
+                    all_partials.append((round_idx, ts, cnt))
+
+        assert not all_partials, (
+            f"{len(all_partials)} post-restart samples (across {restart_rounds} "
+            f"rounds) had row counts != {expected_count}: {all_partials}"
+        )
+
+
 def workflow_default(c: Composition) -> None:
     def process(name: str) -> None:
         if name == "default":