diff --git a/.github/workflows/pr-tests.yml b/.github/workflows/pr-tests.yml
index 0bb7ae967..d4f394227 100644
--- a/.github/workflows/pr-tests.yml
+++ b/.github/workflows/pr-tests.yml
@@ -5,20 +5,38 @@ on:
     branches: [main]
 
 jobs:
-  test-mongodb:
-    name: Tests (MongoDB)
+  # Derive the test matrix from dev/compose.yaml so targets are declared in one
+  # place. Each entry has the target name, its compose profile, connection
+  # string, and engine name.
+  discover-targets:
+    name: Discover test targets
     runs-on: ubuntu-latest
+    outputs:
+      targets: ${{ steps.matrix.outputs.targets }}
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: pip install -r requirements.txt
 
-    services:
-      mongodb:
-        image: mongo:8.2.4
-        ports:
-          - 27017:27017
-        options: >-
-          --health-cmd "mongosh --eval 'db.runCommand({ ping: 1 })'"
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
+      - name: Build target matrix from compose
+        id: matrix
+        run: |
+          TARGETS=$(python -m documentdb_tests.framework.ci_matrix)
+          echo "targets=$TARGETS" >> "$GITHUB_OUTPUT"
+
+  test:
+    name: "Tests (${{ matrix.target.name }})"
+    needs: discover-targets
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        target: ${{ fromJson(needs.discover-targets.outputs.targets) }}
 
     steps:
       - uses: actions/checkout@v6
@@ -30,21 +48,24 @@ jobs:
       - name: Install dependencies
         run: pip install -r requirements.txt
 
-      - name: Run compatibility tests against MongoDB
+      - name: Start ${{ matrix.target.name }} target
+        run: docker compose -f dev/compose.yaml --profile ${{ matrix.target.profile }} up -d --wait
+
+      - name: Run compatibility tests against ${{ matrix.target.name }}
         run: |
           pytest documentdb_tests/compatibility/tests \
-            --connection-string "mongodb://localhost:27017" \
-            --engine-name mongodb \
+            --connection-string "${{ matrix.target.connection_string }}" \
+            --engine-name "${{ matrix.target.engine }}" \
             -n auto \
             -v \
-            --json-report --json-report-file=${{ github.workspace }}/.test-results/mongodb-report.json \
-            --junitxml=${{ github.workspace }}/.test-results/mongodb-results.xml
+            --json-report --json-report-file=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-report.json \
+            --junitxml=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-results.xml
 
       - name: Upload test results
         if: always()
         uses: actions/upload-artifact@v7
         with:
-          name: test-results-mongodb
+          name: test-results-${{ matrix.target.name }}
           include-hidden-files: true
           path: ${{ github.workspace }}/.test-results/
           if-no-files-found: warn
@@ -52,19 +73,28 @@ jobs:
       - name: Generate test summary
         if: always()
         run: |
-          if [ -f ${{ github.workspace }}/.test-results/mongodb-report.json ]; then
-            python -m documentdb_tests.compatibility.result_analyzer -i ${{ github.workspace }}/.test-results/mongodb-report.json -o ${{ github.workspace }}/.test-results/mongodb-analysis.txt -f text || true
-            echo "## MongoDB Test Results" >> $GITHUB_STEP_SUMMARY
+          REPORT=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-report.json
+          ANALYSIS=${{ github.workspace }}/.test-results/${{ matrix.target.name }}-analysis.txt
+          if [ -f "$REPORT" ]; then
+            python -m documentdb_tests.compatibility.result_analyzer -i "$REPORT" -o "$ANALYSIS" -f text || true
+            echo "## ${{ matrix.target.name }} Test Results" >> $GITHUB_STEP_SUMMARY
             echo '```' >> $GITHUB_STEP_SUMMARY
-            cat ${{ github.workspace }}/.test-results/mongodb-analysis.txt >> $GITHUB_STEP_SUMMARY
+            cat "$ANALYSIS" >> $GITHUB_STEP_SUMMARY
             echo '```' >> $GITHUB_STEP_SUMMARY
           fi
 
-  collect-xcrash:
+  # Crash tests kill the server, so each runs in isolation in its own job (one
+  # job per intended target/test pair, each starting its own server). A pair is
+  # "intended" when the test is not deselected against that target by its
+  # requires(...) markers, so a test that does not apply to a target's topology
+  # produces no job at all. Discovery brings each target up and collects against
+  # it, so the intended set (and the test ids) match what the run will see.
+  discover-xcrash:
     name: Collect crash tests
+    needs: discover-targets
     runs-on: ubuntu-latest
     outputs:
-      tests: ${{ steps.collect.outputs.tests }}
+      pairs: ${{ steps.collect.outputs.pairs }}
     steps:
       - uses: actions/checkout@v6
 
@@ -75,38 +105,60 @@ jobs:
       - name: Install dependencies
         run: pip install -r requirements.txt
 
-      - name: Collect engine_xcrash test IDs
+      - name: Enumerate intended target/test pairs
         id: collect
+        env:
+          TARGETS: ${{ needs.discover-targets.outputs.targets }}
         run: |
-          TESTS=$(pytest documentdb_tests/compatibility/tests \
-            --connection-string mongodb://localhost:27017 \
-            --engine-name mongodb \
-            --collect-only -m engine_xcrash 2>/dev/null \
-            | grep '<Function' \
-            | sed 's/.*<Function \(.*\)>/\1/' \
-            | jq -R -s -c 'split("\n") | map(select(length > 0))')
-          echo "tests=$TESTS" >> "$GITHUB_OUTPUT"
-
-  test-mongodb-xcrash:
-    name: "Crash Test: ${{ matrix.test }}"
-    needs: collect-xcrash
-    if: needs.collect-xcrash.outputs.tests != '[]'
+          # For each target, bring its server up and collect the crash tests
+          # that are not deselected by their requires(...) markers against that
+          # target (zero-config discovery resolves the live topology, exactly as
+          # the run will). Collection runs as its own command so a failure fails
+          # the job. The result is one matrix entry per (target, test) pair,
+          # carrying the full target. Each target is torn down before the next
+          # so only one topology is live at a time.
+          PAIRS='[]'
+          for row in $(echo "$TARGETS" | jq -c '.[]'); do
+            PROFILE=$(echo "$row" | jq -r .profile)
+            docker compose -f dev/compose.yaml --profile "$PROFILE" up -d --wait
+            # Bring a replica-set target up to a writable primary so topology
+            # detection classifies it correctly (collection does not initiate).
+            python -m documentdb_tests.framework.engine_registry
+            # Collect the crash tests not deselected against this target. No
+            # crash test applying to a target is a valid outcome: pytest then
+            # exits 5 (no tests collected), which is not an error here. Any
+            # other non-zero exit is a real collection failure and fails the job.
+            set +e
+            pytest documentdb_tests/compatibility/tests \
+              --collect-only -m engine_xcrash --run-crash-tests -q > collect.out
+            STATUS=$?
+            set -e
+            docker compose -f dev/compose.yaml --profile "$PROFILE" down
+            if [ "$STATUS" -ne 0 ] && [ "$STATUS" -ne 5 ]; then
+              cat collect.out
+              echo "::error::Collecting crash tests for $PROFILE failed (exit $STATUS)"
+              exit 1
+            fi
+            TESTS=$(grep '<Function' collect.out \
+              | sed 's/.*<Function \(.*\)>/\1/' \
+              | jq -R -s -c 'split("\n") | map(select(length > 0))')
+            PAIRS=$(jq -c \
+              --argjson target "$row" \
+              --argjson tests "$TESTS" \
+              '. + [$tests[] | { target: $target, test: . }]' \
+              <<< "$PAIRS")
+          done
+          echo "pairs=$PAIRS" >> "$GITHUB_OUTPUT"
+
+  test-xcrash:
+    name: "Crash Test: ${{ matrix.pair.target.name }} / ${{ matrix.pair.test }}"
+    needs: discover-xcrash
+    if: needs.discover-xcrash.outputs.pairs != '[]'
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        test: ${{ fromJson(needs.collect-xcrash.outputs.tests) }}
-
-    services:
-      mongodb:
-        image: mongo:8.2.4
-        ports:
-          - 27017:27017
-        options: >-
-          --health-cmd "mongosh --eval 'db.runCommand({ ping: 1 })'"
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
+        pair: ${{ fromJson(needs.discover-xcrash.outputs.pairs) }}
 
     steps:
       - uses: actions/checkout@v6
@@ -118,16 +170,35 @@ jobs:
       - name: Install dependencies
         run: pip install -r requirements.txt
 
-      - name: "Run: ${{ matrix.test }}"
+      - name: Start ${{ matrix.pair.target.name }} target
+        run: docker compose -f dev/compose.yaml --profile ${{ matrix.pair.target.profile }} up -d --wait
+
+      - name: "Run: ${{ matrix.pair.test }} against ${{ matrix.pair.target.name }}"
         run: |
-          # Run the test expecting it to crash the server (non-zero exit).
-          # If pytest exits 0, the test passed - meaning the bug is fixed
-          # and the engine_xcrash marker should be removed.
-          if pytest documentdb_tests/compatibility/tests \
-            -m engine_xcrash -k "${{ matrix.test }}" \
-            --connection-string "mongodb://localhost:27017" \
-            --engine-name mongodb-xcrash-probe \
-            --timeout=10 -v; then
-            echo "::error::Test passed unexpectedly - server bug may be fixed, remove engine_xcrash marker"
+          # The pair survived discovery, so the test is intended to run against
+          # this target. --run-crash-tests lets it execute (it is skipped by
+          # default); zero-config discovery resolves the same live target as in
+          # the discovery step, so the test id is stable and always selected.
+          #
+          # Outcomes:
+          #   - exit 5 (no tests selected): discovery and run disagree, a
+          #     harness bug, not a crash -> fail the job.
+          #   - exit 0 (the test ran and passed): the server no longer crashes,
+          #     the bug is fixed and the engine_xcrash marker should be removed
+          #     -> fail the job.
+          #   - any other non-zero exit: the server crashed (expected)
+          #     -> success.
+          set +e
+          pytest documentdb_tests/compatibility/tests \
+            -m engine_xcrash -k "${{ matrix.pair.test }}" \
+            --run-crash-tests \
+            --timeout=10 -v
+          STATUS=$?
+          set -e
+          if [ "$STATUS" -eq 5 ]; then
+            echo "::error::No tests selected for ${{ matrix.pair.test }} on ${{ matrix.pair.target.name }} - discovery and run disagree"
+            exit 1
+          elif [ "$STATUS" -eq 0 ]; then
+            echo "::error::Test passed unexpectedly on ${{ matrix.pair.target.name }} - server bug may be fixed, remove engine_xcrash marker"
             exit 1
           fi
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index a02fba7f3..e89f1664c 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -31,7 +31,7 @@ repos:
 
       - id: mypy
         name: mypy
-        entry: python -m mypy documentdb_tests/ --no-site-packages
+        entry: python -m mypy documentdb_tests/
         language: system
         pass_filenames: false
 
diff --git a/README.md b/README.md
index 33db07727..31bcf87d1 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,7 @@ This testing framework provides:
 ### Prerequisites
 
 - Python 3.9 or higher
-- Access to a DocumentDB or MongoDB instance
+- Docker (with Compose v2) to run database targets locally, or access to an existing instance
 - pip package manager
 
 ### Installation
@@ -31,21 +31,52 @@ cd functional-tests
 pip install -r requirements.txt
 ```
 
+### Database Targets
+
+Tests run against a target, identified by both its engine and deployment
+topology (e.g. `mongo-standalone`, `mongo-replset`). Topology is an
+engine-specific concept and does not map across engines, so each target is its
+own named environment. `dev/compose.yaml` is the single source of truth for
+these targets and is used by both local runs and CI, so local matches CI by
+construction.
+
+Bring up a target with its profile (each binds a distinct host port, so several
+can run at once):
+
+```bash
+# Standalone server
+docker compose -f dev/compose.yaml --profile mongo-standalone up -d --wait
+
+# Single-node replica set
+docker compose -f dev/compose.yaml --profile mongo-replset up -d --wait
+
+# Everything at once
+docker compose -f dev/compose.yaml --profile all up -d --wait
+```
+
+Tear down with the same profile and `down`. Then point pytest at the matching
+target (see Running Tests below).
+
 ### Running Tests
 
 #### Basic Usage
 
 ```bash
-# Run all tests against default localhost
+# Run all tests against every live target discovered from dev/compose.yaml
 pytest
 
-# Run against specific engine
-pytest --connection-string mongodb://localhost:27017 --engine-name documentdb
+# Run against the mongo-standalone target
+pytest --connection-string "mongodb://localhost:27017" --engine-name mongodb
 
-# Run with just connection string (engine-name defaults to "default")
-pytest --connection-string mongodb://localhost:27017
+# Run against the mongo-replset target
+pytest --connection-string "mongodb://localhost:27018/?directConnection=true" --engine-name mongodb
 ```
 
+With no `--connection-string`, the suite discovers the live targets from
+`dev/compose.yaml` and runs against each. When `--connection-string` is given it
+pins that single target, and `--engine-name` must name a known engine so the
+target's capabilities resolve the same way as for a discovered one.
+
 #### Filter by Tags
 
 ```bash
