diff --git a/.github/workflows/pr-tests.yml b/.github/workflows/pr-tests.yml
index 30e583619..74f5c724b 100644
--- a/.github/workflows/pr-tests.yml
+++ b/.github/workflows/pr-tests.yml
@@ -83,13 +83,67 @@ jobs:
             | sed -u 's/^/resource-monitor out-of-memory /' ) &
           trap 'kill %1 %2 2>/dev/null || true' EXIT
 
+          # Fail fast if the target dies mid-run. If a server crashes, every
+          # remaining test otherwise blocks on connection timeouts against a
+          # dead target one by one, and the job burns the full 6h runner limit
+          # before failing. This watchdog polls the profile's containers and, if
+          # any expected service stops running, terminates the test run so the
+          # job fails in about a minute instead of hours.
+          PROFILE='${{ matrix.target.profile }}'
+          watch_target() {
+            set +e
+            local pid=$1 expected running low=0
+            expected=$(docker compose -f dev/compose.yaml --profile "$PROFILE" config --services | grep -c .)
+            # Sanity-check the probe once. The stack is already up (compose
+            # --wait), so every service should read as running; if it does not,
+            # the ps query is unreliable on this runner and the watchdog steps
+            # aside rather than risk a false abort.
+            if [ "$(docker compose -f dev/compose.yaml --profile "$PROFILE" ps --status running -q 2>/dev/null | grep -c .)" -lt "$expected" ]; then
+              echo "resource-monitor target-watchdog disabled: could not read a healthy baseline"
+              return
+            fi
+            while kill -0 "$pid" 2>/dev/null; do
+              running=$(docker compose -f dev/compose.yaml --profile "$PROFILE" ps --status running -q 2>/dev/null | grep -c .)
+              # Require two consecutive low readings before aborting. A single
+              # low count can be a transient docker-CLI hiccup on this
+              # resource-heavy runner (a failed query reads as 0); a real crash
+              # stays low, so demanding two polls in a row rules out a false
+              # abort while adding only one poll interval to a genuine one.
+              if [ "$running" -lt "$expected" ]; then
+                low=$((low + 1))
+                if [ "$low" -ge 2 ]; then
+                  echo "::error::A $PROFILE target container exited mid-run (likely a server crash). Aborting the test run so the job fails fast instead of timing out every remaining test against a dead target."
+                  docker compose -f dev/compose.yaml --profile "$PROFILE" ps -a
+                  kill -TERM "$pid" 2>/dev/null; sleep 10; kill -KILL "$pid" 2>/dev/null
+                  return
+                fi
+              else
+                low=0
+              fi
+              sleep 15
+            done
+          }
+
+          # --timeout caps any single hung test as a backstop independent of the
+          # watchdog: a wedged query fails its own test (default signal method,
+          # as the crash-test job uses) instead of stalling a worker forever.
           pytest documentdb_tests/compatibility/tests \
             --connection-string "${{ matrix.target.connection_string }}" \
             --engine-name "${{ matrix.target.engine }}" \
             -n auto \
             -v \
+            --timeout=120 \
             --json-report --json-report-file=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-report.json \
-            --junitxml=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-results.xml
+            --junitxml=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-results.xml &
+          pytest_pid=$!
+          watch_target "$pytest_pid" &
+          watch_pid=$!
+          set +e
+          wait "$pytest_pid"
+          status=$?
+          set -e
+          kill "$watch_pid" 2>/dev/null || true
+          exit "$status"
 
       - name: Dump container logs
         if: always()
diff --git a/dev/compose.yaml b/dev/compose.yaml
index 60485a08f..e2c2fc80d 100644
--- a/dev/compose.yaml
+++ b/dev/compose.yaml
@@ -37,6 +37,23 @@
 # cache keeps the combined footprint within the VM. There is intentionally no
 # per-container mem_limit: that caps total process memory (not just cache) and a
 # transient working-set spike then kills mongod even when the VM has room.
