diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java
index b404ceaf41dc..23819a0a69c4 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java
@@ -32,6 +32,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
@@ -70,6 +72,10 @@
 })
 public class TestDataflowRunner extends PipelineRunner<DataflowPipelineJob> {
   private static final String TENTATIVE_COUNTER = "tentative";
+
+  @VisibleForTesting
+  static final String DAILY_STAGING_DIR = "staging-" + LocalDate.now(ZoneOffset.UTC);
+
   static final String MAX_CONCURRENT_STANDALONE_JOBS_PROPERTY =
       "beam.dataflow.maxConcurrentStandaloneJobs";
   private static final int DEFAULT_MAX_CONCURRENT_STANDALONE_JOBS = 12;
@@ -148,9 +154,20 @@ public static TestDataflowRunner fromOptions(PipelineOptions options) {
     dataflowOptions.setTempLocation(tempLocation);
     String defaultPerJobStagingLocation =
         FileSystems.matchNewDirectory(tempLocation, "staging").toString();
-    if (defaultPerJobStagingLocation.equals(dataflowOptions.getStagingLocation())) {
+    String unversionedSharedStagingLocation =
+        FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), "staging").toString();
+    String rawStagingLocation = dataflowOptions.getStagingLocation();
+    String configuredStagingLocation =
+        rawStagingLocation != null
+            ? FileSystems.matchNewDirectory(rawStagingLocation).toString()
+            : null;
+    // Scope shared test staging directories by UTC date so tests running on the same day reuse
+    // staged JARs without referencing objects near the 7-day GCS lifecycle deletion threshold.
+    if (defaultPerJobStagingLocation.equals(configuredStagingLocation)
+        || unversionedSharedStagingLocation.equals(configuredStagingLocation)) {
       dataflowOptions.setStagingLocation(
-          FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), "staging").toString());
+          FileSystems.matchNewDirectory(dataflowOptions.getTempRoot(), DAILY_STAGING_DIR)
+              .toString());
     }
 
     return new TestDataflowRunner(
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java
index 6e1d9f627ffa..0fb4b0eb6f3e 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/TestDataflowRunnerTest.java
@@ -119,7 +119,19 @@ public void testFromOptionsUsesSharedStagingLocationUnderTempRoot() {
     TestDataflowRunner.fromOptions(options);
     assertEquals("gs://test/test-job-1/output/results", options.getTempLocation());
     assertEquals("gs://test/test-job-1/output/results", options.getGcpTempLocation());
-    assertEquals("gs://test/staging/", options.getStagingLocation());
+    assertEquals(
+        "gs://test/" + TestDataflowRunner.DAILY_STAGING_DIR + "/", options.getStagingLocation());
+  }
+
+  @Test
+  public void testFromOptionsScopesUnversionedTempRootStagingLocationByDate() {
+    options.setJobName("test-job-unversioned-staging");
+    options.setStagingLocation("gs://test/staging");
+    TestDataflowRunner.fromOptions(options);
+    assertEquals(
+        "gs://test/test-job-unversioned-staging/output/results", options.getTempLocation());
+    assertEquals(
+        "gs://test/" + TestDataflowRunner.DAILY_STAGING_DIR + "/", options.getStagingLocation());
   }
 
   @Test