@@ -237,7 +268,25 @@ tests/
 - `smoke`: Quick smoke tests for feature detection
 - `slow`: Tests that take longer to execute
 - `no_parallel`: Tests that must run sequentially (e.g., tests that kill sessions/ops, modify server config, or drop all users/roles). Automatically deferred to Phase 2 when using `-n`.
-- `replica_set`: Tests that require a replica set topology (e.g., change streams, encryption, certain admin commands). Skipped by default in CI. To run locally, pass a replica set connection string: `pytest -m replica_set --connection-string "mongodb://localhost:27017/?directConnection=true"`
+
+### Capability Requirements
+
+Some behaviors are only available in certain deployment environments. A test
+declares the capabilities it needs with the `requires` marker, for example
+`@pytest.mark.requires(change_streams=True)` for a behavior that needs change
+streams, or `@pytest.mark.requires(change_streams=False)` for one that only
+applies where they are absent.
+
+A capability is a named fact about a target rather than a topology. Which
+capabilities a target has is determined by its engine and topology, and the full
+set of capabilities and how they map to each environment lives in
+`documentdb_tests/framework/preconditions.py`, which is the single source of
+truth. A test runs only against the targets whose capabilities match what it
+requires, and is otherwise skipped.
+
+When targets are discovered automatically (running `pytest` with no
+`--connection-string`), each test runs against every discovered target it
+applies to, so no manual selection is needed.
 
 ## Writing Tests
 
diff --git a/dev/compose.yaml b/dev/compose.yaml
new file mode 100644
index 000000000..4258cecfb
--- /dev/null
+++ b/dev/compose.yaml
@@ -0,0 +1,73 @@
+# Development database targets for the functional test suite.
+#
+# Each service is an engine-topology target the tests can run against. A target
+# names both the engine and its deployment topology (e.g. standalone, replset),
+# because topology is an engine-specific concept and does not map cleanly across
+# engines. Bring up a single target with its profile, or the whole stack with
+# the "all" profile, e.g.:
+#
+#   docker compose -f dev/compose.yaml --profile mongo-standalone up -d --wait
+#   docker compose -f dev/compose.yaml --profile mongo-replset    up -d --wait
+#   docker compose -f dev/compose.yaml --profile all              up -d --wait
+#
+# Each target binds a distinct host port (see the ports: mapping below) so
+# several can run at once locally; the same ports are used in CI for
+# consistency. See the README for the canonical connection string per target.
+#
+# This file is the single source of truth for these topologies: CI brings them
+# up from here rather than declaring containers separately, so local runs match
+# CI by construction. It is also the source of truth for the test harness: each
+# runnable target carries an `x-test-target` block that the test registry reads
+# to build its connection string and engine identity (see
+# documentdb_tests/framework/engine_registry.py). The host port comes from the
+# `ports:` mapping, so nothing is duplicated.
+#
+#   x-test-target:
+#     engine: <engine name used for engine_xfail / engine_xcrash markers>
+#     query: <optional connection-string query string, without leading '?'>
+#
+# A service with no `x-test-target` is not a test target and is ignored by the
+# registry.
+#
+# Memory: each mongod caps its WiredTiger cache (--wiredTigerCacheSizeGB). By
+# default a mongod sizes its cache to ~50% of the host/VM RAM; with several
+# targets running at once that double-counts memory and the whole suite (which
+# creates many collections across both targets) gradually fills both caches
+# until the VM is exhausted and a container is OOM-killed mid-run. Capping the
+# 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.
+
+services:
+  # mongo-standalone: a single standalone server.
+  mongo-standalone:
+    image: mongo:8.2.4
+    profiles: ["mongo-standalone", "all"]
+    command: ["--wiredTigerCacheSizeGB", "1.5"]
+    ports:
+      - "27017:27017"
+    healthcheck:
+      test: ["CMD", "mongosh", "--quiet", "--eval", "db.runCommand({ping:1}).ok"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+    x-test-target:
+      engine: mongodb
+
+  # mongo-replset: a single-node replica set. Provides capabilities the
+  # reference server gates behind a replica set (change streams, transactions,
+  # queryable encryption, cluster-wide admin). Connect with directConnection=true.
+  mongo-replset:
+    image: mongo:8.2.4
+    profiles: ["mongo-replset", "all"]
+    command: ["--replSet", "rs0", "--bind_ip_all", "--wiredTigerCacheSizeGB", "1.5"]
+    ports:
+      - "27018:27017"
+    healthcheck:
+      test: ["CMD", "mongosh", "--quiet", "--eval", "db.runCommand({ping:1}).ok"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+    x-test-target:
+      engine: mongodb
+      query: directConnection=true
diff --git a/docs/REVIEW.md b/docs/REVIEW.md
index 9d442f1d5..b8bfec176 100644
--- a/docs/REVIEW.md
+++ b/docs/REVIEW.md
@@ -59,7 +59,7 @@ Follow every rule in [`FOLDER_STRUCTURE.md`](testing/FOLDER_STRUCTURE.md). Addit
 - [ ] Shared test data in `utils/` modules
 - [ ] Type annotations on helper functions
 - [ ] Order-independent output uses `ignore_doc_order=True`
-- [ ] Tests that require a replica set are tagged `@pytest.mark.replica`
+- [ ] Tests that depend on an environment capability are gated with `@pytest.mark.requires(<capability>=<bool>)` (e.g. `requires(change_streams=True)`), not a topology marker
 - [ ] Tests that cannot run in parallel are tagged `@pytest.mark.no_parallel`
 - [ ] Tests where MongoDB itself fails are tagged with `engine_xfail` (not skipped)
 - [ ] Tests with non-deterministic output (e.g. `$rand`, `$sample`, server timestamps) assert structure/bounds, not exact values
diff --git a/docs/SMOKE_TEST_NOTE.md b/docs/SMOKE_TEST_NOTE.md
index f4e14450e..a66784f78 100644
--- a/docs/SMOKE_TEST_NOTE.md
+++ b/docs/SMOKE_TEST_NOTE.md
@@ -41,8 +41,8 @@ transaction, no encrypted collection, not permitted, etc.).
 
 ## Replica Set Tests — Standalone Errors (24)
 
-These smoke tests are marked with `replica_set` and cannot run against a standalone MongoDB instance.
-They are automatically skipped when the server is not a replica set member.
+These smoke tests carry a `requires_*` capability marker (change streams, transactions, queryable encryption, or cluster admin) and cannot run against a standalone MongoDB instance.
+They are automatically skipped when the connected server does not provide the capability.
 Below are the errors returned when running each against a standalone MongoDB instance.
 
 ### Change Streams (16)
diff --git a/documentdb_tests/compatibility/result_analyzer/report_generator.py b/documentdb_tests/compatibility/result_analyzer/report_generator.py
index 8314f2a9e..ba38e89a6 100644
--- a/documentdb_tests/compatibility/result_analyzer/report_generator.py
+++ b/documentdb_tests/compatibility/result_analyzer/report_generator.py
@@ -157,7 +157,7 @@ def generate_text_report(analysis: Dict[str, Any], output_path: str):
     lines.append("=" * 80)
 
     with open(output_path, "w") as f:
-        f.write("\n".join(lines))
+        f.write("\n".join(lines) + "\n")
 
 
 def print_summary(analysis: Dict[str, Any]):
diff --git a/documentdb_tests/compatibility/tests/changeStreams/create/test_smoke_changeStream_create.py b/documentdb_tests/compatibility/tests/changeStreams/create/test_smoke_changeStream_create.py
index 20e625525..d5009d8c7 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/create/test_smoke_changeStream_create.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/create/test_smoke_changeStream_create.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_create(collection):
     """Test basic create change stream event behavior."""
     result = execute_command(
diff --git a/documentdb_tests/compatibility/tests/changeStreams/createIndexes/test_smoke_changeStream_createIndexes.py b/documentdb_tests/compatibility/tests/changeStreams/createIndexes/test_smoke_changeStream_createIndexes.py
index 20b167e97..44041944c 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/createIndexes/test_smoke_changeStream_createIndexes.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/createIndexes/test_smoke_changeStream_createIndexes.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_createIndexes(collection):
     """Test basic createIndexes change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/delete/test_smoke_changeStream_delete.py b/documentdb_tests/compatibility/tests/changeStreams/delete/test_smoke_changeStream_delete.py
index 8813d24f1..e7e5cd0e3 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/delete/test_smoke_changeStream_delete.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/delete/test_smoke_changeStream_delete.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_delete(collection):
     """Test basic delete change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/drop/test_smoke_changeStream_drop.py b/documentdb_tests/compatibility/tests/changeStreams/drop/test_smoke_changeStream_drop.py
index cab5e395c..368db567d 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/drop/test_smoke_changeStream_drop.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/drop/test_smoke_changeStream_drop.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_drop(collection):
     """Test basic drop change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/dropDatabase/test_smoke_changeStream_dropDatabase.py b/documentdb_tests/compatibility/tests/changeStreams/dropDatabase/test_smoke_changeStream_dropDatabase.py
index fca9e497b..16c941e16 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/dropDatabase/test_smoke_changeStream_dropDatabase.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/dropDatabase/test_smoke_changeStream_dropDatabase.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_dropDatabase(collection):
     """Test basic dropDatabase change stream event behavior."""
     execute_command(collection, {"create": f"{collection.name}_temp"})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/dropIndexes/test_smoke_changeStream_dropIndexes.py b/documentdb_tests/compatibility/tests/changeStreams/dropIndexes/test_smoke_changeStream_dropIndexes.py
index 182cf6ddb..cbac28e18 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/dropIndexes/test_smoke_changeStream_dropIndexes.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/dropIndexes/test_smoke_changeStream_dropIndexes.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_dropIndexes(collection):
     """Test basic dropIndexes change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/insert/test_smoke_changeStream_insert.py b/documentdb_tests/compatibility/tests/changeStreams/insert/test_smoke_changeStream_insert.py
index 554de3d2b..751605217 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/insert/test_smoke_changeStream_insert.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/insert/test_smoke_changeStream_insert.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_insert(collection):
     """Test basic insert change stream event behavior."""
     result = execute_command(
diff --git a/documentdb_tests/compatibility/tests/changeStreams/invalidate/test_smoke_changeStream_invalidate.py b/documentdb_tests/compatibility/tests/changeStreams/invalidate/test_smoke_changeStream_invalidate.py
index eefa03bdb..b07e108ab 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/invalidate/test_smoke_changeStream_invalidate.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/invalidate/test_smoke_changeStream_invalidate.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_invalidate(collection):
     """Test basic invalidate change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/modify/test_smoke_changeStream_modify.py b/documentdb_tests/compatibility/tests/changeStreams/modify/test_smoke_changeStream_modify.py
index 03bacafde..0a98ca8ee 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/modify/test_smoke_changeStream_modify.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/modify/test_smoke_changeStream_modify.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_modify(collection):
     """Test basic modify change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/refineCollectionShardKey/test_smoke_changeStream_refineCollectionShardKey.py b/documentdb_tests/compatibility/tests/changeStreams/refineCollectionShardKey/test_smoke_changeStream_refineCollectionShardKey.py
index ff91bcd7f..b84b13eb1 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/refineCollectionShardKey/test_smoke_changeStream_refineCollectionShardKey.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/refineCollectionShardKey/test_smoke_changeStream_refineCollectionShardKey.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 @pytest.mark.skip(reason="refineCollectionShardKey operations not supported in this environment")
 def test_smoke_changeStream_refineCollectionShardKey(collection):
     """Test basic refineCollectionShardKey change stream event behavior."""
diff --git a/documentdb_tests/compatibility/tests/changeStreams/rename/test_smoke_changeStream_rename.py b/documentdb_tests/compatibility/tests/changeStreams/rename/test_smoke_changeStream_rename.py
index 372a9fa48..a8d57e3c0 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/rename/test_smoke_changeStream_rename.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/rename/test_smoke_changeStream_rename.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_rename(collection):
     """Test basic rename change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/replace/test_smoke_changeStream_replace.py b/documentdb_tests/compatibility/tests/changeStreams/replace/test_smoke_changeStream_replace.py
index e3e82f41d..38e68e0f1 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/replace/test_smoke_changeStream_replace.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/replace/test_smoke_changeStream_replace.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_replace(collection):
     """Test basic replace change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/changeStreams/reshardCollection/test_smoke_changeStream_reshardCollection.py b/documentdb_tests/compatibility/tests/changeStreams/reshardCollection/test_smoke_changeStream_reshardCollection.py
index ff8f99107..f4f4418b7 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/reshardCollection/test_smoke_changeStream_reshardCollection.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/reshardCollection/test_smoke_changeStream_reshardCollection.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 @pytest.mark.skip(reason="reshardCollection operations not supported in this environment")
 def test_smoke_changeStream_reshardCollection(collection):
     """Test basic reshardCollection change stream event behavior."""
diff --git a/documentdb_tests/compatibility/tests/changeStreams/shardCollection/test_smoke_changeStream_shardCollection.py b/documentdb_tests/compatibility/tests/changeStreams/shardCollection/test_smoke_changeStream_shardCollection.py
index 2979d93d1..918ea1505 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/shardCollection/test_smoke_changeStream_shardCollection.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/shardCollection/test_smoke_changeStream_shardCollection.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 @pytest.mark.skip(reason="shardCollection events not captured even with showExpandedEvents")
 def test_smoke_changeStream_shardCollection(collection):
     """Test basic shardCollection change stream event behavior."""
diff --git a/documentdb_tests/compatibility/tests/changeStreams/update/test_smoke_changeStream_update.py b/documentdb_tests/compatibility/tests/changeStreams/update/test_smoke_changeStream_update.py
index cc3ff3e94..01c30b4db 100644
--- a/documentdb_tests/compatibility/tests/changeStreams/update/test_smoke_changeStream_update.py
+++ b/documentdb_tests/compatibility/tests/changeStreams/update/test_smoke_changeStream_update.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream_update(collection):
     """Test basic update change stream event behavior."""
     collection.insert_one({"_id": 1, "x": 1})
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_pipeline.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_pipeline.py
index b5c4ff7ee..2d07970bf 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_pipeline.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_pipeline.py
@@ -162,25 +162,28 @@
             error_code=TYPE_MISMATCH_ERROR,
             msg=f"aggregate should reject {tid} pipeline",
         )
-        for tid, val in [
-            ("null", None),
-            ("bool", True),
-            ("int", 1),
-            ("int64", Int64(1)),
-            ("double", 1.5),
-            ("decimal128", Decimal128("1")),
-            ("string", "hello"),
-            ("document", {"$match": {}}),
-            ("objectid", ObjectId()),
-            ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)),
-            ("timestamp", Timestamp(1, 1)),
-            ("binary", Binary(b"data")),
-            ("regex", Regex(".*")),
-            ("code", Code("function(){}")),
-            ("code_with_scope", Code("function(){}", {"x": 1})),
-            ("minkey", MinKey()),
-            ("maxkey", MaxKey()),
-        ]
+        for tid, val in cast(
+            list[tuple[str, object]],
+            [
+                ("null", None),
+                ("bool", True),
+                ("int", 1),
+                ("int64", Int64(1)),
+                ("double", 1.5),
+                ("decimal128", Decimal128("1")),
+                ("string", "hello"),
+                ("document", {"$match": {}}),
+                ("objectid", ObjectId()),
+                ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)),
+                ("timestamp", Timestamp(1, 1)),
+                ("binary", Binary(b"data")),
+                ("regex", Regex(".*")),
+                ("code", Code("function(){}")),
+                ("code_with_scope", Code("function(){}", {"x": 1})),
+                ("minkey", MinKey()),
+                ("maxkey", MaxKey()),
+            ],
+        )
     ],
     CommandTestCase(
         "pipeline_missing",
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_stages.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_stages.py
index 215de7cc1..563747074 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_stages.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_stages.py
@@ -253,6 +253,7 @@
         },
         error_code=NOT_A_REPLICA_SET_ERROR,
         msg="aggregate should reject readConcern 'linearizable' on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     CommandTestCase(
         "rc_standalone_snapshot",