+#
+# Open files: each mongod raises its open-file limit (ulimits.nofile). The suite
+# creates and drops tens of thousands of collections in a run, and WiredTiger
+# holds an open file descriptor per table (collection + indexes) in its handle
+# cache, closing idle ones only lazily. On the replica set a dropped collection
+# also becomes a drop-pending ident whose files linger until the drop timestamp
+# is majority-committed and swept, so descriptors accumulate faster than they
+# are reclaimed. WiredTiger also keeps a file handle open per table it has
+# touched and closes idle ones only very lazily, so the descriptor count tracks
+# the number of collections created over the run, not the number live at once.
+# The replica set peaks near 60k open files partway through the suite; when that
+# reaches the ceiling mongod runs out of descriptors mid-run, which surfaces
+# inside WiredTiger as a WT_PANIC and crashes the server (EMFILE -> fassert).
+# 64000 was not enough headroom (the suite crashed at ~31k collections against
+# it), so this is set to 1048576 -- the limit MongoDB's own packaging ships and
+# well above the suite's peak. Standalone reclaims dropped files immediately so
+# it never approaches the ceiling, but the limit is raised on both for parity.
 
 services:
   # mongo-standalone: a single standalone server.
@@ -44,6 +61,12 @@ services:
     image: mongo:8.2.4
     profiles: ["mongo-standalone", "all"]
     command: ["--wiredTigerCacheSizeGB", "1.5"]
+    # Raise the open-file limit above the container default so the suite's
+    # collection churn cannot exhaust mongod's descriptors (see header note).
+    ulimits:
+      nofile:
+        soft: 1048576
+        hard: 1048576
     ports:
       - "27017:27017"
     healthcheck:
@@ -80,6 +103,14 @@ services:
       - "skipAuthenticationToMongot=true"
       - "--setParameter"
       - "skipAuthenticationToSearchIndexManagementServer=true"
+    # Raise the open-file limit well above the peak. This matters most here: the
+    # replica set defers dropped-collection files as drop-pending idents, so
+    # descriptors accumulate under churn and exhaust a lower limit mid-run,
+    # crashing mongod with a WT_PANIC (see header note).
+    ulimits:
+      nofile:
+        soft: 1048576
+        hard: 1048576
     ports:
       - "27018:27017"
     healthcheck:
diff --git a/documentdb_tests/compatibility/tests/core/operator/expressions/test_expressions_combination_variables.py b/documentdb_tests/compatibility/tests/core/operator/expressions/test_expressions_combination_variables.py
index ff55a776f..43882418e 100644
--- a/documentdb_tests/compatibility/tests/core/operator/expressions/test_expressions_combination_variables.py
+++ b/documentdb_tests/compatibility/tests/core/operator/expressions/test_expressions_combination_variables.py
@@ -160,12 +160,12 @@ def test_let_nested_combinations(collection, test):
 # ---------------------------------------------------------------------------
 def test_let_two_lets_same_projection(collection):
     """Test two separate $let expressions in same projection with same variable name."""
