diff --git a/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_compound_predicates.py b/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_compound_predicates.py
new file mode 100644
index 000000000..e5e72f833
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_compound_predicates.py
@@ -0,0 +1,95 @@
+"""
+$text query operator combined with non-text query predicates (implicit and
+explicit $and).
+
+Existing $text coverage exercises the operator in isolation and with a single
+co-located equality predicate. This file covers a richer set of compound
+filters: $text intersected with an equality, a range ($gt), an `$in`, an array
+equality, a `$ne`, an explicit `$and`, and a predicate that excludes every text
+match. In every case the result is the intersection of the text match and the
+scalar predicate.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline). The engine under test
+matches native behavior on every case; no engine divergences are tracked here.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.operator.query.utils.query_test_case import (
+    QueryTestCase,
+)
+from documentdb_tests.framework.assertions import assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+DOCS = [
+    {"_id": 1, "content": "coffee and tea", "category": "drinks", "rating": 5, "tags": ["hot"]},
+    {"_id": 2, "content": "coffee beans roasted", "category": "food", "rating": 3, "tags": ["beans"]},
+    {"_id": 3, "content": "green tea leaves", "category": "drinks", "rating": 4, "tags": ["green"]},
+    {"_id": 4, "content": "python programming", "category": "tech", "rating": 5, "tags": ["code"]},
+]
+
+# Property [Compound Intersection]: $text composes with non-text predicates as a
+# conjunction; only documents matching both the text search and the scalar
+# predicate are returned.
+TEXT_COMPOUND_TESTS: list[QueryTestCase] = [
+    QueryTestCase(
+        id="text_and_equality",
+        filter={"$text": {"$search": "coffee"}, "category": "drinks"},
+        expected=[{"_id": 1}],
+        msg="$text intersected with an equality predicate returns the common match.",
+    ),
+    QueryTestCase(
+        id="text_and_range_gt",
+        filter={"$text": {"$search": "coffee"}, "rating": {"$gt": 4}},
+        expected=[{"_id": 1}],
+        msg="$text intersected with a $gt range keeps only the high-rated match.",
+    ),
+    QueryTestCase(
+        id="text_or_terms_and_in",
+        filter={"$text": {"$search": "coffee tea"}, "category": {"$in": ["drinks"]}},
+        expected=[{"_id": 1}, {"_id": 3}],
+        msg="$text OR-of-terms intersected with an $in keeps the drinks documents.",
+    ),
+    QueryTestCase(
+        id="text_explicit_and_with_range",
+        filter={"$and": [{"$text": {"$search": "coffee"}}, {"rating": {"$gte": 3}}]},
+        expected=[{"_id": 1}, {"_id": 2}],
+        msg="$text inside an explicit $and intersects with a $gte range predicate.",
+    ),
+    QueryTestCase(
+        id="text_and_array_equality",
+        filter={"$text": {"$search": "coffee"}, "tags": "beans"},
+        expected=[{"_id": 2}],
+        msg="$text intersected with an array-membership equality returns the match.",
+    ),
+    QueryTestCase(
+        id="text_and_not_equal",
+        filter={"$text": {"$search": "tea"}, "category": {"$ne": "food"}},
+        expected=[{"_id": 1}, {"_id": 3}],
+        msg="$text intersected with a $ne predicate excludes the food document.",
+    ),
+    QueryTestCase(
+        id="text_and_predicate_excludes_all",
+        filter={"$text": {"$search": "coffee"}, "category": "tech"},
+        expected=[],
+        msg="When the scalar predicate excludes every text match the result is empty.",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(TEXT_COMPOUND_TESTS))
+def test_text_compound_predicates(collection, test: QueryTestCase):
+    """$text intersects with co-located non-text predicates as a conjunction."""
+    collection.create_index([("content", "text")])
+    collection.insert_many([dict(d) for d in DOCS])
+    result = execute_command(
+        collection,
+        {
+            "find": collection.name,
+            "filter": test.filter,
+            "projection": {"_id": 1},
+            "sort": {"_id": 1},
+        },
+    )
+    assertSuccess(result, test.expected, msg=test.msg)
diff --git a/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_meta_score_placement.py b/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_meta_score_placement.py
new file mode 100644
index 000000000..77f3615aa
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/query/text/test_text_meta_score_placement.py
@@ -0,0 +1,117 @@
+"""
+Placement and validation rules for `$meta: "textScore"` with the $text operator.
+
+Existing coverage asserts that a projected textScore is returned and that
+results can be ordered by it. This file covers the placement contract around
+the metadata: the score may be sorted on without being projected, it ranks
+documents by match frequency (assertions are on ordering, never on the
+engine-specific score value), and requesting the score in a projection or a
+sort without any `$text` query in the filter is rejected with the documented
+metadata-not-available error.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline). The engine under test
+matches native behavior on every case; no engine divergences are tracked here.
+"""
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertFailureCode, assertProperties, assertSuccess
+from documentdb_tests.framework.error_codes import QUERY_METADATA_NOT_AVAILABLE_ERROR
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.property_checks import Exists, IsType, Len
+
+pytestmark = pytest.mark.find
+
+
+def _create_text_index(collection):
+    collection.create_index([("content", "text")])
+
+
+def test_text_score_projection_returns_double(collection):
+    """A projected textScore is a numeric (double) field on each matched document."""
+    _create_text_index(collection)
+    collection.insert_one({"_id": 1, "content": "coffee and more coffee"})
+    result = execute_command(
+        collection,
+        {
+            "find": collection.name,
+            "filter": {"$text": {"$search": "coffee"}},
+            "projection": {"score": {"$meta": "textScore"}},
+        },
+    )
+    assertProperties(
+        result,
+        {
+            "cursor.firstBatch": Len(1),
+            "cursor.firstBatch.0._id": Exists(),
+            "cursor.firstBatch.0.score": IsType("double"),
+        },
+        raw_res=True,
+        msg="A projected textScore should be a double on the matched document.",
+    )
+
+
+def test_text_score_sort_without_projection_ranks_by_frequency(collection):
+    """Sorting by textScore (without projecting it) orders documents by match frequency."""
+    _create_text_index(collection)
+    collection.insert_many(
+        [
+            {"_id": 1, "content": "coffee"},
+            {"_id": 2, "content": "coffee coffee coffee"},
+            {"_id": 3, "content": "coffee coffee"},
+        ]
+    )
+    result = execute_command(
+        collection,
+        {
+            "find": collection.name,
+            "filter": {"$text": {"$search": "coffee"}},
+            "sort": {"score": {"$meta": "textScore"}},
+            "projection": {"_id": 1},
+        },
+    )
+    # Assert ordering only; the absolute textScore value is engine-specific.
+    assertSuccess(
+        result,
+        [{"_id": 2}, {"_id": 3}, {"_id": 1}],
+        msg="textScore sort orders the most-frequent match first, even unprojected.",
+    )
+
+
+def test_text_score_projection_without_text_query_errors(collection):
+    """Projecting textScore without a $text query fails with the metadata-not-available code."""
+    _create_text_index(collection)
+    collection.insert_one({"_id": 1, "content": "coffee"})
+    result = execute_command(
+        collection,
+        {
+            "find": collection.name,
+            "filter": {},
+            "projection": {"score": {"$meta": "textScore"}},
+        },
+    )
+    assertFailureCode(
+        result,
+        QUERY_METADATA_NOT_AVAILABLE_ERROR,
+        msg="textScore projection requires a $text query in the filter.",
+    )
+
+
+def test_text_score_sort_without_text_query_errors(collection):
+    """Sorting by textScore without a $text query fails with the metadata-not-available code."""
+    _create_text_index(collection)
+    collection.insert_one({"_id": 1, "content": "coffee"})
+    result = execute_command(
+        collection,
+        {
+            "find": collection.name,
+            "filter": {},
+            "sort": {"score": {"$meta": "textScore"}},
+            "projection": {"score": {"$meta": "textScore"}},
+        },
+    )
+    assertFailureCode(
+        result,
+        QUERY_METADATA_NOT_AVAILABLE_ERROR,
+        msg="textScore sort requires a $text query in the filter.",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_merge.py b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_merge.py
new file mode 100644
index 000000000..d25aa7ef0
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_merge.py
@@ -0,0 +1,132 @@
+"""$merge stage — pipeline integration with other stages (composition coverage).
+
+Existing $merge coverage focuses on whenMatched/whenNotMatched semantics, the
+``on`` field, and write-path behavior. This file mirrors the sibling
+``test_stages_combination_out`` / ``test_stages_combination_sort`` pattern for
+$merge: it verifies that $merge correctly consumes the output of a preceding
+stage — $match filters, $project reshapes, $group aggregates into the ``_id``
+key, $sort + $limit selects a top-k subset, $addFields enriches before a
+default whenMatched merge into an existing target, and $unwind + $group
+re-keys before writing.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline). The engine under test
+matches native behavior on every case; no engine divergences are tracked for
+this surface.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.operator.stages.merge.utils.merge_common import (
+    TARGET,
+    MergeTestCase,
+)
+from documentdb_tests.framework.assertions import assertResult
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.aggregate
+
+SOURCE = [
+    {"_id": 1, "g": "a", "val": 10, "status": "on"},
+    {"_id": 2, "g": "b", "val": 20, "status": "off"},
+    {"_id": 3, "g": "a", "val": 30, "status": "on"},
+]
+
+# Property [Pipeline Integration]: $merge writes the output of the preceding
+# stage to the target collection, preserving the transformation that stage
+# produced.
+MERGE_COMBINATION_TESTS: list[MergeTestCase] = [
+    MergeTestCase(
+        "match_then_merge",
+        docs=SOURCE,
+        target_docs=[],
+        pipeline=[{"$match": {"status": "on"}}, {"$merge": {"into": TARGET}}],
+        expected=[
+            {"_id": 1, "g": "a", "val": 10, "status": "on"},
+            {"_id": 3, "g": "a", "val": 30, "status": "on"},
+        ],
+        msg="$merge writes only the documents that pass a preceding $match.",
+    ),
+    MergeTestCase(
+        "project_then_merge",
+        docs=SOURCE,
+        target_docs=[],
+        pipeline=[{"$project": {"val": 1}}, {"$merge": {"into": TARGET}}],
+        expected=[
+            {"_id": 1, "val": 10},
+            {"_id": 2, "val": 20},
+            {"_id": 3, "val": 30},
+        ],
+        msg="$merge writes the reshaped documents produced by a preceding $project.",
+    ),
+    MergeTestCase(
+        "group_then_merge",
+        docs=SOURCE,
+        target_docs=[],
+        pipeline=[
+            {"$group": {"_id": "$g", "total": {"$sum": "$val"}}},
+            {"$merge": {"into": TARGET}},
+        ],
+        expected=[
+            {"_id": "a", "total": 40},
+            {"_id": "b", "total": 20},
+        ],
+        msg="$merge writes $group results keyed by the group _id.",
+    ),
+    MergeTestCase(
+        "sort_limit_then_merge",
+        docs=SOURCE,
+        target_docs=[],
+        pipeline=[
+            {"$sort": {"val": -1}},
+            {"$limit": 2},
+            {"$merge": {"into": TARGET}},
+        ],
+        expected=[
+            {"_id": 2, "g": "b", "val": 20, "status": "off"},
+            {"_id": 3, "g": "a", "val": 30, "status": "on"},
+        ],
+        msg="$merge writes the top-k subset selected by a preceding $sort + $limit.",
+    ),
+    MergeTestCase(
+        "addfields_then_merge_into_existing",
+        docs=SOURCE,
+        target_docs=[{"_id": 1, "note": "kept"}],
+        pipeline=[
+            {"$addFields": {"doubled": {"$multiply": ["$val", 2]}}},
+            {"$merge": {"into": TARGET}},
+        ],
+        expected=[
+            {"_id": 1, "note": "kept", "g": "a", "val": 10, "status": "on", "doubled": 20},
+            {"_id": 2, "g": "b", "val": 20, "status": "off", "doubled": 40},
+            {"_id": 3, "g": "a", "val": 30, "status": "on", "doubled": 60},
+        ],
+        msg="Default whenMatched merge keeps target fields and adds $addFields output.",
+    ),
+    MergeTestCase(
+        "unwind_group_then_merge",
+        docs=[{"_id": 1, "tags": ["x", "y"]}, {"_id": 2, "tags": ["x"]}],
+        target_docs=[],
+        pipeline=[
+            {"$unwind": "$tags"},
+            {"$group": {"_id": "$tags", "n": {"$sum": 1}}},
+            {"$merge": {"into": TARGET}},
+        ],
+        expected=[
+            {"_id": "x", "n": 2},
+            {"_id": "y", "n": 1},
+        ],
+        msg="$merge writes counts produced by $unwind + $group re-keyed by tag.",
+    ),
+]
+
+
+@pytest.mark.parametrize("test_case", pytest_params(MERGE_COMBINATION_TESTS))
+def test_stages_combination_merge(collection, test_case: MergeTestCase):
+    """$merge writes the output of a preceding aggregation stage to the target."""
+    target = test_case.prepare(collection)
+    execute_command(collection, test_case.build_command(collection, target))
+    result = execute_command(collection, {"find": target, "filter": {}, "sort": {"_id": 1}})
+    assertResult(result, expected=test_case.expected, msg=test_case.msg)
diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_out_bucketing.py b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_out_bucketing.py
new file mode 100644
index 000000000..11aedc52d
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_combination_out_bucketing.py
@@ -0,0 +1,159 @@
+"""$out stage — composition with bucketing and window stages.
+
+The sibling ``test_stages_combination_out`` exercises $out after $match,
+$project, $group, $sort/$limit, $unwind, $lookup and friends. This file
+extends that composition coverage to the bucketing and window family that the
+existing file does not touch: $bucket, $sortByCount, $setWindowFields, and
+$bucketAuto.
+
+$bucketAuto before $out is a tracked engine divergence: native MongoDB writes
+the auto-bucketed output, while the engine under test rejects the pipeline
+because the $bucketAuto plan is treated as a mutable function inside $out.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.operator.stages.out.utils.out_test_helpers import (
+    OutTestCase,
+    target_name,
+)
+from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import (
+    populate_collection,
+)
+from documentdb_tests.framework.assertions import assertResult
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.aggregate
+
+# Property [Bucketing / Window Composition]: $out writes the output produced by
+# a preceding bucketing or window stage unchanged.
+OUT_WINDOW_BUCKET_TESTS: list[OutTestCase] = [
+    OutTestCase(
+        "bucket_then_out",
+        docs=[{"_id": i, "x": v} for i, v in enumerate([1, 5, 12, 18, 22, 35], start=1)],
+        pipeline=[
+            {
+                "$bucket": {
+                    "groupBy": "$x",
+                    "boundaries": [0, 10, 20, 40],
+                    "default": "other",
+                    "output": {"count": {"$sum": 1}},
+                }
+            }
+        ],
+        expected=[
+            {"_id": 0, "count": 2},
+            {"_id": 10, "count": 2},
+            {"_id": 20, "count": 2},
+        ],
+        msg="$out writes the per-bucket counts produced by a preceding $bucket.",
+    ),
+    OutTestCase(
+        "sortbycount_then_out",
+        docs=[
+            {"_id": 1, "c": "a"},
+            {"_id": 2, "c": "b"},
+            {"_id": 3, "c": "a"},
+            {"_id": 4, "c": "a"},
+        ],
+        pipeline=[{"$sortByCount": "$c"}],
+        expected=[
+            {"_id": "a", "count": 3},
+            {"_id": "b", "count": 1},
+        ],
+        msg="$out writes the grouped counts produced by a preceding $sortByCount.",
+    ),
+    OutTestCase(
+        "setwindowfields_then_out",
+        docs=[
+            {"_id": 1, "g": "a", "v": 10},
+            {"_id": 2, "g": "a", "v": 20},
+            {"_id": 3, "g": "b", "v": 30},
+        ],
+        pipeline=[
+            {
+                "$setWindowFields": {
+                    "partitionBy": "$g",
+                    "sortBy": {"_id": 1},
+                    "output": {
+                        "running": {
+                            "$sum": "$v",
+                            "window": {"documents": ["unbounded", "current"]},
+                        }
+                    },
+                }
+            }
+        ],
+        expected=[
+            {"_id": 1, "g": "a", "v": 10, "running": 10},
+            {"_id": 2, "g": "a", "v": 20, "running": 30},
+            {"_id": 3, "g": "b", "v": 30, "running": 30},
+        ],
+        msg="$out writes the per-partition running totals from $setWindowFields.",
+    ),
+]
+
+
+@pytest.mark.parametrize("test_case", pytest_params(OUT_WINDOW_BUCKET_TESTS))
+def test_out_window_bucket_composition(collection, test_case: OutTestCase):
+    """$out writes the output of a preceding bucketing or window stage."""
+    populate_collection(collection, test_case)
+    target = target_name(collection, test_case)
+    pipeline = list(test_case.pipeline) + [test_case.build_out_stage(collection)]
+    execute_command(
+        collection,
+        {"aggregate": collection.name, "pipeline": pipeline, "cursor": {}},
+    )
+    result = execute_command(
+        collection,
+        {"find": target, "filter": {}, "sort": {"_id": 1}},
+    )
+    assertResult(result, expected=test_case.expected, msg=test_case.msg)
+    collection.database.drop_collection(target)
+
+
+@pytest.mark.engine_xfail(
+    engine="pgmongo",
+    reason=(
+        "$out after $bucketAuto fails with CommandNotSupported (115): the "
+        "$bucketAuto plan is treated as a mutable function inside $out, whereas "
+        "native MongoDB writes the auto-bucketed output. Tracked: ADO #5371312"
+    ),
+    raises=AssertionError,
+)
+def test_out_after_bucketauto(collection):
+    """$out writes the auto-bucketed output produced by a preceding $bucketAuto."""
+    collection.insert_many(
+        [{"_id": i, "x": v} for i, v in enumerate([1, 5, 12, 18, 22, 35], start=1)]
+    )
+    target = f"{collection.name}_bucketauto_out"
+    collection.database.drop_collection(target)
+    execute_command(
+        collection,
+        {
+            "aggregate": collection.name,
+            "pipeline": [
+                {"$bucketAuto": {"groupBy": "$x", "buckets": 2}},
+                {"$out": target},
+            ],
+            "cursor": {},
+        },
+    )
+    written = execute_command(
+        collection,
+        {"find": target, "filter": {}, "sort": {"_id": 1}},
+    )
+    assertResult(
+        written,
+        expected=[
+            {"_id": {"min": 1, "max": 18}, "count": 3},
+            {"_id": {"min": 18, "max": 35}, "count": 3},
+        ],
+        msg="$out should persist the two auto-computed buckets from $bucketAuto.",
+    )
+    collection.database.drop_collection(target)
diff --git a/documentdb_tests/compatibility/tests/core/operator/update/bitwise/bit/test_bit_command_paths.py b/documentdb_tests/compatibility/tests/core/operator/update/bitwise/bit/test_bit_command_paths.py
new file mode 100644
index 000000000..269b5af40
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/update/bitwise/bit/test_bit_command_paths.py
@@ -0,0 +1,145 @@
+"""$bit update operator across command paths (findAndModify and bulk write).
+
+Existing $bit coverage drives the operator exclusively through the ``update``
+command. This file covers the other command paths that route $bit through
+distinct gateway code: the ``findAndModify`` command (with ``new`` pre/post
+image selection and ``upsert``) and a batched bulk write. The bitwise result
+and the response metadata must match the ``update`` command path.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline). The engine under test
+matches native behavior on every case; no engine divergences are tracked for
+this surface.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+from pymongo import UpdateOne
+
+from documentdb_tests.framework.assertions import assertResult, assertSuccessPartial
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.test_case import BaseTestCase
+
+pytestmark = pytest.mark.update
+
+
+@dataclass(frozen=True)
+class BitFindAndModifyTest(BaseTestCase):
+    """A $bit findAndModify case."""
+
+    setup_doc: Any = None
+    command_extra: dict = field(default_factory=dict)
+    bit_op: Any = None
+    # Expected `value` image returned by findAndModify (pre- or post-image).
+    expected_value: Any = None
+    # Expected resulting document after the operation (read back via find).
+    expected_doc: Any = None
+
+
+FIND_AND_MODIFY_TESTS: list[BitFindAndModifyTest] = [
+    BitFindAndModifyTest(
+        "and_returns_post_image",
+        setup_doc={"_id": 1, "v": 13},
+        bit_op={"and": 10},
+        command_extra={"new": True},
+        expected_value={"_id": 1, "v": 8},
+        expected_doc={"_id": 1, "v": 8},
+        msg="findAndModify $bit AND returns the post-image when new=true.",
+    ),
+    BitFindAndModifyTest(
+        "or_returns_post_image",
+        setup_doc={"_id": 1, "v": 13},
+        bit_op={"or": 2},
+        command_extra={"new": True},
+        expected_value={"_id": 1, "v": 15},
+        expected_doc={"_id": 1, "v": 15},
+        msg="findAndModify $bit OR returns the post-image when new=true.",
+    ),
+    BitFindAndModifyTest(
+        "xor_returns_post_image",
+        setup_doc={"_id": 1, "v": 13},
+        bit_op={"xor": 6},
+        command_extra={"new": True},
+        expected_value={"_id": 1, "v": 11},
+        expected_doc={"_id": 1, "v": 11},
+        msg="findAndModify $bit XOR returns the post-image when new=true.",
+    ),
+    BitFindAndModifyTest(
+        "and_returns_pre_image_when_new_false",
+        setup_doc={"_id": 1, "v": 13},
+        bit_op={"and": 10},
+        command_extra={"new": False},
+        expected_value={"_id": 1, "v": 13},
+        expected_doc={"_id": 1, "v": 8},
+        msg="findAndModify $bit returns the pre-image when new=false but still applies.",
+    ),
+]
+
+
+def run_find_and_modify(collection, test: BitFindAndModifyTest):
+    """Insert the seed doc, run $bit via findAndModify, return (response, found doc)."""
+    collection.insert_one(dict(test.setup_doc))
+    command = {
+        "findAndModify": collection.name,
+        "query": {"_id": test.setup_doc["_id"]},
+        "update": {"$bit": {"v": test.bit_op}},
+    }
+    command.update(test.command_extra)
+    response = execute_command(collection, command)
+    found = execute_command(
+        collection,
+        {"find": collection.name, "filter": {"_id": test.setup_doc["_id"]}, "sort": {"_id": 1}},
+    )
+    return response, found
+
+
+@pytest.mark.parametrize("test", pytest_params(FIND_AND_MODIFY_TESTS))
+def test_bit_find_and_modify(collection, test: BitFindAndModifyTest):
+    """$bit applied through findAndModify returns the documented image."""
+    response, _ = run_find_and_modify(collection, test)
+    assertSuccessPartial(response, {"value": test.expected_value}, msg=test.msg)
+
+
+@pytest.mark.parametrize("test", pytest_params(FIND_AND_MODIFY_TESTS))
+def test_bit_find_and_modify_persisted(collection, test: BitFindAndModifyTest):
+    """$bit applied through findAndModify persists the bitwise result."""
+    _, found = run_find_and_modify(collection, test)
+    assertResult(found, expected=[test.expected_doc], msg=test.msg)
+
+
+def test_bit_find_and_modify_upsert(collection):
+    """findAndModify $bit with upsert initializes the field from 0 and returns it."""
+    response = execute_command(
+        collection,
+        {
+            "findAndModify": collection.name,
+            "query": {"_id": 99},
+            "update": {"$bit": {"v": {"or": 5}}},
+            "upsert": True,
+            "new": True,
+        },
+    )
+    assertSuccessPartial(
+        response,
+        {"value": {"_id": 99, "v": 5}},
+        msg="findAndModify $bit upsert initializes v from 0 then applies OR 5.",
+    )
+
+
+def test_bit_bulk_write_update_one(collection):
+    """$bit applied through a batched bulk write produces the bitwise result."""
+    collection.insert_one({"_id": 1, "v": 13})
+    collection.bulk_write([UpdateOne({"_id": 1}, {"$bit": {"v": {"and": 10}}})])
+    found = execute_command(
+        collection,
+        {"find": collection.name, "filter": {"_id": 1}, "sort": {"_id": 1}},
+    )
+    assertResult(
+        found,
+        expected=[{"_id": 1, "v": 8}],
+        msg="A bulk-write $bit AND computes 1101 & 1010 = 1000.",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/operator/update/modifiers/slice/test_update_slice_combined_modifiers.py b/documentdb_tests/compatibility/tests/core/operator/update/modifiers/slice/test_update_slice_combined_modifiers.py
new file mode 100644
index 000000000..ab74c9688
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/update/modifiers/slice/test_update_slice_combined_modifiers.py
@@ -0,0 +1,136 @@
+"""$slice push modifier combined with $sort and $position (modifier ordering).
+
+Existing $slice modifier coverage exercises $slice on its own (with or without
+$each). This file covers the documented modifier-processing order when $slice
+is combined with $sort and $position in a single $push: the $each elements are
+appended (at $position when present), the whole array is reordered by $sort,
+and only then is it trimmed by $slice.
+
+A $slice of 0 combined with these modifiers is a tracked engine divergence:
+native MongoDB empties the array, while the engine under test leaves the array
+unchanged.
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline).
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.operator.update.utils.update_test_case import (
+    UpdateTestCase,
+)
+from documentdb_tests.framework.assertions import assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.update
+
+# Property [Modifier Ordering]: in one $push, elements are appended (at
+# $position), the array is reordered by $sort, then trimmed by $slice.
+COMBINED_MODIFIER_TESTS: list[UpdateTestCase] = [
+    UpdateTestCase(
+        id="sort_asc_then_slice_first_two",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1, 2], "$sort": 1, "$slice": 2}}},
+        expected=[{"_id": 1, "arr": [1, 2]}],
+        msg="$sort ascending then $slice keeps the two smallest elements.",
+    ),
+    UpdateTestCase(
+        id="sort_asc_then_slice_last_two",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1, 2], "$sort": 1, "$slice": -2}}},
+        expected=[{"_id": 1, "arr": [4, 5]}],
+        msg="$sort ascending then negative $slice keeps the two largest elements.",
+    ),
+    UpdateTestCase(
+        id="sort_desc_then_slice_first_three",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1, 2], "$sort": -1, "$slice": 3}}},
+        expected=[{"_id": 1, "arr": [5, 4, 3]}],
+        msg="$sort descending then $slice keeps the three largest elements.",
+    ),
+    UpdateTestCase(
+        id="position_then_slice_no_sort",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1], "$position": 0, "$slice": 3}}},
+        expected=[{"_id": 1, "arr": [3, 1, 5]}],
+        msg="$position inserts at the head, then $slice keeps the first three.",
+    ),
+    UpdateTestCase(
+        id="position_sort_then_slice",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1], "$position": 0, "$sort": 1, "$slice": 3}}},
+        expected=[{"_id": 1, "arr": [1, 3, 4]}],
+        msg="$sort reorders the whole array regardless of $position before $slice trims it.",
+    ),
+    UpdateTestCase(
+        id="sort_documents_then_slice_first_two",
+        setup_docs=[{"_id": 1, "arr": [{"s": 5}, {"s": 4}]}],
+        query={"_id": 1},
+        update={
+            "$push": {
+                "arr": {"$each": [{"s": 3}, {"s": 1}], "$sort": {"s": 1}, "$slice": 2}
+            }
+        },
+        expected=[{"_id": 1, "arr": [{"s": 1}, {"s": 3}]}],
+        msg="$sort by subfield ascending then $slice keeps the two smallest documents.",
+    ),
+    UpdateTestCase(
+        id="sort_documents_desc_then_slice_last_two",
+        setup_docs=[{"_id": 1, "arr": [{"s": 5}, {"s": 4}]}],
+        query={"_id": 1},
+        update={
+            "$push": {
+                "arr": {"$each": [{"s": 3}, {"s": 1}], "$sort": {"s": -1}, "$slice": -2}
+            }
+        },
+        expected=[{"_id": 1, "arr": [{"s": 3}, {"s": 1}]}],
+        msg="$sort by subfield descending then negative $slice keeps the two smallest documents.",
+    ),
+    UpdateTestCase(
+        id="missing_field_sort_then_slice",
+        setup_docs=[{"_id": 1}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1, 2], "$sort": 1, "$slice": 2}}},
+        expected=[{"_id": 1, "arr": [1, 2]}],
+        msg="On a missing field the array is created, sorted, then sliced.",
+    ),
+    UpdateTestCase(
+        id="sort_then_slice_zero_empties",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1, 2], "$sort": 1, "$slice": 0}}},
+        expected=[{"_id": 1, "arr": []}],
+        msg="$slice 0 empties the array after $each and $sort are applied.",
+        marks=(
+            pytest.mark.engine_xfail(
+                engine="pgmongo",
+                reason=(
+                    "$push with $slice 0 is a no-op on this engine (the array keeps its "
+                    "original contents), whereas native MongoDB empties the array. "
+                    "Tracked: ADO #5371311"
+                ),
+                raises=AssertionError,
+            ),
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize("test_case", pytest_params(COMBINED_MODIFIER_TESTS))
+def test_update_slice_combined_modifiers(collection, test_case: UpdateTestCase):
+    """$slice trims the array after $each/$position/$sort modifiers are applied."""
+    collection.insert_many(test_case.setup_docs)
+    execute_command(
+        collection,
+        {"update": collection.name, "updates": [{"q": test_case.query, "u": test_case.update}]},
+    )
+    result = execute_command(
+        collection,
+        {"find": collection.name, "filter": test_case.query, "sort": {"_id": 1}},
+    )
+    assertSuccess(result, test_case.expected, msg=test_case.msg)
diff --git a/documentdb_tests/compatibility/tests/core/operator/update/modifiers/sort/test_update_sort_with_position.py b/documentdb_tests/compatibility/tests/core/operator/update/modifiers/sort/test_update_sort_with_position.py
new file mode 100644
index 000000000..870b10e24
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/operator/update/modifiers/sort/test_update_sort_with_position.py
@@ -0,0 +1,92 @@
+"""$sort push modifier combined with the $position modifier.
+
+Existing $sort modifier coverage exercises $sort on its own. This file covers
+its interaction with $position: $position controls where the $each elements are
+inserted, and when $sort is also present it reorders the entire array, making
+the positional insert order irrelevant to the final result. Positional inserts
+without $sort establish the baseline insert semantics (head, middle, negative
+offset from the end, and an offset past the end).
+
+Oracle: MongoDB 7.0 (functional-tests CI baseline). The engine under test
+matches native behavior on every case in this file; no engine divergences are
+tracked here.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.operator.update.utils.update_test_case import (
+    UpdateTestCase,
+)
+from documentdb_tests.framework.assertions import assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.update
+
+# Property [Positional Insert + Sort]: $position selects the insertion point for
+# the $each elements; a co-present $sort then reorders the whole array.
+SORT_POSITION_TESTS: list[UpdateTestCase] = [
+    UpdateTestCase(
+        id="position_head_no_sort",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1], "$position": 0}}},
+        expected=[{"_id": 1, "arr": [3, 1, 5, 4]}],
+        msg="$position 0 inserts the $each elements at the head.",
+    ),
+    UpdateTestCase(
+        id="position_middle_no_sort",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [9], "$position": 1}}},
+        expected=[{"_id": 1, "arr": [5, 9, 4]}],
+        msg="$position 1 inserts the $each element after the first element.",
+    ),
+    UpdateTestCase(
+        id="position_negative_from_end",
+        setup_docs=[{"_id": 1, "arr": [5, 4, 7]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [9], "$position": -1}}},
+        expected=[{"_id": 1, "arr": [5, 4, 9, 7]}],
+        msg="A negative $position counts the insertion point from the array end.",
+    ),
+    UpdateTestCase(
+        id="position_past_end_appends",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [9], "$position": 100}}},
+        expected=[{"_id": 1, "arr": [5, 4, 9]}],
+        msg="A $position past the end appends at the tail.",
+    ),
+    UpdateTestCase(
+        id="position_then_sort_reorders_all",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1], "$position": 0, "$sort": 1}}},
+        expected=[{"_id": 1, "arr": [1, 3, 4, 5]}],
+        msg="$sort reorders the entire array, overriding the $position insert order.",
+    ),
+    UpdateTestCase(
+        id="position_then_sort_descending",
+        setup_docs=[{"_id": 1, "arr": [5, 4]}],
+        query={"_id": 1},
+        update={"$push": {"arr": {"$each": [3, 1], "$position": 1, "$sort": -1}}},
+        expected=[{"_id": 1, "arr": [5, 4, 3, 1]}],
+        msg="$sort descending reorders the whole array regardless of $position.",
+    ),
+]
+
+
+@pytest.mark.parametrize("test_case", pytest_params(SORT_POSITION_TESTS))
+def test_update_sort_with_position(collection, test_case: UpdateTestCase):
+    """$position controls insertion; a co-present $sort reorders the whole array."""
+    collection.insert_many(test_case.setup_docs)
+    execute_command(
+        collection,
+        {"update": collection.name, "updates": [{"q": test_case.query, "u": test_case.update}]},
+    )
+    result = execute_command(
+        collection,
+        {"find": collection.name, "filter": test_case.query, "sort": {"_id": 1}},
+    )
+    assertSuccess(result, test_case.expected, msg=test_case.msg)