@@ -265,6 +266,7 @@
         },
         error_code=NOT_A_REPLICA_SET_ERROR,
         msg="aggregate should reject readConcern 'snapshot' on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_subfield_afterclustertime.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_subfield_afterclustertime.py
index 294164e62..17ae81ddd 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_subfield_afterclustertime.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_readconcern_subfield_afterclustertime.py
@@ -261,6 +261,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="aggregate should reject non-zero afterClusterTime on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_writeconcern_w.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_writeconcern_w.py
index ff93a657a..529965fd9 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_writeconcern_w.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/aggregate/test_aggregate_writeconcern_w.py
@@ -74,6 +74,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should accept w=50 as in-range; standalone rejects w>1 not as out-of-range",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_int64_zero",
@@ -194,6 +195,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should truncate w=50.99 to 50; standalone rejects w>1 not as out-of-range",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_decimal128_bankers_0_5",
@@ -218,6 +220,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should round Decimal128 w=1.5 to 2; standalone rejects w>1 not out-of-range",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_decimal128_bankers_50_5",
@@ -230,6 +233,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should round Decimal128 w=50.5 to 50; standalone rejects not out-of-range",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_string_majority",
@@ -295,6 +299,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should reject null w coerced to empty string",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_reject_negative",
@@ -331,6 +336,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="aggregate should reject w>1 on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     *[
         CommandTestCase(
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/count/test_count_read_concern.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/count/test_count_read_concern.py
index 26d32f45c..3b7795326 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/count/test_count_read_concern.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/count/test_count_read_concern.py
@@ -413,6 +413,7 @@
         },
         error_code=NOT_A_REPLICA_SET_ERROR,
         msg="count with linearizable readConcern should fail on non-replica-set",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     CommandTestCase(
         "type_readconcern_snapshot",
@@ -448,6 +449,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="count afterClusterTime should be rejected on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     *[
         CommandTestCase(
diff --git a/documentdb_tests/compatibility/tests/core/aggregation/commands/distinct/test_distinct_readconcern_subfields.py b/documentdb_tests/compatibility/tests/core/aggregation/commands/distinct/test_distinct_readconcern_subfields.py
index d464f082b..cd5fcb7d3 100644
--- a/documentdb_tests/compatibility/tests/core/aggregation/commands/distinct/test_distinct_readconcern_subfields.py
+++ b/documentdb_tests/compatibility/tests/core/aggregation/commands/distinct/test_distinct_readconcern_subfields.py
@@ -82,6 +82,7 @@
         },
         error_code=NOT_A_REPLICA_SET_ERROR,
         msg="distinct with linearizable readConcern should fail on non-replica-set",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     CommandTestCase(
         "readconcern_snapshot",
@@ -93,6 +94,7 @@
         },
         error_code=NOT_A_REPLICA_SET_ERROR,
         msg="distinct with snapshot readConcern should fail on non-replica-set",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     *[
         CommandTestCase(
@@ -156,6 +158,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="distinct afterClusterTime should be rejected on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     *[
         CommandTestCase(
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/cloneCollectionAsCapped/test_cloneCollectionAsCapped_wc_w.py b/documentdb_tests/compatibility/tests/core/collections/commands/cloneCollectionAsCapped/test_cloneCollectionAsCapped_wc_w.py
index 3dd1656eb..39f7669d6 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/cloneCollectionAsCapped/test_cloneCollectionAsCapped_wc_w.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/cloneCollectionAsCapped/test_cloneCollectionAsCapped_wc_w.py
@@ -121,6 +121,7 @@
         },
         expected={"ok": 1.0},
         msg="w=object with numeric tag value should succeed",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
@@ -162,6 +163,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=null should coerce to empty string and fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_objectid",
@@ -477,6 +479,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=2 on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_custom_string",
@@ -489,6 +492,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='custom' on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_majority_case_sensitive",
@@ -501,6 +505,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='Majority' (wrong case) on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_decimal128_1_5",
@@ -513,6 +518,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=Decimal128('1.5') rounds to 2 and should fail with bad value on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_empty_string",
@@ -525,6 +531,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=empty string should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_bool_type_errors.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_bool_type_errors.py
index 72d195b5a..5d3b58040 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_bool_type_errors.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_bool_type_errors.py
@@ -265,6 +265,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_BOOL_TYPE_ERROR_TESTS))
 def test_compact_bool_type_errors(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_collection_types.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_collection_types.py
index 55176c2e2..922849eb0 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_collection_types.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_collection_types.py
@@ -170,6 +170,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_COLLECTION_TYPE_ALL_TESTS))
 def test_compact_collection_types(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_comment.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_comment.py
index 5efecde45..0bed8cb27 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_comment.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_comment.py
@@ -335,6 +335,7 @@
 ]
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_COMMENT_ACCEPTANCE_TESTS))
 def test_compact_comment(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace.py
index 07099f29a..4145bc1d2 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace.py
@@ -310,6 +310,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_FREESPACE_TESTS))
 def test_compact_freespace(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace_overflow.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace_overflow.py
index ece6dc565..f4238b502 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace_overflow.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_freespace_overflow.py
@@ -89,6 +89,7 @@
 ]
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_FREESPACE_OVERFLOW_TESTS))
 def test_compact_freespace_overflow(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_invalid_names.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_invalid_names.py
index 33d60db82..688f954db 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_invalid_names.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_invalid_names.py
@@ -198,6 +198,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_INVALID_NAME_TESTS))
 def test_compact_invalid_names(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_maxtimems.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_maxtimems.py
index f811d3da6..ccd35b7d1 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_maxtimems.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_maxtimems.py
@@ -395,6 +395,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_MAXTIMEMS_TESTS))
 def test_compact_maxtimems(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_parameters.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_parameters.py
index 1671b3732..b105cc24a 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_parameters.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_parameters.py
@@ -11,7 +11,8 @@
 from documentdb_tests.framework.parametrize import pytest_params
 
 # Property [Null and Missing Behavior]: null values for optional parameters
-# are accepted and treated identically to omitting the field.
+# are accepted and treated identically to omitting the field. These omit force,
+# so they are gated on the precondition that compact succeeds without it.
 COMPACT_NULL_AND_MISSING_TESTS: list[CommandTestCase] = [
     CommandTestCase(
         "dry_run_null",
@@ -19,6 +20,7 @@
         command=lambda ctx: {"compact": ctx.collection, "dryRun": None},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="null dryRun should be treated as omitted (defaults to false)",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "force_null",
@@ -26,6 +28,7 @@
         command=lambda ctx: {"compact": ctx.collection, "force": None},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="null force should be treated as omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "free_space_target_mb_null",
@@ -33,6 +36,7 @@
         command=lambda ctx: {"compact": ctx.collection, "freeSpaceTargetMB": None},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="null freeSpaceTargetMB should be treated as omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "comment_null",
@@ -40,6 +44,7 @@
         command=lambda ctx: {"compact": ctx.collection, "comment": None},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="null comment should be accepted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "all_null",
@@ -53,6 +58,7 @@
         },
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="All-null optional parameters should be treated as all omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "write_concern_null",
@@ -60,11 +66,13 @@
         command=lambda ctx: {"compact": ctx.collection, "writeConcern": None},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="null writeConcern should be treated as omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
 ]
 
 # Property [Response Format]: the response structure differs based on
-# whether dryRun is enabled.
+# whether dryRun is enabled. These omit force, so they are gated on the
+# precondition that compact succeeds without it.
 COMPACT_RESPONSE_FORMAT_TESTS: list[CommandTestCase] = [
     CommandTestCase(
         "without_dry_run",
@@ -72,6 +80,7 @@
         command=lambda ctx: {"compact": ctx.collection},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="Should return bytesFreed and ok when dryRun is omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "dry_run_false",
@@ -79,6 +88,7 @@
         command=lambda ctx: {"compact": ctx.collection, "dryRun": False},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="Should return bytesFreed and ok when dryRun is false",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
     CommandTestCase(
         "dry_run_true",
@@ -86,11 +96,12 @@
         command=lambda ctx: {"compact": ctx.collection, "dryRun": True},
         expected={"estimatedBytesFreed": 0, "ok": 1.0},
         msg="Should return estimatedBytesFreed when dryRun is true",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
 ]
 
-# Property [force Behavior]: force=true is accepted on standalone with no
-# observable difference from force=false or omitted.
+# Property [force Behavior]: force=true is accepted with no observable
+# difference from force=false or omitted.
 COMPACT_FORCE_BEHAVIOR_TESTS: list[CommandTestCase] = [
     CommandTestCase(
         "force_true",
@@ -105,6 +116,7 @@
         command=lambda ctx: {"compact": ctx.collection, "force": False},
         expected={"bytesFreed": 0, "ok": 1.0},
         msg="force=false should be accepted with same response as omitted",
+        marks=(pytest.mark.requires(unforced_compact=True),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_readconcern.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_readconcern.py
index 4a177676b..9d20951e7 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_readconcern.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_readconcern.py
@@ -237,6 +237,7 @@
 )
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_READCONCERN_TESTS))
 def test_compact_readconcern(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_unrecognized_fields.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_unrecognized_fields.py
index 97c74fb5d..ca375eb03 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_unrecognized_fields.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_unrecognized_fields.py
@@ -32,6 +32,7 @@
 ]
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_UNRECOGNIZED_FIELDS_TESTS))
 def test_compact_unrecognized_fields(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_valid_names.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_valid_names.py
index 7942bf446..68ca15de6 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_valid_names.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_valid_names.py
@@ -241,6 +241,7 @@
 ]
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_VALID_NAME_TESTS))
 def test_compact_valid_names(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_write_concern.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_write_concern.py
index fec614a43..6cecfcb37 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_write_concern.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_compact_write_concern.py
@@ -97,6 +97,7 @@
 ]
 
 