+    collection.insert_one({})
     result = execute_command(
         collection,
         {
-            "aggregate": 1,
+            "aggregate": collection.name,
             "pipeline": [
-                {"$documents": [{}]},
                 {
                     "$project": {
                         "_id": 0,
@@ -248,12 +248,12 @@ def test_let_across_multiple_documents(collection):
 
 def test_let_error_cross_let_variable_ref(collection):
     """Test $let where variable defined in one $let is referenced in sibling $let."""
+    collection.insert_one({})
     result = execute_command(
         collection,
         {
-            "aggregate": 1,
+            "aggregate": collection.name,
             "pipeline": [
-                {"$documents": [{}]},
                 {
                     "$project": {
                         "_id": 0,
diff --git a/documentdb_tests/compatibility/tests/core/operator/expressions/utils/utils.py b/documentdb_tests/compatibility/tests/core/operator/expressions/utils/utils.py
index ec6d21c75..7918d55e6 100644
--- a/documentdb_tests/compatibility/tests/core/operator/expressions/utils/utils.py
+++ b/documentdb_tests/compatibility/tests/core/operator/expressions/utils/utils.py
@@ -39,6 +39,21 @@ def execute_project(collection, project):
     """
     Execute a projection with literal input values.
 
+    Evaluates the projection against a single empty document. The document is
+    inserted into the collection and the pipeline runs over that collection,
+    rather than synthesizing the row with a ``$documents`` stage. This keeps the
+    helper free of any dependency on ``$documents`` support while producing the
+    same single-row input the projection sees.
+
+    Note: the inserted document carries an auto-generated ``_id``, so the input
+    row is ``{_id: <ObjectId>}`` rather than the field-less row ``$documents:
+    [{}]`` produces. Literal expressions and references to any other (missing)
+    field behave identically, and the output projection excludes ``_id``; but an
+    expression that reads ``$_id``, ``$$ROOT``, or ``$$CURRENT`` sees that id and
+    will diverge. Callers that need a truly field-less input (e.g. ``$$ROOT`` must
+    be ``{}``) or exactly one row over a pre-populated collection must shape their
+    own pipeline instead of using this helper.
+
     Args:
         collection: MongoDB collection object
         project: Fields to project. Do not include _id; the function always
@@ -51,12 +66,12 @@ def execute_project(collection, project):
         >>> execute_project(collection, {"sum": {"$add": [1, 2]}})
         # Returns result with {"sum": 3} in firstBatch
     """
+    collection.insert_one({})
     return execute_command(
         collection,
         {
-            "aggregate": 1,
+            "aggregate": collection.name,
             "pipeline": [
-                {"$documents": [{}]},
                 {"$project": {**materialize(project), "_id": 0}},
             ],
             "cursor": {},
@@ -100,10 +115,24 @@ def execute_project_with_insert(collection, document, project):
 
 def execute_expression(collection, expression):
     """
-    Execute an aggregation expression using $documents stage.
-
-    Evaluates an expression against an empty document using the $documents
-    stage. Useful for testing expressions with literal values.
+    Execute an aggregation expression against a single empty document.
+
+    Evaluates an expression against an empty document. The document is inserted
+    into the collection and the pipeline runs over that collection, rather than
+    synthesizing the row with a ``$documents`` stage. This keeps the helper free
+    of any dependency on ``$documents`` support while producing the same
+    single-row input the expression is evaluated against. Useful for testing
+    expressions with literal values; references to fields other than ``_id``
+    resolve to missing, just as they would against a ``$documents: [{}]`` row.
+
+    Note: the inserted document carries an auto-generated ``_id``, so the input
+    row is ``{_id: <ObjectId>}`` rather than the field-less row ``$documents:
+    [{}]`` produces. Literal expressions and references to any other (missing)
+    field are unaffected, and the output projection excludes ``_id``; but an
+    expression that reads ``$_id``, ``$$ROOT``, or ``$$CURRENT`` sees that id and
+    will diverge. Callers that need a truly field-less input (e.g. ``$$ROOT`` must
+    be ``{}``) or exactly one row over a pre-populated collection must shape their
+    own pipeline instead of using this helper.
 
     Args:
         collection: MongoDB collection object
@@ -117,12 +146,12 @@ def execute_expression(collection, expression):
         >>> execute_expression(collection, {"$add": [1, 2]})
         # Returns result with {"result": 3} in firstBatch
     """
+    collection.insert_one({})
     return execute_command(
         collection,
         {
-            "aggregate": 1,
+            "aggregate": collection.name,
             "pipeline": [
-                {"$documents": [{}]},
                 {"$project": {"_id": 0, "result": expression}},
             ],
             "cursor": {},
diff --git a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/cluster_time/test_cluster_time_clock_progression.py b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/cluster_time/test_cluster_time_clock_progression.py
index f49a9c390..1ac073cb1 100644
--- a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/cluster_time/test_cluster_time_clock_progression.py
+++ b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/cluster_time/test_cluster_time_clock_progression.py
@@ -429,7 +429,17 @@ def test_cluster_time_advances_after_a_write(collection):
     earlier = _read_cluster_time(collection)
     collection.insert_one({"_id": "advance"})
 
-    result = execute_expression(collection, {"$gt": ["$$CLUSTER_TIME", earlier]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$gt": ["$$CLUSTER_TIME", earlier]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -444,9 +454,17 @@ def test_cluster_time_monotonic_across_repeated_executions(collection):
         observed.append(_read_cluster_time(collection))
         collection.insert_one({"_id": i})
 
-    result = execute_expression(
+    non_decreasing = {"$eq": [{"$sortArray": {"input": observed, "sortBy": 1}}, observed]}
+    result = execute_command(
         collection,
-        {"$eq": [{"$sortArray": {"input": observed, "sortBy": 1}}, observed]},
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": non_decreasing}},
+            ],
+            "cursor": {},
+        },
     )
     assert_expression_result(
         result,
@@ -462,7 +480,18 @@ def test_cluster_time_not_frozen_across_repeated_executions(collection):
         observed.append(_read_cluster_time(collection))
         collection.insert_one({"_id": i})
 
-    result = execute_expression(collection, {"$gt": [{"$size": {"$setUnion": [observed]}}, 1]})
+    saw_more_than_one_value = {"$gt": [{"$size": {"$setUnion": [observed]}}, 1]}
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": saw_more_than_one_value}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -523,7 +552,17 @@ def test_cluster_time_is_not_later_than_a_following_write(collection):
     value = _read_cluster_time(collection)
     write = execute_command(collection, {"insert": collection.name, "documents": [{"_id": 1}]})
 
-    result = execute_expression(collection, {"$lte": [value, write["operationTime"]]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$lte": [value, write["operationTime"]]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -536,7 +575,17 @@ def test_cluster_time_is_not_earlier_than_a_preceding_write(collection):
     write = execute_command(collection, {"insert": collection.name, "documents": [{"_id": 1}]})
     write_time = write["operationTime"]
 
-    result = execute_expression(collection, {"$gte": ["$$CLUSTER_TIME", write_time]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$gte": ["$$CLUSTER_TIME", write_time]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -554,7 +603,17 @@ def test_cluster_time_is_not_earlier_than_after_cluster_time(collection):
         collection, {"readConcern": {"level": "majority", "afterClusterTime": after}}
     )
 
-    result = execute_expression(collection, {"$gte": [pipeline_value, after]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$gte": [pipeline_value, after]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -721,7 +780,17 @@ def test_cursor_value_survives_writes_landing_between_batches(collection):
 
     values = _drain_cursor(collection, first, batch_size=3, write_between=True)
 
-    result = execute_expression(collection, {"$size": {"$setUnion": [values]}})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$size": {"$setUnion": [values]}}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=1,
@@ -743,7 +812,17 @@ def test_writes_between_batches_do_advance_the_deployment_clock(collection):
     cursor_value = first["cursor"]["firstBatch"][0]["t"]
     _drain_cursor(collection, first, batch_size=3, write_between=True)
 
-    result = execute_expression(collection, {"$gt": ["$$CLUSTER_TIME", cursor_value]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$gt": ["$$CLUSTER_TIME", cursor_value]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -776,7 +855,17 @@ def test_idle_cursor_resumes_with_the_same_value(collection):
     )
     resumed_value = resumed["cursor"]["nextBatch"][0]["t"]
 
-    result = execute_expression(collection, {"$eq": [first_value, resumed_value]})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$eq": [first_value, resumed_value]}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=True,
@@ -809,7 +898,17 @@ def test_change_stream_resolves_the_variable_for_its_events(collection):
     )
     values = [doc["t"] for doc in batch["cursor"]["nextBatch"]]
 
-    result = execute_expression(collection, {"$type": {"$arrayElemAt": [values, 0]}})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$type": {"$arrayElemAt": [values, 0]}}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected="timestamp",
@@ -843,7 +942,17 @@ def test_change_stream_value_is_frozen_for_the_life_of_the_stream(collection):
         batch = execute_command(collection, {"getMore": cursor_id, "collection": collection.name})
         observed.extend(doc["t"] for doc in batch["cursor"]["nextBatch"])
 
-    result = execute_expression(collection, {"$size": {"$setUnion": [observed]}})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$size": {"$setUnion": [observed]}}}},
+            ],
+            "cursor": {},
+        },
+    )
     assert_expression_result(
         result,
         expected=1,
diff --git a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/now/test_now_core_semantics.py b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/now/test_now_core_semantics.py
index 8dc17e8ec..bf1db1b59 100644
--- a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/now/test_now_core_semantics.py
+++ b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/now/test_now_core_semantics.py
@@ -198,10 +198,23 @@ def test_now_identical_across_getmore_batches(collection):
         seen.extend(doc["t"] for doc in batch["cursor"]["nextBatch"])
         cursor_id = batch["cursor"]["id"]
 
-    result = execute_expression(collection, {"$size": {"$setUnion": [seen]}})
-    assert_expression_result(
+    # The collection is pre-populated (300 docs), so ``execute_expression`` — which
+    # aggregates over the whole collection — would emit one row per document. A
+    # ``$limit: 1`` reduces it to the single row this assertion expects.
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$limit": 1},
+                {"$project": {"_id": 0, "result": {"$size": {"$setUnion": [seen]}}}},
+            ],
+            "cursor": {},
+        },
+    )
+    assertSuccess(
         result,
-        expected=1,
+        [{"result": 1}],
         msg="$$NOW should be identical across every getMore batch of one cursor",
     )
 
diff --git a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/root/test_root_core_behavior.py b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/root/test_root_core_behavior.py
index 22ed192ed..38b93ab90 100644
--- a/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/root/test_root_core_behavior.py
+++ b/documentdb_tests/compatibility/tests/core/operator/expressions/variable/system_variables/root/test_root_core_behavior.py
@@ -17,10 +17,10 @@
 )
 from documentdb_tests.compatibility.tests.core.operator.expressions.utils.utils import (
     assert_expression_result,
-    execute_expression,
     execute_expression_with_insert,
 )
 from documentdb_tests.framework.assertions import assertSuccess
+from documentdb_tests.framework.executor import execute_command
 from documentdb_tests.framework.parametrize import pytest_params
 from documentdb_tests.framework.test_constants import DOUBLE_PRECISION_LOSS, INT64_MAX
 
@@ -122,27 +122,32 @@ def test_root_echoes_doc(collection, test):
 
 # Property [Empty Document]: $$ROOT is an empty object when the input document has
 # no fields.
-ROOT_EMPTY_DOCUMENT_TESTS: list[ExpressionTestCase] = [
-    ExpressionTestCase(
-        id="empty_document",
-        expression="$$ROOT",
-        doc=None,
-        expected={},
-        msg="$$ROOT should return an empty object when the input document has no fields",
-    ),
-]
-
-
-@pytest.mark.parametrize("test", pytest_params(ROOT_EMPTY_DOCUMENT_TESTS))
-def test_root_empty_document(collection, test):
+def test_root_empty_document(collection):
     """$$ROOT over a field-less input document.
 
-    ``doc=None`` selects execute_expression, which evaluates the expression over
-    a ``$documents: [{}]`` stage rather than inserting a document, since an
-    inserted document would always be given an ``_id``.
+    This case needs a truly field-less input row, so it cannot use the shared
+    ``execute_expression`` helper: that helper inserts a document (which always
+    carries an auto-generated ``_id``), which would make ``$$ROOT`` a one-field
+    document. Instead a document is inserted and ``$replaceWith: {$literal: {}}``
+    strips it back to a field-less row before ``$$ROOT`` is read.
     """
-    result = execute_expression(collection, test.expression)
-    assert_expression_result(result, expected=test.expected, msg=test.msg)
+    collection.insert_one({})
+    result = execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$replaceWith": {"$literal": {}}},
+                {"$project": {"_id": 0, "result": "$$ROOT"}},
+            ],
+            "cursor": {},
+        },
+    )
+    assertSuccess(
+        result,
+        [{"result": {}}],
+        msg="$$ROOT should return an empty object when the input document has no fields",
+    )
 
 
 # Property [Reported Type]: $$ROOT always reports BSON type "object".