+@pytest.mark.requires(unforced_compact=True)
 @pytest.mark.collection_mgmt
 @pytest.mark.parametrize("test", pytest_params(COMPACT_WRITE_CONCERN_REJECTION_TESTS))
 def test_compact_write_concern(database_client, collection, test):
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_smoke_compact.py b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_smoke_compact.py
index 6be0e397b..11441370f 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_smoke_compact.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/compact/test_smoke_compact.py
@@ -12,6 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
+@pytest.mark.requires(unforced_compact=True)
 def test_smoke_compact(collection):
     """Test basic compact command behavior."""
     collection.insert_many([{"_id": 1, "value": 10}, {"_id": 2, "value": 20}])
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc.py b/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc.py
index e099ce194..8a073c176 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc.py
@@ -101,6 +101,9 @@
         },
         expected={"ok": 1.0},
         msg=f"wtimeout={id} should succeed",
+        # A negative wtimeout is accepted cleanly on a standalone server but a
+        # replica set reports a writeConcernError, so gate that case.
+        marks=((pytest.mark.requires(quorum_write_concern=False),) if id == "negative" else ()),
     )
     for id, val in [
         ("zero", 0),
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc_w.py b/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc_w.py
index 6398f6e2a..ecf3cd38c 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc_w.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/convertToCapped/test_convertToCapped_wc_w.py
@@ -116,6 +116,7 @@
         },
         expected={"ok": 1.0},
         msg="w=object with numeric tag value should succeed",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
@@ -154,6 +155,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=null should coerce to empty string and fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_objectid",
@@ -466,6 +468,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=2 on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_custom_string",
@@ -477,6 +480,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='custom' on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_majority_case_sensitive",
@@ -488,6 +492,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='Majority' (wrong case) on standalone should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_decimal128_1_5",
@@ -499,6 +504,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=Decimal128('1.5') rounds to 2 and should fail with bad value on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "w_empty_string",
@@ -510,6 +516,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=empty string should fail with bad value",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_clustered_index_acceptance.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_clustered_index_acceptance.py
index 518c22e89..fd11d2b99 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_clustered_index_acceptance.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_clustered_index_acceptance.py
@@ -473,7 +473,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="Clustered with encryptedFields should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_acceptance.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_acceptance.py
index e58f560b3..2bb5cc2d9 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_acceptance.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_acceptance.py
@@ -35,7 +35,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="Minimal encryptedFields with path and keyId should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="optional_bson_type",
@@ -53,7 +53,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with optional bsonType should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="optional_queries_object",
@@ -72,7 +72,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with queries as object should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="optional_queries_array",
@@ -91,7 +91,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with queries as array should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
 ]
 
@@ -109,7 +109,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="Custom escCollection with valid naming pattern should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="ecoc_collection_custom",
@@ -122,7 +122,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="Custom ecocCollection with valid naming pattern should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="ecc_collection_custom",
@@ -135,7 +135,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="eccCollection does not have naming pattern validation",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="null_elements_in_fields_array",
@@ -151,7 +151,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="Null elements in fields array should be silently accepted",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
 ]
 
@@ -168,7 +168,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with clusteredIndex should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="compatible_with_expire_after_seconds",
@@ -180,7 +180,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with expireAfterSeconds should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="compatible_with_storage_engine",
@@ -191,7 +191,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with storageEngine should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="compatible_with_validator",
@@ -202,7 +202,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with validator should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="compatible_with_collation",
@@ -213,7 +213,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with collation should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="compatible_with_change_stream_pre_post_images",
@@ -224,7 +224,7 @@
         },
         expected={"ok": Eq(1.0)},
         msg="encryptedFields with changeStreamPreAndPostImages should succeed",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_query.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_query.py
index 97bd72272..2b377b5a6 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_query.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_encrypted_fields_query.py
@@ -477,7 +477,7 @@
         },
         error_code=ENCRYPTED_FIELD_TRIM_FACTOR_OUT_OF_RANGE_ERROR,
         msg="trimFactor must be less than the bit width of the field type",
-        marks=(pytest.mark.replica_set,),
+        marks=(pytest.mark.requires(queryable_encryption=True),),
     ),
     CommandTestCase(
         id="ef_err_contention_fractional",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_view_errors.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_view_errors.py
index 8236070f2..3f43f1e16 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_view_errors.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_view_errors.py
@@ -282,6 +282,7 @@
         },
         error_code=CHANGE_STREAM_NOT_ALLOWED_ERROR,
         msg="$changeStream in view pipeline should fail",
+        marks=(pytest.mark.requires(change_streams=False),),
     ),
     CommandTestCase(
         id="current_op_stage",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_acceptance.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_acceptance.py
index 0117fe4a2..dcb94bda8 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_acceptance.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_acceptance.py
@@ -110,6 +110,7 @@
         },
         expected={"ok": 1.0},
         msg="w as tagged object with numeric values should succeed",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_validation.py b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_validation.py
index bcfbd0163..d9b8f96b3 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_validation.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/create/test_create_write_concern_validation.py
@@ -110,6 +110,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w:null coerced to empty string should fail",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         id="wc_w_negative",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_w.py b/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_w.py
index 3b9bdf598..8df3466f7 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_w.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_w.py
@@ -79,114 +79,133 @@
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": 1}}},
         expected={"ok": 1.0},
         msg="w:tagged object with int32 value should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_int32",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": Int64(1)}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Int64 value should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_int64",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": 1.5}}},
         expected={"ok": 1.0},
         msg="w:tagged object with double value should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_double",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": Decimal128("1")}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 value should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": 0}}},
         expected={"ok": 1.0},
         msg="w:tagged object with zero should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_zero",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": -1}}},
         expected={"ok": 1.0},
         msg="w:tagged object with negative value should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_negative",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": FLOAT_NAN}}},
         expected={"ok": 1.0},
         msg="w:tagged object with NaN should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_nan",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": FLOAT_INFINITY}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_infinity",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": FLOAT_NEGATIVE_INFINITY}}},
         expected={"ok": 1.0},
         msg="w:tagged object with -Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_neg_infinity",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DOUBLE_NEGATIVE_ZERO}}},
         expected={"ok": 1.0},
         msg="w:tagged object with -0.0 should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_neg_zero",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": INT64_MAX}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Int64 max should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_int64_max",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": INT64_MIN}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Int64 min should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_int64_min",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DECIMAL128_NAN}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 NaN should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128_nan",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DECIMAL128_INFINITY}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128_infinity",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": 1, "b": 2}}},
         expected={"ok": 1.0},
         msg="w:multi-key tagged object should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_multi_key",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": FLOAT_NEGATIVE_NAN}}},
         expected={"ok": 1.0},
         msg="w:tagged object with negative NaN should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_neg_nan",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DECIMAL128_NEGATIVE_NAN}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 negative NaN should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128_neg_nan",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DECIMAL128_NEGATIVE_INFINITY}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 -Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128_neg_infinity",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": {"a": DECIMAL128_NEGATIVE_ZERO}}},
         expected={"ok": 1.0},
         msg="w:tagged object with Decimal128 -0 should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_tagged_decimal128_neg_zero",
     ),
 ]
@@ -278,24 +297,28 @@
         command={"dropDatabase": 1, "writeConcern": {"w": 2}},
         error_code=BAD_VALUE_ERROR,
         msg="w:2 should produce a bad value error on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_2_standalone",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": 50}},
         error_code=BAD_VALUE_ERROR,
         msg="w:50 should produce a bad value error on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_50_standalone",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": "foobar"}},
         error_code=BAD_VALUE_ERROR,
         msg="w:non-majority string should produce a bad value error on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_string_non_majority",
     ),
     CommandTestCase(
         command={"dropDatabase": 1, "writeConcern": {"w": ""}},
         error_code=BAD_VALUE_ERROR,
         msg="w:empty string should produce a bad value error on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_string_empty",
     ),
 ]
@@ -516,6 +539,7 @@
         command={"dropDatabase": 1, "writeConcern": {"w": None}},
         error_code=BAD_VALUE_ERROR,
         msg="w:null should be treated as empty string and produce a bad value error",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_null",
     ),
 ]
@@ -607,6 +631,7 @@
         command={"dropDatabase": 1, "writeConcern": {"w": DECIMAL128_ONE_AND_HALF}},
         error_code=BAD_VALUE_ERROR,
         msg="w:Decimal128('1.5') rounds up, crossing into standalone rejection",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="w_decimal128_1_5",
     ),
     CommandTestCase(
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_wtimeout.py b/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_wtimeout.py
index da13aaa64..28564de83 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_wtimeout.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/dropDatabase/test_dropDatabase_wc_wtimeout.py
@@ -97,6 +97,7 @@
         },
         expected={"ok": 1.0},
         msg="wtimeout:Int64 min should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="wtimeout_int64_min",
     ),
     CommandTestCase(
@@ -115,6 +116,7 @@
         command={"dropDatabase": 1, "writeConcern": {"wtimeout": FLOAT_NEGATIVE_INFINITY}},
         expected={"ok": 1.0},
         msg="wtimeout:-Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="wtimeout_float_neg_infinity",
     ),
     CommandTestCase(
@@ -133,6 +135,7 @@
         command={"dropDatabase": 1, "writeConcern": {"wtimeout": DECIMAL128_NEGATIVE_INFINITY}},
         expected={"ok": 1.0},
         msg="wtimeout:Decimal128('-Infinity') should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
         id="wtimeout_decimal128_neg_infinity",
     ),
     CommandTestCase(
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_read_concern.py b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_read_concern.py
index f5fd95322..0fa348462 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_read_concern.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_read_concern.py
@@ -316,6 +316,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="afterClusterTime should be rejected on standalone",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     CommandTestCase(
         "rc_after_cluster_time_null",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_special_collections.py b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_special_collections.py
index 95280842f..ff94becdd 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_special_collections.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_special_collections.py
@@ -39,6 +39,7 @@
         },
         expected={"ok": 1.0},
         msg="local.oplog (no suffix) as target should be accepted",
+        marks=(pytest.mark.requires(local_rename=True),),
     ),
     CommandTestCase(
         "oplog_rs_non_local_db",
@@ -291,6 +292,7 @@
         },
         expected={"ok": 1.0},
         msg="Renaming to local database (non-oplog collection) should succeed",
+        marks=(pytest.mark.requires(local_rename=True),),
     ),
     CommandTestCase(
         "to_config_db",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_w.py b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_w.py
index d02f4d390..77fb71b50 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_w.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_w.py
@@ -193,6 +193,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=2 on standalone should be rejected",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_50_standalone",
@@ -204,6 +205,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=50 on standalone should be rejected",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_decimal128_1_5_bankers_rounding",
@@ -215,6 +217,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=Decimal128('1.5') rounds to 2 (banker's rounding) → w>1 on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_decimal128_2_5_bankers_rounding",
@@ -226,6 +229,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=Decimal128('2.5') rounds to 2 (banker's rounding) → w>1 on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_decimal128_50_5_bankers_rounding",
@@ -237,6 +241,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=Decimal128('50.5') rounds to 50 (banker's rounding) → w>1 on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
@@ -368,6 +373,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=null should coerce to empty string and be rejected",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
@@ -385,6 +391,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='' (empty string) should be rejected on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_string_case_sensitive",
@@ -396,6 +403,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='Majority' (wrong case) should be rejected on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_string_leading_space",
@@ -407,6 +415,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w=' majority' (leading space) should be rejected on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_string_other",
@@ -418,6 +427,7 @@
         },
         error_code=BAD_VALUE_ERROR,
         msg="w='other' should be rejected on standalone",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
@@ -521,6 +531,7 @@
         },
         expected={"ok": 1.0},
         msg="w={tag:1} (number value) should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_object_dollar_key",
@@ -532,6 +543,7 @@
         },
         expected={"ok": 1.0},
         msg="w={'$tag':1} (dollar-prefixed key) should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_w_object_empty_key",
@@ -543,6 +555,7 @@
         },
         expected={"ok": 1.0},
         msg="w={'':1} (empty-string key) should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_wtimeout.py b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_wtimeout.py
index 4bbee1742..c04139e9b 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_wtimeout.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/renameCollection/test_renameCollection_wc_wtimeout.py
@@ -128,6 +128,7 @@
         },
         expected={"ok": 1.0},
         msg="wtimeout=-1 (negative) should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_wtimeout_neg_infinity",
@@ -139,6 +140,7 @@
         },
         expected={"ok": 1.0},
         msg="wtimeout=-Infinity should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_wtimeout_nan",
@@ -271,6 +273,7 @@
         },
         expected={"ok": 1.0},
         msg="wtimeout=Decimal128('-Infinity') should be accepted",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     CommandTestCase(
         "wc_wtimeout_neg_zero",
diff --git a/documentdb_tests/compatibility/tests/core/collections/commands/validateDBMetadata/test_validateDBMetadata_read_write_concern.py b/documentdb_tests/compatibility/tests/core/collections/commands/validateDBMetadata/test_validateDBMetadata_read_write_concern.py
index 7a076ed71..f7a8b6e6f 100644
--- a/documentdb_tests/compatibility/tests/core/collections/commands/validateDBMetadata/test_validateDBMetadata_read_write_concern.py
+++ b/documentdb_tests/compatibility/tests/core/collections/commands/validateDBMetadata/test_validateDBMetadata_read_write_concern.py
@@ -141,6 +141,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="validateDBMetadata should reject afterClusterTime in readConcern",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_lifecycle.py b/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_lifecycle.py
index 3b3d50099..9932ec6f4 100644
--- a/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_lifecycle.py
+++ b/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_lifecycle.py
@@ -491,12 +491,12 @@ def test_killCursors_after_drop_database(engine_client, database_client, collect
 
 # Property [Cross-Connection Behavior]: a cursor created on one
 # connection can be killed from a different connection to the same server.
-def test_killCursors_cross_connection(request, database_client, collection):
+def test_killCursors_cross_connection(connection_string, database_client, collection):
     """Test killing a cursor from a different connection."""
     collection.insert_many([{"_id": i} for i in range(10)])
     (cursor_id,) = open_find_cursors(collection, 1)
 
-    second_client = fixtures.create_engine_client(request.config.connection_string, "second")
+    second_client = fixtures.create_engine_client(connection_string, "second")
     try:
         second_coll = second_client[database_client.name][collection.name]
         result = execute_command(
diff --git a/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_readconcern.py b/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_readconcern.py
index 39a2fdc67..135748142 100644
--- a/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_readconcern.py
+++ b/documentdb_tests/compatibility/tests/core/cursors/commands/killCursors/test_killCursors_readconcern.py
@@ -240,6 +240,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="killCursors should reject afterClusterTime in readConcern",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
     CommandTestCase(
         "readconcern_provenance_invalid",
diff --git a/documentdb_tests/compatibility/tests/core/indexes/commands/dropIndexes/test_dropIndexes_type_validation.py b/documentdb_tests/compatibility/tests/core/indexes/commands/dropIndexes/test_dropIndexes_type_validation.py
index 61889b966..09c5527d9 100644
--- a/documentdb_tests/compatibility/tests/core/indexes/commands/dropIndexes/test_dropIndexes_type_validation.py
+++ b/documentdb_tests/compatibility/tests/core/indexes/commands/dropIndexes/test_dropIndexes_type_validation.py
@@ -281,6 +281,7 @@ def test_dropIndexes_writeConcern_non_object(collection, test):
         write_concern={"w": "invalid"},
         error_code=BAD_VALUE_ERROR,
         msg="Invalid string w value should fail with BadValue",
+        marks=(pytest.mark.requires(quorum_write_concern=False),),
     ),
     IndexTestCase(
         "w_negative",
diff --git a/documentdb_tests/compatibility/tests/core/indexes/commands/reIndex/test_smoke_reIndex.py b/documentdb_tests/compatibility/tests/core/indexes/commands/reIndex/test_smoke_reIndex.py
index 95280c56c..6d4a5adcf 100644
--- a/documentdb_tests/compatibility/tests/core/indexes/commands/reIndex/test_smoke_reIndex.py
+++ b/documentdb_tests/compatibility/tests/core/indexes/commands/reIndex/test_smoke_reIndex.py
@@ -12,6 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
+@pytest.mark.requires(reindex=True)
 def test_smoke_reIndex(collection):
     """Test basic reIndex command behavior."""
     collection.insert_one({"_id": 1, "name": "test"})
diff --git a/documentdb_tests/compatibility/tests/core/operator/query/arrays/elemMatch/test_elemMatch_matching.py b/documentdb_tests/compatibility/tests/core/operator/query/arrays/elemMatch/test_elemMatch_matching.py
index f914fe38e..70dddbd8d 100644
--- a/documentdb_tests/compatibility/tests/core/operator/query/arrays/elemMatch/test_elemMatch_matching.py
+++ b/documentdb_tests/compatibility/tests/core/operator/query/arrays/elemMatch/test_elemMatch_matching.py
@@ -376,7 +376,7 @@
 )
 
 
-UTC_CODEC = CodecOptions(tz_aware=True, tzinfo=timezone.utc)
+UTC_CODEC: CodecOptions = CodecOptions(tz_aware=True, tzinfo=timezone.utc)
 
 
 @pytest.mark.parametrize("test", pytest_params(ALL_MATCHING_TESTS))
diff --git a/documentdb_tests/compatibility/tests/core/operator/query/misc/exists/test_exists_bson_type_coverage.py b/documentdb_tests/compatibility/tests/core/operator/query/misc/exists/test_exists_bson_type_coverage.py
index ec47a96c7..757213edc 100644
--- a/documentdb_tests/compatibility/tests/core/operator/query/misc/exists/test_exists_bson_type_coverage.py
+++ b/documentdb_tests/compatibility/tests/core/operator/query/misc/exists/test_exists_bson_type_coverage.py
@@ -173,7 +173,7 @@
 ALL_TESTS = EXISTS_TRUE_BSON_TESTS
 
 
-TZ_AWARE_CODEC = CodecOptions(tz_aware=True)
+TZ_AWARE_CODEC: CodecOptions = CodecOptions(tz_aware=True)
 
 
 @pytest.mark.parametrize("test", pytest_params(ALL_TESTS))
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/changeStreamSplitLargeEvent/test_smoke_changeStreamSplitLargeEvent.py b/documentdb_tests/compatibility/tests/core/operator/stages/changeStreamSplitLargeEvent/test_smoke_changeStreamSplitLargeEvent.py
index ea45beac5..1b59ac7d1 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/changeStreamSplitLargeEvent/test_smoke_changeStreamSplitLargeEvent.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/changeStreamSplitLargeEvent/test_smoke_changeStreamSplitLargeEvent.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStreamSplitLargeEvent(collection):
     """Test basic $changeStreamSplitLargeEvent stage behavior."""
     result = execute_command(
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/collStats/test_collStats_scale.py b/documentdb_tests/compatibility/tests/core/operator/stages/collStats/test_collStats_scale.py
index eac4a8e54..2a1706532 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/collStats/test_collStats_scale.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/collStats/test_collStats_scale.py
@@ -5,7 +5,7 @@
 from datetime import datetime
 
 import pytest
-from bson import Binary, Code, Decimal128, Int64, ObjectId, Regex, Timestamp
+from bson import Binary, Code, Decimal128, Int64, ObjectId, Regex, Timestamp, encode
 from bson.max_key import MaxKey
 from bson.min_key import MinKey
 
@@ -330,40 +330,25 @@ def test_collStats_scale(database_client, collection, test):
 
 
 # Property [Scale Factor Divides Size Fields]: specifying storageStats.scale=N
-# divides size fields (size, storageSize, totalSize, totalIndexSize, each entry
-# in indexSizes) by N using floor division, while avgObjSize, count, nindexes,
-# capped, and scaleFactor are unaffected.
-SCALE_AFFECTED_FIELDS = [
-    "size",
-    "storageSize",
-    "totalSize",
-    "totalIndexSize",
-]
-
-SCALE_UNAFFECTED_FIELDS = [
-    "avgObjSize",
-    "count",
-    "nindexes",
-    "capped",
-    "scaleFactor",
-    "indexBuilds",
-]
-
-
+# divides the size field by N using floor division, while the logical fields
+# count, avgObjSize, nindexes, capped, and scaleFactor are unaffected.
 @pytest.mark.aggregate
 def test_collStats_scale_divides_size_fields(collection):
-    """Test that scale=N floor-divides size fields and leaves others unchanged."""
-    collection.insert_many([{"_id": i, "x": "a" * 100} for i in range(50)])
-    collection.create_index("x")
+    """Test that scale=N floor-divides the size field and leaves logical fields unchanged.
+
+    The contract is checked on ``size`` because it is deterministic: it is the
+    sum of the inserted documents' BSON sizes, so its raw value is known and the
+    scaled value can be asserted exactly from a single read. The other size
+    fields (storageSize, totalSize, totalIndexSize, indexSizes) are
+    engine-managed on-disk allocations that can change between reads (e.g. a
+    background checkpoint); they share the same scaling code path, so asserting
+    it on a known field validates the contract without depending on
+    storage-engine timing.
+    """
+    docs = [{"_id": i, "x": "a" * 100} for i in range(50)]
+    raw_size = sum(len(encode(doc)) for doc in docs)
+    collection.insert_many(docs)
     scale = 3
-    base = execute_command(
-        collection,
-        {
-            "aggregate": collection.name,
-            "pipeline": [{"$collStats": {"storageStats": {}}}],
-            "cursor": {},
-        },
-    )
     scaled = execute_command(
         collection,
         {
@@ -372,20 +357,16 @@ def test_collStats_scale_divides_size_fields(collection):
             "cursor": {},
         },
     )
-    if isinstance(base, Exception):
-        raise AssertionError(f"unexpected error: {base}")
-    b = base["cursor"]["firstBatch"][0]["storageStats"]
-    checks: dict[str, Eq] = {"storageStats.nindexes": Eq(2)}
-    for field in SCALE_AFFECTED_FIELDS:
-        checks[f"storageStats.{field}"] = Eq(int(b[field] // scale))
-    for field in SCALE_UNAFFECTED_FIELDS:
-        checks[f"storageStats.{field}"] = Eq(scale if field == "scaleFactor" else b[field])
-    for key, val in b["indexSizes"].items():
-        checks[f"storageStats.indexSizes.{key}"] = Eq(int(val // scale))
     assertProperties(
         scaled,
-        checks,
-        msg="scale factor should divide size fields and leave others unchanged",
+        {
+            "storageStats.size": Eq(raw_size // scale),
+            "storageStats.count": Eq(len(docs)),
+            "storageStats.avgObjSize": Eq(raw_size // len(docs)),
+            "storageStats.nindexes": Eq(1),
+            "storageStats.scaleFactor": Eq(scale),
+        },
+        msg="scale factor should divide the size field and leave logical fields unchanged",
     )
 
 
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/group/test_group_errors.py b/documentdb_tests/compatibility/tests/core/operator/stages/group/test_group_errors.py
index 0df569430..ea458b64f 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/group/test_group_errors.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/group/test_group_errors.py
@@ -284,6 +284,7 @@
         msg="Bare '$$' in _id should produce a failed-to-parse error",
     ),
     StageTestCase(
+        marks=(pytest.mark.requires(cluster_time=False),),
         id="cluster_time_in_id_standalone",
         docs=[{"_id": 1}],
         pipeline=[{"$group": {"_id": "$$CLUSTER_TIME"}}],
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/listSampledQueries/test_smoke_listSampledQueries.py b/documentdb_tests/compatibility/tests/core/operator/stages/listSampledQueries/test_smoke_listSampledQueries.py
index a85d1c52e..dcadcca23 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/listSampledQueries/test_smoke_listSampledQueries.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/listSampledQueries/test_smoke_listSampledQueries.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_listSampledQueries(collection):
     """Test basic $listSampledQueries stage behavior."""
     result = execute_admin_command(
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/lookup/test_lookup_sub_pipeline.py b/documentdb_tests/compatibility/tests/core/operator/stages/lookup/test_lookup_sub_pipeline.py
index a22d2037d..c348d930f 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/lookup/test_lookup_sub_pipeline.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/lookup/test_lookup_sub_pipeline.py
@@ -78,6 +78,7 @@
         ],
         error_code=CHANGE_STREAM_NOT_ALLOWED_ERROR,
         msg="$lookup should reject $changeStream in the sub-pipeline",
+        marks=(pytest.mark.requires(change_streams=False),),
     ),
     LookupTestCase(
         "current_op_in_sub_pipeline",
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/out/test_out_target_restriction_errors.py b/documentdb_tests/compatibility/tests/core/operator/stages/out/test_out_target_restriction_errors.py
index d12f87815..a8bf4b785 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/out/test_out_target_restriction_errors.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/out/test_out_target_restriction_errors.py
@@ -24,6 +24,7 @@
     DUPLICATE_KEY_ERROR,
     ILLEGAL_OPERATION_ERROR,
     INVALID_OPTIONS_ERROR,
+    OPERATION_NOT_SUPPORTED_IN_TRANSACTION_ERROR,
     OPTION_NOT_SUPPORTED_ON_VIEW_ERROR,
     OUT_CAPPED_COLLECTION_ERROR,
     OUT_TIMESERIES_COLLECTION_TYPE_ERROR,
@@ -420,8 +421,9 @@ def test_out_schema_validation_error_unchanged(collection, test_case: OutTestCas
     OutTestCase(
         "transaction_out",
         docs=[{"_id": 1, "value": 10}],
-        error_code=ILLEGAL_OPERATION_ERROR,
+        error_code=OPERATION_NOT_SUPPORTED_IN_TRANSACTION_ERROR,
         msg="$out inside a transaction should produce an error",
+        marks=(pytest.mark.requires(transactions=True),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/querySettings/test_querySettings_structure.py b/documentdb_tests/compatibility/tests/core/operator/stages/querySettings/test_querySettings_structure.py
index 07b8a5027..c86ff569e 100644
--- a/documentdb_tests/compatibility/tests/core/operator/stages/querySettings/test_querySettings_structure.py
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/querySettings/test_querySettings_structure.py
@@ -190,7 +190,7 @@ def _teardown_query_settings(collection: Collection, query_settings: list[dict[s
 
 
 @pytest.mark.aggregate
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 @pytest.mark.no_parallel
 @pytest.mark.parametrize("test_case", pytest_params(QUERYSETTINGS_STRUCTURE_TESTS))
 def test_querySettings_structure(collection: Collection, test_case: QuerySettingsStructureTestCase):
diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_smoke_changeStream.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_smoke_changeStream.py
index 12d6efc36..21ebe8644 100644
--- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_smoke_changeStream.py
+++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_smoke_changeStream.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(change_streams=True)
 def test_smoke_changeStream(collection):
     """Test basic $changeStream system stage behavior."""
     result = execute_command(
diff --git a/documentdb_tests/compatibility/tests/core/query-planning/commands/removeQuerySettings/test_smoke_removeQuerySettings.py b/documentdb_tests/compatibility/tests/core/query-planning/commands/removeQuerySettings/test_smoke_removeQuerySettings.py
index b0417dcfa..2b14924e6 100644
--- a/documentdb_tests/compatibility/tests/core/query-planning/commands/removeQuerySettings/test_smoke_removeQuerySettings.py
+++ b/documentdb_tests/compatibility/tests/core/query-planning/commands/removeQuerySettings/test_smoke_removeQuerySettings.py
@@ -12,7 +12,7 @@
 pytestmark = [pytest.mark.smoke, pytest.mark.no_parallel]
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_removeQuerySettings(collection):
     """Test basic removeQuerySettings command behavior."""
     collection.insert_one({"_id": 1, "name": "Alice"})
diff --git a/documentdb_tests/compatibility/tests/core/query-planning/commands/setQuerySettings/test_smoke_setQuerySettings.py b/documentdb_tests/compatibility/tests/core/query-planning/commands/setQuerySettings/test_smoke_setQuerySettings.py
index 0e065a736..5d9107ebf 100644
--- a/documentdb_tests/compatibility/tests/core/query-planning/commands/setQuerySettings/test_smoke_setQuerySettings.py
+++ b/documentdb_tests/compatibility/tests/core/query-planning/commands/setQuerySettings/test_smoke_setQuerySettings.py
@@ -12,7 +12,7 @@
 pytestmark = [pytest.mark.smoke, pytest.mark.no_parallel]
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_setQuerySettings(collection):
     """Test basic setQuerySettings command behavior."""
     collection.insert_one({"_id": 1, "name": "Alice"})
diff --git a/documentdb_tests/compatibility/tests/core/sessions/commands/killAllSessions/test_killAllSessions_readconcern_errors.py b/documentdb_tests/compatibility/tests/core/sessions/commands/killAllSessions/test_killAllSessions_readconcern_errors.py
index 4abd9e50b..bed7f202f 100644
--- a/documentdb_tests/compatibility/tests/core/sessions/commands/killAllSessions/test_killAllSessions_readconcern_errors.py
+++ b/documentdb_tests/compatibility/tests/core/sessions/commands/killAllSessions/test_killAllSessions_readconcern_errors.py
@@ -159,6 +159,7 @@
         },
         error_code=ILLEGAL_OPERATION_ERROR,
         msg="killAllSessions should reject afterClusterTime in readConcern",
+        marks=(pytest.mark.requires(cluster_read_concern=False),),
     ),
 ]
 
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/getDefaultRWConcern/test_smoke_getDefaultRWConcern.py b/documentdb_tests/compatibility/tests/system/administration/commands/getDefaultRWConcern/test_smoke_getDefaultRWConcern.py
index 57af1100e..31865d0d7 100644
--- a/documentdb_tests/compatibility/tests/system/administration/commands/getDefaultRWConcern/test_smoke_getDefaultRWConcern.py
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/getDefaultRWConcern/test_smoke_getDefaultRWConcern.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_getDefaultRWConcern(collection):
     """Test basic getDefaultRWConcern behavior."""
     result = execute_admin_command(collection, {"getDefaultRWConcern": 1})
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setDefaultRWConcern/test_smoke_setDefaultRWConcern.py b/documentdb_tests/compatibility/tests/system/administration/commands/setDefaultRWConcern/test_smoke_setDefaultRWConcern.py
index 71e5eedc4..8d7866f1e 100644
--- a/documentdb_tests/compatibility/tests/system/administration/commands/setDefaultRWConcern/test_smoke_setDefaultRWConcern.py
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/setDefaultRWConcern/test_smoke_setDefaultRWConcern.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_setDefaultRWConcern(collection):
     """Test basic setDefaultRWConcern behavior."""
     result = execute_admin_command(
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_smoke_setUserWriteBlockMode.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_smoke_setUserWriteBlockMode.py
index 9974d06d8..671e255eb 100644
--- a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_smoke_setUserWriteBlockMode.py
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_smoke_setUserWriteBlockMode.py
@@ -12,7 +12,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(cluster_admin=True)
 def test_smoke_setUserWriteBlockMode(collection):
     """Test basic setUserWriteBlockMode behavior."""
     result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False})
diff --git a/documentdb_tests/compatibility/tests/system/security/encryption/test_smoke_encryption.py b/documentdb_tests/compatibility/tests/system/security/encryption/test_smoke_encryption.py
index 68e7cfd27..ea4f8282e 100644
--- a/documentdb_tests/compatibility/tests/system/security/encryption/test_smoke_encryption.py
+++ b/documentdb_tests/compatibility/tests/system/security/encryption/test_smoke_encryption.py
@@ -15,7 +15,7 @@
 pytestmark = pytest.mark.smoke
 
 
-@pytest.mark.replica_set
+@pytest.mark.requires(queryable_encryption=True)
 def test_smoke_encryption(collection):
     """Test basic encryption behavior."""
     result = execute_command(
diff --git a/documentdb_tests/conftest.py b/documentdb_tests/conftest.py
index b88a99733..90bf88938 100644
--- a/documentdb_tests/conftest.py
+++ b/documentdb_tests/conftest.py
@@ -7,6 +7,8 @@
 - Test isolation
 """
 
+from __future__ import annotations
+
 import pytest
 
 # Enable assertion rewriting BEFORE importing framework modules
@@ -15,9 +17,21 @@
 from pathlib import Path  # noqa: E402
 
 from documentdb_tests.framework import fixtures  # noqa: E402
+from documentdb_tests.framework.engine_registry import (  # noqa: E402
+    Target,
+    ensure_initiated,
+    live_targets,
+)
 from documentdb_tests.framework.error_codes_validator import (  # noqa: E402
     validate_error_codes_sorted,
 )
+from documentdb_tests.framework.preconditions import (  # noqa: E402
+    REQUIRES_MARKER,
+    detect_capabilities,
+    known_engines,
+    marker_spec,
+    unmet_requirements,
+)
 from documentdb_tests.framework.test_format_validator import validate_test_format  # noqa: E402
 from documentdb_tests.framework.test_structure_validator import (  # noqa: E402
     validate_python_files_in_tests,
@@ -39,63 +53,153 @@ def pytest_addoption(parser):
         default="default",
         help="Optional engine identifier for metadata. " "Example: --engine-name documentdb",
     )
+    parser.addoption(
+        "--run-crash-tests",
+        action="store_true",
+        default=False,
+        help="Run tests marked engine_xcrash against the engine they crash. They "
+        "are skipped by default because they kill the server; enable this only "
+        "in an isolated job that can tolerate (and expects) a server crash.",
+    )
 
 
 def pytest_configure(config):
-    """Configure pytest with custom settings."""
-    # Get connection string and engine name
+    """Configure pytest with custom settings.
+
+    Resolves the set of test targets the session will run against and stores it
+    on the config as ``test_targets``:
+
+    - If ``--connection-string`` is given, it pins a single ad-hoc target using
+      that string and ``--engine-name`` (used by CI and for pointing at an
+      arbitrary instance not in the registry). Discovery is bypassed.
+    - Otherwise the live targets from the dev compose registry are discovered
+      and the session runs against each (the zero-config local workflow).
+
+    Tests are parametrized over these targets in ``pytest_generate_tests``.
+    """
     connection_string = config.getoption("--connection-string")
     engine_name = config.getoption("--engine-name")
 
-    # Store in config for access by fixtures
-    config.connection_string = connection_string
-    config.engine_name = engine_name
+    if connection_string:
+        # Explicit override: a single pinned target, discovery bypassed. The
+        # engine must be a known one so its preconditions can be resolved the
+        # same way as for an auto-discovered target.
+        if engine_name not in known_engines():
+            raise pytest.UsageError(
+                f"--engine-name must be one of {sorted(known_engines())} when "
+                f"--connection-string is given, got {engine_name!r}"
+            )
+        config.test_targets = [
+            Target(
+                name=engine_name,
+                engine=engine_name,
+                connection_string=connection_string,
+            )
+        ]
+    else:
+        # Zero-config: run against whichever registered targets are live.
+        config.test_targets = live_targets()
+
+    # A target started as a replica set member accepts connections (so it is
+    # discovered as live) but is not usable until the set is initiated. Initiate
+    # it here, once, before tests run: idempotent, and a no-op for an
+    # already-initiated set or a standalone server. Under xdist this runs on the
+    # controller before workers fork, so there is no cross-worker race.
+    #
+    # Collection must work without a live server, so skip initiation when only
+    # collecting. An unreachable target at run time surfaces as a per-test
+    # connection error via the engine_client fixture.
+    if not hasattr(config, "workerinput") and not config.option.collectonly:
+        for target in config.test_targets:
+            ensure_initiated(target.connection_string)
+
+    # Register the requires marker from its single source of truth so it need
+    # not be duplicated in the pytest configuration file.
+    config.addinivalue_line("markers", marker_spec())
+
+
+def pytest_generate_tests(metafunc):
+    """Parametrize tests over the resolved test targets.
+
+    Every test that reaches a live engine does so through the ``engine_client``
+    fixture, so parametrizing that fixture (indirectly) fans each test out into
+    one instance per target. The target name is the parametrization id, so a
+    failure is reported against the specific target it occurred on.
+    """
+    if "engine_client" not in metafunc.fixturenames:
+        return
+    targets = getattr(metafunc.config, "test_targets", [])
+    metafunc.parametrize(
+        "engine_client",
+        targets,
+        ids=[t.name for t in targets],
+        indirect=True,
+    )
 
-    # If no connection string specified, default to localhost
-    if not connection_string:
-        config.connection_string = "mongodb://localhost:27017"
+
+def _item_target(item) -> Target | None:
+    """Return the Target a parametrized item is bound to, or None.
+
+    Each test is parametrized over the session's targets via the indirect
+    ``engine_client`` param, so the bound Target is the ``engine_client`` value
+    in the item's callspec.
+    """
+    callspec = getattr(item, "callspec", None)
+    if callspec is None:
+        return None
+    target = callspec.params.get("engine_client")
+    # Only a real Target counts; other param types (or pytest's NOTSET sentinel
+    # for tests not parametrized over engine_client) are not targets.
+    return target if isinstance(target, Target) else None
 
 
 def pytest_runtest_setup(item):
-    """Apply engine-specific xfail and xcrash markers."""
+    """Apply engine-specific xfail and xcrash markers for the item's target."""
+    target = _item_target(item)
+    engine = target.engine if target is not None else None
     for marker in item.iter_markers("engine_xfail"):
-        if getattr(item.config, "engine_name", None) == marker.kwargs.get("engine"):
+        if engine == marker.kwargs.get("engine"):
             item.add_marker(
                 pytest.mark.xfail(
                     reason=marker.kwargs.get("reason", ""),
                     raises=marker.kwargs.get("raises", AssertionError),
                 )
             )
+    # A crash test kills the server, so it is skipped against the engine it
+    # crashes unless the run-crash-tests option opts in. The dedicated crash job
+    # sets it and runs each such test in isolation against a server it can lose.
+    run_crash_tests = item.config.getoption("--run-crash-tests")
     for marker in item.iter_markers("engine_xcrash"):
-        if getattr(item.config, "engine_name", None) == marker.kwargs.get("engine"):
+        if engine == marker.kwargs.get("engine") and not run_crash_tests:
             pytest.skip(marker.kwargs.get("reason", "crashes the server"))
 
 
 @pytest.fixture(scope="session")
 def engine_client(request):
     """
-    Create a MongoDB client for the configured engine.
+    Create a database client for the test's target engine.
 
-    Session-scoped for performance - MongoClient is thread-safe and maintains
-    an internal connection pool. This significantly improves test execution speed
-    by eliminating redundant connection overhead.
+    The target is supplied indirectly by ``pytest_generate_tests``, which
+    parametrizes this fixture over the session's resolved targets. Session-scoped
+    for performance: pytest creates one client per distinct target and shares it
+    across the session. The client is thread-safe and pools connections, so this
+    avoids redundant connection overhead.
 
-    Per-test isolation is maintained through database_client and collection fixtures
-    which create unique databases/collections for each test.
+    Per-test isolation is maintained through the database_client and collection
+    fixtures, which create unique databases/collections for each test.
 
     Args:
-        request: pytest request object
+        request: pytest request object; ``request.param`` is the Target.
 
     Yields:
-        MongoClient: Connected MongoDB client (shared across session)
+        MongoClient: Connected client for the target (shared across session).
 
     Raises:
         ConnectionError: If unable to connect to the database
     """
-    connection_string = request.config.connection_string
-    engine_name = request.config.engine_name
+    target: Target = request.param
 
-    client = fixtures.create_engine_client(connection_string, engine_name)
+    client = fixtures.create_engine_client(target.connection_string, target.engine)
 
     yield client
 
@@ -103,6 +207,32 @@ def engine_client(request):
     client.close()
 
 
+@pytest.fixture
+def engine_name(request) -> str:
+    """Return the engine name of the target the current test runs against.
+
+    Tests are parametrized over targets via the indirect ``engine_client``
+    fixture, so the engine is per-target. Tests that gate on the engine (e.g.
+    skip unless a specific engine) read this fixture.
+    """
+    target = _item_target(request.node)
+    assert target is not None, "engine_name requires a test parametrized over a target"
+    return target.engine
+
+
+@pytest.fixture
+def connection_string(request) -> str:
+    """Return the connection string of the target the current test runs against.
+
+    Tests are parametrized over targets via the indirect ``engine_client``
+    fixture, so the connection string is per-target. Tests that open an
+    additional connection to the same target read this fixture.
+    """
+    target = _item_target(request.node)
+    assert target is not None, "connection_string requires a test parametrized over a target"
+    return target.connection_string
+
+
 @pytest.fixture(scope="session")
 def worker_id(request):
     """
@@ -206,26 +336,38 @@ def pytest_collection_modifyitems(session, config, items):
         pytest -m no_parallel -p no:xdist
     Or run them manually with: pytest -m no_parallel -p no:xdist
 
-    Tests marked 'replica_set' are skipped when the server is not a replica set member.
+    Tests carrying a ``requires`` marker are deselected when their target's
+    capabilities do not match what the test requires, so they do not run against
+    a target they do not apply to (rather than appearing as skips). A target's
+    capabilities are determined by its engine and topology, resolved per target
+    at runtime (see ``framework.preconditions``).
     """
-    # Skip replica_set tests when not connected to a replica set
-    conn_str = getattr(config, "connection_string", "") or ""
-    try:
-        from pymongo import MongoClient
-
-        client = MongoClient(conn_str, serverSelectionTimeoutMS=5000, directConnection=True)
-        is_replica_set = bool(client.admin.command("hello").get("setName"))
-        client.close()
-    except Exception:
-        is_replica_set = False
-    if not is_replica_set:
-        for item in items:
-            if item.get_closest_marker("replica_set"):
-                item.add_marker(
-                    pytest.mark.skip(
-                        reason="requires replica set " "(server is not a replica set member)"
-                    )
-                )
+    # Deselect a capability-gated test when its target's capabilities do not
+    # match its requires(...) marker. Each item is parametrized over a target;
+    # probe each distinct target once.
+    capabilities_by_target: dict[str, frozenset[str]] = {}
+    kept: list = []
+    requires_deselected: list = []
+    for item in items:
+        marker = item.get_closest_marker(REQUIRES_MARKER)
+        if marker is None or not marker.kwargs:
+            kept.append(item)
+            continue
+        target = _item_target(item)
+        if target is None:
+            kept.append(item)
+            continue
+        capabilities = capabilities_by_target.get(target.connection_string)
+        if capabilities is None:
+            capabilities = detect_capabilities(target.engine, target.connection_string)
+            capabilities_by_target[target.connection_string] = capabilities
+        if unmet_requirements(marker.kwargs, capabilities):
+            requires_deselected.append(item)
+        else:
+            kept.append(item)
+    if requires_deselected:
+        config.hook.pytest_deselected(items=requires_deselected)
+        items[:] = kept
 
     # Deselect no_parallel tests when running under xdist
     is_xdist = bool(getattr(config.option, "numprocesses", None)) or hasattr(config, "workerinput")
@@ -372,8 +514,13 @@ def pytest_sessionfinish(session, exitstatus):
         cmd.extend(["-m", f"no_parallel and ({user_marker})"])
     else:
         cmd.extend(["-m", "no_parallel"])
-    cmd.extend(["--connection-string", config.connection_string])
-    cmd.extend(["--engine-name", config.engine_name])
+    # Pass through the engine selection so Phase 2 targets the same engines as
+    # Phase 1. If an explicit connection string was given (override / CI), pass
+    # it through; otherwise Phase 2 re-discovers live targets the same way.
+    override_conn = config.getoption("--connection-string")
+    if override_conn:
+        cmd.extend(["--connection-string", override_conn])
+        cmd.extend(["--engine-name", config.getoption("--engine-name")])
 
     # Detect Phase 1 report paths and set up Phase 2 temp report files
     phase1_json = getattr(config.option, "json_report_file", None)
diff --git a/documentdb_tests/framework/assertions.py b/documentdb_tests/framework/assertions.py
index bac94a5bf..b4326463e 100644
--- a/documentdb_tests/framework/assertions.py
+++ b/documentdb_tests/framework/assertions.py
@@ -15,6 +15,32 @@
 
 _MAX_REPR_LEN = 1000
 
+# Top-level fields that a replica set / sharded topology appends to command
+# responses as cluster and replication gossip. They appear on the connected
+# server's responses regardless of the command and are never the subject of a
+# compatibility test, so they are stripped from a raw command result before
+# comparison. This keeps assertions exact across topologies: on a standalone
+# server these fields are absent (stripping is a no-op), and on a replica set
+# they no longer cause spurious mismatches.
+#
+# Stripping is TOP-LEVEL ONLY and limited to this fixed, audited set. Behavioral
+# fields that only appear on a replica set (e.g. createIndexes' commitQuorum) are
+# NOT included and remain asserted, as does any nested occurrence of these names
+# (e.g. the opTime nested in a hello response), so topology-specific behavior
+# stays testable.
+_REPLICATION_GOSSIP_FIELDS = frozenset({"$clusterTime", "operationTime", "electionId", "opTime"})
+
+
+def _strip_replication_gossip(result: Any) -> Any:
+    """Remove top-level replication/cluster gossip fields from a raw result.
+
+    Only the fixed ``_REPLICATION_GOSSIP_FIELDS`` are removed, and only at the
+    top level of a dict result. Non-dict results are returned unchanged.
+    """
+    if not isinstance(result, dict):
+        return result
+    return {k: v for k, v in result.items() if k not in _REPLICATION_GOSSIP_FIELDS}
+
 
 def _truncate_repr(obj: Any) -> str:
     """Format an object for error output, truncating if too long."""
@@ -146,6 +172,14 @@ def assertSuccess(
 
     if not raw_res:
         result = result["cursor"]["firstBatch"]
+    else:
+        # Raw command result: drop replica-set/cluster gossip fields so the
+        # comparison stays exact across topologies. Strip the expected side too,
+        # since some tests pass another raw command result as the expected value
+        # (e.g. consistency checks comparing two responses).
+        result = _strip_replication_gossip(result)
+        if isinstance(expected, dict):
+            expected = _strip_replication_gossip(expected)
 
     if transform:
         result = transform(result)
diff --git a/documentdb_tests/framework/ci_matrix.py b/documentdb_tests/framework/ci_matrix.py
new file mode 100644
index 000000000..159470bd4
--- /dev/null
+++ b/documentdb_tests/framework/ci_matrix.py
@@ -0,0 +1,63 @@
+"""Emit the CI test matrix from the compose file.
+
+The CI workflow runs the suite once per test target. Rather than duplicating the
+target list (ports, profiles, connection strings) in the workflow YAML, this
+module derives the matrix from ``dev/compose.yaml`` — the same single source of
+truth the test harness reads via :mod:`documentdb_tests.framework.engine_registry`.
+
+Run as ``python -m documentdb_tests.framework.ci_matrix`` to print a JSON array
+of target objects, one per target::
+
+    [{"name": ..., "profile": ..., "connection_string": ..., "engine": ...}, ...]
+
+Each target's compose profile is the service's first declared profile, which by
+convention equals the service (target) name.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import yaml
+
+from documentdb_tests.framework.engine_registry import COMPOSE_PATH, load_targets
+
+
+def _profile_for(service_name: str, compose_path: Path) -> str:
+    """Return the target-specific compose profile for a service.
+
+    A target service declares ``profiles: [<target>, "all"]``; the first entry
+    is the profile that brings up just that target. It equals the service name
+    by convention, but is read from the file so the convention lives in one
+    place (the compose file) rather than being assumed here.
+    """
+    document = yaml.safe_load(compose_path.read_text())
+    profiles = document["services"][service_name].get("profiles") or [service_name]
+    return str(profiles[0])
+
+
+def build_matrix(compose_path: Path = COMPOSE_PATH) -> list[dict[str, str]]:
+    """Return the CI matrix entries for every declared test target."""
+    return [
+        {
+            "name": t.name,
+            "profile": _profile_for(t.name, compose_path),
+            "connection_string": t.connection_string,
+            "engine": t.engine,
+        }
+        for t in load_targets(compose_path)
+    ]
+
+
+if __name__ == "__main__":
+    import sys
+
+    matrix = build_matrix()
+    if not matrix:
+        # The compose file declares no test targets. The suite has nothing to
+        # run against, which is a misconfiguration; exit non-zero so CI fails
+        # loudly instead of producing an empty matrix that silently skips jobs.
+        print("no test targets found in the compose file", file=sys.stderr)
+        sys.exit(1)
+    print(json.dumps(matrix))
diff --git a/documentdb_tests/framework/engine_registry.py b/documentdb_tests/framework/engine_registry.py
new file mode 100644
index 000000000..774ab7810
--- /dev/null
+++ b/documentdb_tests/framework/engine_registry.py
@@ -0,0 +1,162 @@
+"""Test-target registry derived from the dev compose file.
+
+``dev/compose.yaml`` is the single source of truth for the local database
+targets. Each runnable service carries an ``x-test-target`` block describing how
+the test harness should reach it:
+
+    x-test-target:
+      engine: <engine name for engine_xfail / engine_xcrash markers>
+      query: <optional connection-string query string, without leading '?'>
+
+The host port comes from the service's ``ports:`` mapping, so it is not
+duplicated here. This module reads that file and exposes the targets so the test
+session can discover which are live and run against each, without any port or
+connection-string information living in two places.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import yaml
+from pymongo import MongoClient
+from pymongo.errors import OperationFailure
+
+# dev/compose.yaml relative to the repository root (two levels up from this
+# module's package: documentdb_tests/framework/ -> repo root).
+COMPOSE_PATH = Path(__file__).resolve().parents[2] / "dev" / "compose.yaml"
+
+
+@dataclass(frozen=True)
+class Target:
+    """A test target read from the compose file.
+
+    Attributes:
+        name: The compose service name, used as the target / parametrization id.
+        engine: Engine name reported to the harness, matched by engine_xfail and
+            engine_xcrash markers.
+        connection_string: Full connection string built from the host port and
+            optional query suffix.
+    """
+
+    name: str
+    engine: str
+    connection_string: str
+
+
+def _host_port(service: dict) -> str | None:
+    """Return the published host port from a service's first ``ports`` entry."""
+    ports = service.get("ports") or []
+    if not ports:
+        return None
+    # Entries look like "27017:27017"; the host port is the left side.
+    return str(ports[0]).split(":")[0]
+
+
+def load_targets(compose_path: Path = COMPOSE_PATH) -> list[Target]:
+    """Parse the compose file and return every declared test target.
+
+    A service is a target only if it carries an ``x-test-target`` block and
+    publishes a host port. Services without that block (e.g. one-shot init
+    containers) are ignored.
+    """
+    document = yaml.safe_load(compose_path.read_text())
+    targets: list[Target] = []
+    for name, service in (document.get("services") or {}).items():
+        spec = service.get("x-test-target")
+        if not spec:
+            continue
+        port = _host_port(service)
+        if port is None:
+            continue
+        query = spec.get("query")
+        connection_string = f"mongodb://localhost:{port}"
+        if query:
+            connection_string += f"/?{query}"
+        targets.append(
+            Target(name=name, engine=spec["engine"], connection_string=connection_string)
+        )
+    return targets
+
+
+def _is_reachable(connection_string: str) -> bool:
+    """Return whether a server accepts a connection within a short timeout."""
+    try:
+        client: MongoClient = MongoClient(connection_string, serverSelectionTimeoutMS=2000)
+    except Exception:
+        return False
+    try:
+        client.admin.command("ping")
+        return True
+    except Exception:
+        return False
+    finally:
+        client.close()
+
+
+# replSetGetStatus error code when the server is a replica set member that has
+# not been initiated yet; the only case in which the harness initiates.
+_NOT_YET_INITIALIZED = 94
+# replSetInitiate error code when the set is already initiated (e.g. a race
+# between concurrent callers); treated as success.
+_ALREADY_INITIALIZED = 23
+
+
+def ensure_initiated(connection_string: str, timeout_s: float = 30.0) -> None:
+    """Idempotently initiate a single-node replica set, if the target is one.
+
+    A server started with ``--replSet`` accepts connections before the set is
+    initiated, but is not usable until it is. This brings such a server up to a
+    writable primary. It is safe to call against any target:
+
+    - An already-initiated replica set: ``replSetGetStatus`` succeeds, so nothing
+      is done.
+    - A standalone server: ``replSetGetStatus`` fails with a code other than
+      NotYetInitialized (replication is not enabled), so nothing is done.
+    - An uninitiated ``--replSet`` server: ``replSetGetStatus`` fails with
+      NotYetInitialized, so ``replSetInitiate`` is issued; a concurrent caller
+      that already initiated it (AlreadyInitialized) is tolerated.
+
+    After initiating, it waits up to ``timeout_s`` for a primary to be elected
+    so callers can write immediately.
+    """
+    client: MongoClient = MongoClient(connection_string, serverSelectionTimeoutMS=5000)
+    try:
+        try:
+            client.admin.command("replSetGetStatus")
+            return  # Already initiated.
+        except OperationFailure as exc:
+            if exc.code != _NOT_YET_INITIALIZED:
+                return  # Not an uninitiated replica set (e.g. a standalone).
+
+        try:
+            client.admin.command("replSetInitiate")
+        except OperationFailure as exc:
+            if exc.code != _ALREADY_INITIALIZED:
+                raise
+
+        deadline = time.monotonic() + timeout_s
+        while time.monotonic() < deadline:
+            if client.admin.command("hello").get("isWritablePrimary"):
+                return
+            time.sleep(0.5)
+        raise TimeoutError(
+            f"replica set at {connection_string} did not elect a primary within {timeout_s}s"
+        )
+    finally:
+        client.close()
+
+
+def live_targets(compose_path: Path = COMPOSE_PATH) -> list[Target]:
+    """Return the declared targets that are currently reachable."""
+    return [t for t in load_targets(compose_path) if _is_reachable(t.connection_string)]
+
+
+if __name__ == "__main__":
+    # Initiate every reachable target's replica set, if any. Useful before a
+    # collection-only run (which does not initiate on its own) so that topology
+    # detection sees a usable server. A no-op for standalone targets.
+    for _target in live_targets():
+        ensure_initiated(_target.connection_string)
diff --git a/documentdb_tests/framework/error_codes.py b/documentdb_tests/framework/error_codes.py
index 2375b9dcd..86884ce4c 100644
--- a/documentdb_tests/framework/error_codes.py
+++ b/documentdb_tests/framework/error_codes.py
@@ -49,6 +49,7 @@
 QUERY_FEATURE_NOT_ALLOWED = 224
 MAX_NESTED_SUB_PIPELINE_ERROR = 232
 CONVERSION_FAILURE_ERROR = 241
+OPERATION_NOT_SUPPORTED_IN_TRANSACTION_ERROR = 263
 NO_QUERY_EXECUTION_PLANS_ERROR = 291
 QUERY_EXCEEDED_MEMORY_NO_DISK_USE_ERROR = 292
 API_VERSION_ERROR = 322
diff --git a/documentdb_tests/framework/executor.py b/documentdb_tests/framework/executor.py
index 4df8464aa..ba3779533 100644
--- a/documentdb_tests/framework/executor.py
+++ b/documentdb_tests/framework/executor.py
@@ -7,7 +7,7 @@
 
 from bson.codec_options import CodecOptions
 
-TZ_AWARE_CODEC = CodecOptions(tz_aware=True, tzinfo=timezone.utc)
+TZ_AWARE_CODEC: CodecOptions = CodecOptions(tz_aware=True, tzinfo=timezone.utc)
 
 
 def execute_command(collection, command: Dict, codec_options=TZ_AWARE_CODEC, session=None) -> Any:
diff --git a/documentdb_tests/framework/fixtures.py b/documentdb_tests/framework/fixtures.py
index 00b5bec72..fdb10bb7b 100644
--- a/documentdb_tests/framework/fixtures.py
+++ b/documentdb_tests/framework/fixtures.py
@@ -24,7 +24,7 @@ def create_engine_client(connection_string: str, engine_name: str = "default"):
     Raises:
         ConnectionError: If unable to connect to the database
     """
-    client = MongoClient(connection_string)
+    client: MongoClient = MongoClient(connection_string)
 
     # Verify connection
     try:
diff --git a/documentdb_tests/framework/preconditions.py b/documentdb_tests/framework/preconditions.py
new file mode 100644
index 000000000..079b17a2b
--- /dev/null
+++ b/documentdb_tests/framework/preconditions.py
@@ -0,0 +1,184 @@
+"""Capability resolution for environment-gated tests.
+
+A test case may apply only to targets that have (or lack) a particular
+capability. It declares this with the ``requires`` marker, naming capabilities
+and whether each must be present::
+
+    @pytest.mark.requires(change_streams=True)     # only where change streams exist
+    @pytest.mark.requires(change_streams=False)    # only where they do not
+    @pytest.mark.requires(transactions=True, cluster_admin=True)
+
+A capability is any named fact about a target -- a feature it provides (change
+streams, transactions) or a behavior it exhibits (compact succeeding without
+``force``). The test is skipped against a target whose capabilities do not match
+what it requires.
+
+For a given engine, which capabilities a target has is fully determined by its
+topology (e.g. a standalone server vs a replica set). The engine is always known
+(it is declared per target) and the topology is detected once from a live
+connection. The pair ``(engine, topology)`` then selects the present
+capabilities from a static table, so there is a single declarative source of
+truth for every engine and topology and the manual-connection path resolves
+identically to auto-discovered targets.
+
+The expected behavior of a case is the reference engine's behavior, applied
+uniformly to all engines (a compatible engine is measured against it, never
+defining its own). Adding an engine or a new topology (e.g. a search-backed
+deployment) is a new row in ``_CAPABILITIES_BY_PROFILE`` plus teaching
+``_detect_topology`` to recognize it.
+"""
+
+from __future__ import annotations
+
+from pymongo import MongoClient
+
+# The capabilities the harness knows about, each mapped to a human-readable
+# description. A capability is satisfied on the profiles listed for it in
+# ``_CAPABILITIES_BY_PROFILE``. Every capability named there must appear here and
+# vice versa (enforced by _check_consistency at import time).
+_CAPABILITY_DESCRIPTIONS: dict[str, str] = {
+    "change_streams": "change streams are available",
+    "transactions": "multi-document transactions are available",
+    "queryable_encryption": "encryptedFields / queryable encryption is available",
+    "cluster_admin": (
+        "cluster-wide admin features are available (query settings, default RW concern, "
+        "user write-block mode, query sampling)"
+    ),
+    "cluster_time": "$$CLUSTER_TIME resolves rather than being unavailable",
+    "cluster_read_concern": (
+        "replication-dependent read concern (afterClusterTime, linearizable, snapshot) "
+        "is accepted rather than rejected"
+    ),
+    "quorum_write_concern": (
+        "a quorum write concern is accepted (reported as a writeConcernError) rather than "
+        "rejected up front"
+    ),
+    "unforced_compact": "compact succeeds without force",
+    "reindex": "reIndex is permitted",
+    "local_rename": "renaming into the unreplicated local database is permitted",
+}
+
+# The capabilities each (engine, topology) target has. To add an engine or
+# topology, add an entry here; every test then gates correctly.
+_CAPABILITIES_BY_PROFILE: dict[tuple[str, str], frozenset[str]] = {
+    ("mongodb", "replica_set"): frozenset(
+        {
+            "change_streams",
+            "transactions",
+            "queryable_encryption",
+            "cluster_admin",
+            "cluster_time",
+            "cluster_read_concern",
+            "quorum_write_concern",
+        }
+    ),
+    ("mongodb", "standalone"): frozenset(
+        {
+            "unforced_compact",
+            "reindex",
+            "local_rename",
+        }
+    ),
+}
+
+# The single marker tests use to declare capability requirements.
+REQUIRES_MARKER = "requires"
+
+# Every known capability name, for validation of requires(...) kwargs.
+CAPABILITIES = frozenset(_CAPABILITY_DESCRIPTIONS)
+
+
+def _check_consistency() -> None:
+    """Fail at import if the descriptions and the profile table diverge.
+
+    Every capability named in the profile table must have a description, and
+    every described capability must be present on at least one profile, so a
+    capability cannot be referenced without being described or described without
+    ever being satisfiable.
+    """
+    described = set(_CAPABILITY_DESCRIPTIONS)
+    mapped: set[str] = set()
+    for capabilities in _CAPABILITIES_BY_PROFILE.values():
+        mapped |= capabilities
+    unknown = mapped - described
+    if unknown:
+        raise RuntimeError(f"profile table references undescribed capabilities: {sorted(unknown)}")
+    unsatisfiable = described - mapped
+    if unsatisfiable:
+        raise RuntimeError(
+            f"described capabilities never present on any profile: {sorted(unsatisfiable)}"
+        )
+
+
+_check_consistency()
+
+
+def _detect_topology(engine: str, client: MongoClient) -> str:
+    """Classify a live target's topology. The only runtime-observed step.
+
+    Engine-specific: each engine names and recognizes its own topologies.
+    """
+    if engine == "mongodb":
+        # A replica set member reports its set name in ``hello``; a standalone
+        # server does not.
+        if client.admin.command("hello").get("setName"):
+            return "replica_set"
+        return "standalone"
+    raise ValueError(f"unknown engine {engine!r}; cannot classify topology")
+
+
+def marker_spec() -> str:
+    """Return the pytest marker definition line for the ``requires`` marker."""
+    return (
+        "requires(**capabilities): gate a test on capabilities the target must have "
+        "(name=True) or lack (name=False); known names: " + ", ".join(sorted(CAPABILITIES))
+    )
+
+
+def known_engines() -> frozenset[str]:
+    """Return the engine names the capability table knows how to resolve."""
+    return frozenset(engine for engine, _topology in _CAPABILITIES_BY_PROFILE)
+
+
+def detect_capabilities(engine: str, connection_string: str) -> frozenset[str]:
+    """Return the capabilities the target has.
+
+    Detects the target's topology from a live connection, then looks up the
+    capabilities that ``(engine, topology)`` has. On a connection failure the
+    set is empty, so capability-gated tests skip rather than error.
+    """
+    try:
+        client: MongoClient = MongoClient(connection_string, serverSelectionTimeoutMS=5000)
+    except Exception:
+        return frozenset()
+
+    try:
+        topology = _detect_topology(engine, client)
+    except Exception:
+        return frozenset()
+    finally:
+        client.close()
+
+    profile = (engine, topology)
+    if profile not in _CAPABILITIES_BY_PROFILE:
+        raise RuntimeError(
+            f"no capability mapping for target profile {profile}; "
+            "add it to _CAPABILITIES_BY_PROFILE"
+        )
+    return _CAPABILITIES_BY_PROFILE[profile]
+
+
+def unmet_requirements(required: dict[str, bool], capabilities: frozenset[str]) -> dict[str, bool]:
+    """Return the subset of ``required`` the target's capabilities do not meet.
+
+    ``required`` maps a capability name to whether the test needs it present
+    (True) or absent (False). A requirement is unmet when the capability's
+    presence does not match. Unknown capability names raise, so a typo in a
+    ``requires(...)`` marker fails loudly rather than silently never gating.
+    """
+    unknown = set(required) - CAPABILITIES
+    if unknown:
+        raise RuntimeError(f"requires(...) names unknown capabilities: {sorted(unknown)}")
+    return {
+        name: expected for name, expected in required.items() if (name in capabilities) != expected
+    }
diff --git a/documentdb_tests/pytest.ini b/documentdb_tests/pytest.ini
index de4799073..b2cd6c4a1 100644
--- a/documentdb_tests/pytest.ini
+++ b/documentdb_tests/pytest.ini
@@ -40,11 +40,13 @@ markers =
     # Special markers
     smoke: Quick smoke tests for feature detection
     slow: Tests that take longer to execute
-    replica: Tests that can only run on a replica
+    unit: Fast unit tests with no external dependencies
     engine_xfail(engine, reason, raises): expected failure for a specific engine
     engine_xcrash(engine, reason): test crashes the server on a specific engine
     no_parallel: Tests that must run sequentially (not in parallel)
-    replica_set: Tests that need to run on replica set
+
+    # The requires(...) marker is registered programmatically from
+    # documentdb_tests/framework/preconditions.py — its single source of truth.
 
 # Timeout for tests (seconds)
 timeout = 300
diff --git a/requirements-dev.txt b/requirements-dev.txt
index b97e2e76c..17e2c9453 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -5,6 +5,7 @@
 black>=23.7.0  # Code formatting
 flake8>=6.1.0  # Linting
 mypy>=1.5.0  # Type checking
+types-PyYAML>=6.0  # Type stubs for PyYAML (no inline types; ships separately)
 isort>=5.12.0  # Import sorting
 
 # Testing