diff --git a/changelog.d/20.added.md b/changelog.d/20.added.md
new file mode 100644
index 0000000..065493e
--- /dev/null
+++ b/changelog.d/20.added.md
@@ -0,0 +1 @@
+Add parallel testing plugin with database pool and serial mechanism, plus comprehensive cross-schema query and test coverage
\ No newline at end of file
diff --git a/src/rhosocial/activerecord/testsuite/conftest.py b/src/rhosocial/activerecord/testsuite/conftest.py
index bf22761..d00c2dd 100644
--- a/src/rhosocial/activerecord/testsuite/conftest.py
+++ b/src/rhosocial/activerecord/testsuite/conftest.py
@@ -35,8 +35,19 @@
     os.environ["TESTSUITE_ACTIVE_SCENARIOS"] = _argv_scenarios
 
 
+# Guard against registering the same options twice when the conftest is loaded
+# both as a plugin (e.g. ``-p rhosocial.activerecord.testsuite.conftest`` in the
+# core CI) and as a path-based conftest for testsuite-path tests.
+_pytest_options_registered = False
+
+
 def pytest_addoption(parser):
     """Register the generic --scenarios option used by all backends."""
+    global _pytest_options_registered
+    if _pytest_options_registered:
+        return
+    _pytest_options_registered = True
+
     parser.addoption(
         "--scenarios",
         action="store",
@@ -45,6 +56,30 @@ def pytest_addoption(parser):
              "(e.g., --scenarios=firebird_5,mysql_80). Backend scenario "
              "modules filter their registered scenarios accordingly.",
     )
+    parser.addoption(
+        "--scenarios-parallel",
+        action="store_true",
+        default=True,
+        help="Allow scenario variants of the same test to run in parallel "
+             "(default: True). With --scenarios-parallel, scenario variants "
+             "are freely distributed across xdist workers; without it, they "
+             "are pinned to the same worker and run sequentially.",
+    )
+    parser.addoption(
+        "--db-pool-size",
+        action="store",
+        type=int,
+        default=None,
+        help="Number of pooled test_db_* databases prepared per scenario "
+             "(default: follow the number of workers, i.e. -n, minimum 1). "
+             "Use 0 to disable pooling and keep per-test unique databases.",
+    )
+    parser.addoption(
+        "--serial-group",
+        action="store",
+        default="serial",
+        help="xdist_group name that serial tests are pinned to (default: serial).",
+    )
 
 
 def pytest_configure(config):
@@ -71,32 +106,183 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "benchmark_transaction: Mark transaction benchmark tests")
     config.addinivalue_line("markers", "benchmark_mixin: Mark mixin benchmark tests")
     config.addinivalue_line("markers", "benchmark_fastapi: Mark FastAPI benchmark tests")
+    config.addinivalue_line(
+        "markers", "serial: Mark tests that must run serially (never concurrently with other tests)"
+    )
 
-@pytest.hookimpl(trylast=True)
+    from rhosocial.activerecord.testsuite.core.pool import configure_pool_size
+
+    configure_pool_size(_resolve_pool_size(config))
+
+    # Preload the provider registry. Importing the registry module pulls in
+    # the backend's provider modules, whose side effect is registering the
+    # backend-specific pool reset handler (e.g. SQLite's drop-all-tables).
+    # This must happen here, in every xdist worker, so that the per-test
+    # setup hook can reset pooled databases before the first test runs.
+    try:
+        from rhosocial.activerecord.testsuite.core.registry import get_provider_registry
+
+        get_provider_registry()
+    except Exception:
+        pass
+
+@pytest.hookimpl(tryfirst=True)
 def pytest_collection_modifyitems(config, items):
     """
-    Hook to automatically skip tests that require unsupported protocols or functions.
+    Pin serial tests to a single xdist worker so they never run concurrently.
+
+    Tests marked ``serial`` (e.g. WorkerPool tests that spawn their own
+    sub-processes) must not overlap with each other: they share a dedicated
+    database and a concurrent run would race on schema setup. They are pinned
+    to one worker via the ``xdist_group`` marker, which only takes effect with
+    ``--dist=loadgroup`` (pytest-xdist appends the group name to the nodeid).
+
+    Note: pytest-xdist runs this hook in *every* worker, where ``-n`` is not
+    present on the command line. xdist is therefore detected through the
+    ``PYTEST_XDIST_WORKER`` environment variable so the markers are applied in
+    workers too. ``tryfirst=True`` makes the markers visible to xdist's own
+    nodeid-rewriting step.
+
+    During collection time, backend-specific capabilities aren't available
+    through the provider interface since providers set up backends per test
+    scenario. Protocol and function checking happens during test execution when
+    provider-configured models are available.
+    """
+    # Apply serial / parallel scheduling only when running under xdist with
+    # more than one worker (otherwise everything is naturally serial).
+    import os
+
+    n_workers = config.getoption("-n", default=None)
+    if n_workers is None:
+        n_workers = config.getoption("--numprocesses", default=None)
+    if isinstance(n_workers, str):
+        n_workers = n_workers.strip("auto")
+    try:
+        workers = int(n_workers)
+    except (TypeError, ValueError):
+        workers = 0
+
+    if workers <= 1 and os.environ.get("PYTEST_XDIST_WORKER") is None:
+        return
+
+    serial_group = config.getoption("--serial-group", default="serial")
+    scenarios_parallel = config.getoption("--scenarios-parallel", default=True)
+    known_scenarios = _known_scenario_names()
+
+    for item in items:
+        is_serial = item.get_closest_marker("serial") is not None
+        if not is_serial:
+            # Cross-scenario tests use 2+ distinct scenario values at once and
+            # cannot be parallelized: pin them to the serial group.
+            scenario_params = _scenario_params_of(item, known_scenarios)
+            if len(scenario_params) > 1:
+                is_serial = True
+        if is_serial:
+            item.add_marker(pytest.mark.xdist_group(serial_group))
+            print(f"[AR_DBG] serial-pinned: {item.nodeid} -> group={serial_group}", flush=True)
+        elif not scenarios_parallel:
+            # Pin scenario variants of the same test to a single worker.
+            base_nodeid = item.nodeid.split("[", 1)[0]
+            item.add_marker(pytest.mark.xdist_group(f"scenario::{base_nodeid}"))
+
 
-    Note: During collection time, we can't access backend-specific capabilities
-    through the provider interface since providers set up backends per test scenario.
-    Protocol and function checking happens during test execution when provider-configured
-    models are available.
+def _resolve_pool_size(config):
+    """Resolve the database pool size.
 
-    ``trylast=True`` guarantees this no-op hook never shadows backend-owned
-    ``pytest_collection_modifyitems`` hooks (e.g. Firebird's xfail registration),
-    which are registered later and must take precedence.
+    Defaults to the number of xdist workers (``-n``, minimum 1) so that every
+    worker owns a dedicated ``test_db_*`` slot. An explicit ``--db-pool-size``
+    overrides it; 0 disables pooling.
+
+    Workers don't see the master's ``-n`` option, so the master publishes the
+    resolved size through ``RHS_AR_POOL_SIZE`` (inherited by worker processes)
+    and workers reuse it.
     """
-    # For now, we just ensure tests with requires_protocol/requires_functions markers exist properly
-    # Actual capability checking occurs at test runtime via fixtures and decorators
-    pass
+    explicit = config.getoption("--db-pool-size", default=None)
+    env_size = os.environ.get("RHS_AR_POOL_SIZE")
+    if explicit is not None:
+        size = explicit
+    elif env_size:
+        size = int(env_size)
+    else:
+        size = _resolve_worker_count(config)
+    if os.environ.get("PYTEST_XDIST_WORKER") is None:
+        os.environ["RHS_AR_POOL_SIZE"] = str(size)
+    return size
+
+
+def _resolve_worker_count(config):
+    """Return the number of xdist workers from the ``-n`` option (default 1)."""
+    workers = 1
+    try:
+        n_workers = config.getoption("-n", default=None)
+    except (ValueError, TypeError):
+        n_workers = None
+    if n_workers is not None:
+        value = str(n_workers).strip().lower()
+        if value == "auto":
+            workers = os.cpu_count() or 1
+        else:
+            try:
+                workers = int(value)
+            except ValueError:
+                workers = 1
+    return max(workers, 1)
+
+
+_KNOWN_SCENARIOS = None
+
+
+def _known_scenario_names():
+    """Collect all scenario names advertised by the registered providers."""
+    global _KNOWN_SCENARIOS
+    if _KNOWN_SCENARIOS is not None:
+        return _KNOWN_SCENARIOS
+    try:
+        from rhosocial.activerecord.testsuite.core.registry import get_provider_registry
+
+        registry = get_provider_registry()
+    except Exception:
+        _KNOWN_SCENARIOS = set()
+        return _KNOWN_SCENARIOS
+    names = set()
+    for provider_class in registry.all_providers():
+        try:
+            names.update(provider_class().get_test_scenarios())
+        except Exception:
+            continue
+    _KNOWN_SCENARIOS = names
+    return names
+
+
+def _scenario_params_of(item, known_scenarios):
+    """Return the set of scenario names used by an item's parametrization."""
+    if not known_scenarios:
+        return set()
+    callspec = getattr(item, "callspec", None)
+    if callspec is None:
+        return set()
+    values = set()
+    for value in callspec.params.values():
+        if isinstance(value, str) and value in known_scenarios:
+            values.add(value)
+    return values
 
 def pytest_sessionstart(session):
     """
     Hook to generate capability support warnings at session start.
-    
+
     This hook generates warnings about important unsupported capabilities
-    to alert developers about backend limitations.
+    to alert developers about backend limitations. It also prepares this
+    worker's pooled databases (ensures they exist) before any test connects.
     """
+    try:
+        from rhosocial.activerecord.testsuite.core.pool import is_xdist_worker, prepare_pool
+
+        if not is_xdist_worker():
+            prepare_pool()
+    except Exception as e:
+        warnings.warn(f"Could not prepare the database pool at session start: {e}", UserWarning)
+
     try:
         from rhosocial.activerecord.backend.dialect.protocols import (
             WindowFunctionSupport,
@@ -105,7 +291,7 @@ def pytest_sessionstart(session):
             ReturningSupport,
         )
 
-        from .utils import get_current_backend
+        from rhosocial.activerecord.testsuite.utils import get_current_backend
         backend = get_current_backend()
 
         if backend is None:
@@ -142,4 +328,27 @@ def pytest_sessionstart(session):
                 UserWarning
             )
     except Exception as e:
-        warnings.warn(f"Could not check capability support at session start: {e}", UserWarning)
\ No newline at end of file
+        warnings.warn(f"Could not check capability support at session start: {e}", UserWarning)
+
+
+def pytest_sessionfinish(session, exitstatus):
+    """Reset the database pool state.
+
+    ``test_db_*`` databases are intentionally NOT removed after the session;
+    they persist across sessions and are re-prepared on the next session.
+    """
+    from rhosocial.activerecord.testsuite.core.pool import reset_pool
+
+    reset_pool()
+
+
+@pytest.hookimpl(trylast=True)
+def pytest_runtest_teardown(item, nextitem):
+    """Release every pool slot acquired during the just-finished test.
+
+    Fixtures have already been torn down, so backends are disconnected before
+    their slots are freed; the next test on any worker can take them.
+    """
+    from rhosocial.activerecord.testsuite.core.pool import release_all
+
+    release_all()
\ No newline at end of file
diff --git a/src/rhosocial/activerecord/testsuite/core/pool.py b/src/rhosocial/activerecord/testsuite/core/pool.py
new file mode 100644
index 0000000..1efd9c9
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/core/pool.py
@@ -0,0 +1,298 @@
+# src/rhosocial/activerecord/testsuite/core/pool.py
+"""Database pool for parallel test execution.
+
+A simple preemptive database-slot allocator. Before the session starts the
+master prepares ``{base}_0`` .. ``{base}_{N-1}`` databases for every scenario
+(creating them if missing and clearing leftover tables), where N is the pool
+size (defaults to the number of xdist workers). Each test then takes any free
+slot when its fixtures need a database, uses it, and gives it back when the
+test finishes. Tests own their schema: they set up and tear down their own
+tables, and the pool only guarantees the slot databases were clean at session
+start.
+
+Pool semantics:
+
+- **Naming**: ``{database}_{index}`` — the prefix is derived from the
+  scenario's configured database name (e.g. MySQL scenarios whose YAML sets
+  ``database: test_db`` produce ``test_db_0``, ``test_db_1``, ...). Each
+  backend registers its scenarios' base names via :func:`register_base_database`;
+  scenarios without a registration fall back to ``test_db``. SQLite appends
+  ``.sqlite`` (``test_db_3.sqlite``); server backends use the bare
+  ``test_db_3`` database/schema name. The scenario name is deliberately NOT
+  part of the database name; the scenario selects the server (host/port) while
+  the index selects the database on that server.
+- **Preparation**: :func:`prepare_pool` runs once per test session on the
+  master, before any test is dispatched. It invokes the backend's registered
+  preparation handler for every slot of every scenario; handlers ensure the
+  database exists and drop leftover tables.
+- **Allocation**: a test acquires a slot per scenario the first time it needs
+  the database name, taking the first slot whose per-slot file lock is free
+  (preemptive allocation). The slot is held for the rest of the test and
+  released by :func:`release_all`. Per-slot file locks coordinate across xdist
+  worker processes; the OS releases a lock automatically if a worker crashes.
+  Because the pool size equals the number of workers, a free slot is always
+  available, so allocation does not spin.
+- **Lifecycle**: ``{database}_*`` databases are not removed after the session;
+  they persist and are prepared (cleared) again at the start of the next
+  session.
+- Pooling is only active when running under xdist workers with a positive
+  pool size. Serial runs (no ``-n``) keep the previous unique-file/unique-
+  database behaviour.
+"""
+import os
+import re
+import tempfile
+import threading
+import time
+from typing import Callable, Dict, Iterable, List, Optional
+
+_POOL_SIZE = 0
+_RESET_HANDLER: Optional[Callable[[str, str], None]] = None
+_CLEAR_ON_ACQUIRE = True
+_POOL_DIR: Optional[str] = None
+
+_DEFAULT_BASE_DATABASE = "test_db"
+_BASE_DATABASES: Dict[str, str] = {}
+
+_WORKER_RE = re.compile(r"^gw(\d+)$")
+
+# Slots held by the current worker for the currently executing test.
+_HELD: Dict[str, int] = {}
+_OPEN_LOCKS: Dict[tuple, object] = {}
+_GUARD = threading.Lock()
+
+
+def register_base_database(scenario: str, base_database: str) -> None:
+    """Register the base database name used for a scenario's pooled databases.
+
+    The pooled database name for a scenario is ``{base_database}_{index}``
+    (e.g. ``test_db_3``). Backends register each scenario's configured database
+    name (e.g. from the YAML ``database`` field) so the pool name derivation
+    follows the scenario configuration instead of a hardcoded prefix.
+    """
+    _BASE_DATABASES[scenario] = base_database
+
+
+def base_database(scenario: str) -> str:
+    """Return the registered base database name for a scenario.
+
+    Falls back to ``test_db`` when the scenario has no registered base name.
+    """
+    return _BASE_DATABASES.get(scenario, _DEFAULT_BASE_DATABASE)
+
+
+def configure_pool_dir(path: Optional[str]) -> None:
+    """Set the directory where pooled ``test_db_*`` files are created.
+
+    Falls back to the default temp directory when ``None``.
+    """
+    global _POOL_DIR
+    _POOL_DIR = path or None
+
+
+def pool_dir() -> str:
+    """Return the directory where pooled ``test_db_*`` files are created."""
+    return _POOL_DIR or os.path.join(tempfile.gettempdir(), "rhosocial-test-db")
+
+
+def configure_pool_size(size: int) -> None:
+    """Set the configured pool size (0 disables pooling).
+
+    The pool size is the number of slot databases prepared per scenario. It
+    defaults to the number of xdist workers so a free slot is always available.
+    """
+    global _POOL_SIZE
+    _POOL_SIZE = max(0, int(size))
+
+
+def pool_size() -> int:
+    """Return the configured pool size (0 means pooling disabled)."""
+    return _POOL_SIZE
+
+
+def is_xdist_worker() -> bool:
+    """Return True when running inside a pytest-xdist worker process."""
+    return bool(os.environ.get("PYTEST_XDIST_WORKER"))
+
+
+def worker_index() -> int:
+    """Return the index of the current xdist worker (``gw0`` -> 0, default 0)."""
+    worker_id = os.environ.get("PYTEST_XDIST_WORKER", "")
+    match = _WORKER_RE.match(worker_id)
+    if match:
+        return int(match.group(1))
+    return 0
+
+
+def pooling_active() -> bool:
+    """Return True when the database pool should be used for the current test.
+
+    Pooling requires running under xdist workers with a positive pool size.
+    Serial runs keep the previous per-test unique-database behaviour.
+    """
+    return _POOL_SIZE > 0 and is_xdist_worker()
+
+
+def register_pool_reset_handler(
+    handler: Optional[Callable[[str, str], None]], *, clear_on_acquire: bool = True
+) -> None:
+    """Register a backend-specific slot-preparation callback.
+
+    The callback receives ``(scenario, database_name)`` where ``database_name``
+    is the bare pooled name (e.g. ``test_db_3``) of a slot to prepare. It must
+    ensure that slot's database exists and is clear of leftover tables.
+
+    ``clear_on_acquire`` controls whether the handler is invoked every time a
+    slot is handed to a test. Backends whose tests are fully self-contained
+    (they create and drop their own tables) can set it to ``False`` and rely on
+    the single session-start preparation, avoiding a DROP-all-tables pass per
+    test. File-based backends whose schema setup is order-sensitive (SQLite
+    drops a parent table before its dependents, which foreign_keys=ON rejects)
+    keep ``True`` so every test still starts from a pristine schema.
+    """
+    global _RESET_HANDLER, _CLEAR_ON_ACQUIRE
+    _RESET_HANDLER = handler
+    _CLEAR_ON_ACQUIRE = bool(clear_on_acquire)
+
+
+def slot_name(index: int, base: str) -> str:
+    """Return the bare pooled database name (e.g. ``test_db_3``)."""
+    return f"{base}_{index}"
+
+
+def prepare_pool(scenarios: Optional[Iterable[str]] = None) -> None:
+    """Create and clear the pooled databases of every scenario (once per run).
+
+    Runs on the master before any test is dispatched. Every slot database of
+    every (enabled) scenario is brought to a clean, empty state so tests start
+    from a pristine schema regardless of leftovers from previous sessions.
+    """
+    if _POOL_SIZE <= 0 or _RESET_HANDLER is None:
+        return
+    scenario_names = list(scenarios) if scenarios is not None else list(_BASE_DATABASES)
+    for scenario in scenario_names:
+        base = base_database(scenario)
+        for index in range(_POOL_SIZE):
+            try:
+                _RESET_HANDLER(scenario, slot_name(index, base))
+            except Exception:
+                pass
+
+
+def _lock_dir() -> str:
+    """Return the directory holding per-slot lock files (shared across workers)."""
+    return os.path.join(pool_dir(), "pool-locks")
+
+
+def _acquire_slot(scenario: str) -> int:
+    """Acquire the first free slot index for a scenario (preemptive allocation).
+
+    Takes the per-slot file lock of the first free slot; the lock is released
+    when the test finishes (or automatically if the worker crashes). The slot
+    database is cleared right after acquisition (the pool manager only hands
+    out clean databases), so a test always starts from a pristine schema even
+    though the slot file was reused by a previous test. When every slot is
+    briefly busy (pool size smaller than the worker count), the caller waits on
+    a slot lock instead of spinning.
+    """
+    import fcntl
+
+    lock_dir = _lock_dir()
+    os.makedirs(lock_dir, exist_ok=True)
+    base = base_database(scenario)
+    lock_base = base.replace(os.sep, "_")
+    size = max(_POOL_SIZE, 1)
+    while True:
+        for index in range(size):
+            path = os.path.join(lock_dir, f"{scenario}-{lock_base}-{index}.lock")
+            fh = open(path, "a+")
+            try:
+                fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except OSError:
+                fh.close()
+                continue
+            with _GUARD:
+                _OPEN_LOCKS[(scenario, index)] = fh
+            if _RESET_HANDLER is not None and _CLEAR_ON_ACQUIRE:
+                try:
+                    _RESET_HANDLER(scenario, slot_name(index, base))
+                except Exception:
+                    pass
+            return index
+        # All slots busy: block on the first slot until it is released.
+        path = os.path.join(lock_dir, f"{scenario}-{lock_base}-0.lock")
+        fh = open(path, "a+")
+        try:
+            fcntl.flock(fh, fcntl.LOCK_EX)
+        finally:
+            fcntl.flock(fh, fcntl.LOCK_UN)
+            fh.close()
+
+
+def _slot_for(scenario: str) -> int:
+    """Return the worker's held slot index for a scenario, acquiring it on demand.
+
+    The slot is acquired lazily on the first database access of the current
+    test and held until :func:`release_all`; repeated lookups within the same
+    test return the same slot.
+    """
+    with _GUARD:
+        index = _HELD.get(scenario)
+        if index is not None:
+            return index
+    index = _acquire_slot(scenario)
+    with _GUARD:
+        _HELD[scenario] = index
+    return index
+
+
+def pooled_database_name(scenario: str) -> Optional[str]:
+    """
+    Return the pooled database name (e.g. ``test_db_3``) for the scenario's
+    currently held slot, or ``None`` when pooling is inactive (callers then
+    fall back to the scenario's configured database).
+    """
+    if not pooling_active():
+        return None
+    return slot_name(_slot_for(scenario), base_database(scenario))
+
+
+def pooled_database_path(scenario: str, suffix: str = ".sqlite") -> Optional[str]:
+    """
+    Return the pooled database on-disk path for the scenario's held slot, or
+    ``None`` when pooling is inactive. File-based backends (e.g. SQLite) use
+    this to obtain the file path.
+    """
+    if not pooling_active():
+        return None
+    name = slot_name(_slot_for(scenario), base_database(scenario))
+    path = os.path.join(pool_dir(), name + suffix)
+    os.makedirs(pool_dir(), exist_ok=True)
+    return path
+
+
+def release_all() -> None:
+    """Release every pool slot held by the current worker.
+
+    Called when a test finishes so the freed slots are immediately available to
+    the next test on any worker.
+    """
+    import fcntl
+
+    with _GUARD:
+        held = list(_HELD.items())
+        _HELD.clear()
+        for scenario, index in held:
+            fh = _OPEN_LOCKS.pop((scenario, index), None)
+            if fh is None:
+                continue
+            try:
+                fcntl.flock(fh, fcntl.LOCK_UN)
+            except Exception:
+                pass
+            fh.close()
+
+
+def reset_pool() -> None:
+    """Release all held slots (used at session end / between sessions)."""
+    release_all()
\ No newline at end of file
diff --git a/src/rhosocial/activerecord/testsuite/core/registry.py b/src/rhosocial/activerecord/testsuite/core/registry.py
index d653e49..cde8d8d 100644
--- a/src/rhosocial/activerecord/testsuite/core/registry.py
+++ b/src/rhosocial/activerecord/testsuite/core/registry.py
@@ -25,6 +25,10 @@ def get_provider(self, interface_path: str) -> Optional[Type]:
         """Retrieves a provider class for a given interface path."""
         return self._providers.get(interface_path)
 
+    def all_providers(self) -> list:
+        """Returns all registered provider classes (values of the registry)."""
+        return list(self._providers.values())
+
 # A global singleton to hold the registry instance.
 _registry_instance = None
 
diff --git a/src/rhosocial/activerecord/testsuite/core/scenario.py b/src/rhosocial/activerecord/testsuite/core/scenario.py
new file mode 100644
index 0000000..3f91c0c
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/core/scenario.py
@@ -0,0 +1,34 @@
+# src/rhosocial/activerecord/testsuite/core/scenario.py
+"""Scenario metadata model for parallel test execution.
+
+A *scenario* describes one concrete database instance/shape (e.g. an SQLite
+file layout, a MySQL server version, a PostgreSQL version with/without
+plugin extensions). Scenarios of the same backend are isolated from each
+other, so the same test case can safely run for several scenarios in
+parallel.
+"""
+from dataclasses import dataclass
+from typing import Callable, Optional, Tuple
+
+
+@dataclass(frozen=True)
+class Scenario:
+    """
+    Metadata describing a single test scenario.
+
+    Attributes:
+        name: Unique scenario name (used as parametrization id).
+        description: Human readable description of the database shape.
+        tags: Optional tags (e.g. "file", "memory", "version:9.7").
+        is_poolable: Whether databases of this scenario may be pooled and
+            reused across tests. File based databases are poolable;
+            in-memory ones are not (they vanish with their connection).
+        config_factory: Optional callable returning the backend config for
+            this scenario.
+    """
+
+    name: str
+    description: str = ""
+    tags: Tuple[str, ...] = ()
+    is_poolable: bool = True
+    config_factory: Optional[Callable] = None
\ No newline at end of file
diff --git a/src/rhosocial/activerecord/testsuite/feature/README.md b/src/rhosocial/activerecord/testsuite/feature/README.md
index b57f68c..333a310 100644
--- a/src/rhosocial/activerecord/testsuite/feature/README.md
+++ b/src/rhosocial/activerecord/testsuite/feature/README.md
@@ -21,6 +21,33 @@ Tests are organized into two-level categories for discoverability and maintainab
 | `examples/` | Documentation examples for the capability-based framework — [details](examples/README.md) |
 | `backend/` | Backend feature declaration utilities (not tests) — [details](backend/README.md) |
 
+### Two-Level Classification (执行分级)
+
+Feature categories are further grouped into two execution levels (**一级 / 二级**).
+A backend integration is expected to run level-1 suites first and in order
+(`basic` → `relation` → `query`), and only proceed to level-2 suites once
+level-1 suites pass (failures limited to correctly capability-gated skips).
+
+| Level | Categories | Rationale |
+|-------|------------|-----------|
+| **一级 (Level 1)** | `basic/`, `relation/`, `query/` | Core ActiveRecord contract: models, CRUD, fields/types, relationships, and ActiveQuery. A backend that cannot pass level-1 is not considered usable. |
+| **二级 (Level 2)** | `events/`, `mixins/`, `interface/`, `examples/` | Extended behaviors built on top of level-1 primitives (lifecycle hooks, timestamp/soft-delete/optimistic-locking mixins, interface utilities). |
+
+Notes:
+- `backend/` is a utility directory (backend feature declaration helpers), not a
+  test suite, and does not belong to either level.
+- `composite_pk/` and `derived_field/` no longer exist as top-level suites;
+  their tests are merged into level-1 (`basic/` and `query/`), so they count as
+  level-1 content.
+- Within each level, run categories one at a time; long-running suites should
+  not be mixed into a single pytest invocation when validating a backend
+  integration.
+- Level assignments are about **execution ordering and acceptance gates**, not
+  capability gating: individual tests may still be skipped via the capability
+  system (`@requires_protocol`, provider registration, etc.) regardless of
+  their level.
+
+
 > **Note:** `composite_pk/` and `derived_field/` have been merged into `basic/` and `query/`.
 > Their `interfaces.py`, `conftest.py`, and `fixtures/` remain as shared providers.
 
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management.py
index 5ceaff1..a0d3d87 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management.py
@@ -12,6 +12,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management_async.py
index a0e07b3..edca2fb 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_connection_management_async.py
@@ -12,6 +12,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud.py
index 866ee39..548c219 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud.py
@@ -13,6 +13,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, PoolState, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions (must be module-level, pickle-able)
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud_async.py
index 56ed786..967f5b0 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_parallel_crud_async.py
@@ -13,6 +13,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, PoolState, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions (must be module-level, pickle-able)
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle.py
index 28c406a..73348c9 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle.py
@@ -12,6 +12,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, PoolState, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle_async.py b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle_async.py
index 387436a..0d8939d 100644
--- a/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/basic/worker/test_worker_pool_lifecycle_async.py
@@ -12,6 +12,8 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, PoolState, TaskContext
 
+pytestmark = pytest.mark.serial
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/conftest.py b/src/rhosocial/activerecord/testsuite/feature/query/conftest.py
index 8b866b6..7ece7ee 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/conftest.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/conftest.py
@@ -262,7 +262,8 @@ def check_protocol_requirements(request):
                           'annotated_query_fixtures', 'mapped_models_fixtures',
                           'async_order_fixtures', 'async_blog_fixtures', 'async_json_user_fixture',
                           'async_tree_fixtures', 'async_combined_fixtures', 'async_extended_order_fixtures',
-                          'async_annotated_query_fixtures', 'async_mapped_models_fixtures']
+                          'async_annotated_query_fixtures', 'async_mapped_models_fixtures',
+                          'sync_pool_and_model', 'async_pool_and_model']
 
         for fixture_name in fixture_options:
             if fixture_name in request.fixturenames:
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context.py b/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context.py
index f74d3b4..758afea 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context.py
@@ -7,10 +7,13 @@
 """
 import pytest
 
+from rhosocial.activerecord.backend.dialect.protocols import CTESupport
 from rhosocial.activerecord.query import CTEQuery, AsyncCTEQuery
+from rhosocial.activerecord.testsuite.utils import requires_protocol
 class TestSyncCTEQueryContext:
     """Test synchronous CTEQuery context awareness."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_backend_without_context(self, sync_pool_and_model):
         """Test CTEQuery.backend() returns constructor backend without context."""
         pool, model = sync_pool_and_model
@@ -20,6 +23,7 @@ def test_cte_query_backend_without_context(self, sync_pool_and_model):
         cte_backend = cte.backend()
         assert cte_backend is original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_backend_in_connection_context(self, sync_pool_and_model):
         """Test CTEQuery.backend() returns connection backend in context."""
         pool, model = sync_pool_and_model
@@ -32,6 +36,7 @@ def test_cte_query_backend_in_connection_context(self, sync_pool_and_model):
             assert cte_backend is conn_backend
             assert cte_backend is not original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_backend_in_transaction_context(self, sync_pool_and_model):
         """Test CTEQuery.backend() returns transaction backend in context."""
         pool, model = sync_pool_and_model
@@ -44,6 +49,7 @@ def test_cte_query_backend_in_transaction_context(self, sync_pool_and_model):
             assert cte_backend is tx_backend
             assert cte_backend is not original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_nested_connection_contexts_reuse(self, sync_pool_and_model):
         """Test nested connection contexts reuse for CTE query."""
         pool, model = sync_pool_and_model
@@ -59,6 +65,7 @@ def test_nested_connection_contexts_reuse(self, sync_pool_and_model):
                 assert inner_cte.backend() is outer_conn
                 assert inner_conn is outer_conn
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_nested_transaction_contexts_reuse(self, sync_pool_and_model):
         """Test nested transaction contexts reuse for CTE query."""
         pool, model = sync_pool_and_model
@@ -74,6 +81,7 @@ def test_nested_transaction_contexts_reuse(self, sync_pool_and_model):
                 assert inner_cte.backend() is outer_tx
                 assert inner_tx is outer_tx
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_from_model_in_context(self, sync_pool_and_model):
         """Test CTEQuery created from model query in context."""
         pool, model = sync_pool_and_model
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context_async.py b/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context_async.py
index 0784231..6b2940e 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/connection/test_cte_query_context_async.py
@@ -6,10 +6,13 @@
 context for backend resolution.
 """
 
+from rhosocial.activerecord.backend.dialect.protocols import CTESupport
 from rhosocial.activerecord.query import CTEQuery, AsyncCTEQuery
+from rhosocial.activerecord.testsuite.utils import requires_protocol
 class TestAsyncCTEQueryContext:
     """Test asynchronous CTEQuery context awareness."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_backend_without_context(self, async_pool_and_model):
         """Test AsyncCTEQuery.backend() returns constructor backend without context."""
         pool, model = async_pool_and_model
@@ -19,6 +22,7 @@ async def test_cte_query_backend_without_context(self, async_pool_and_model):
         cte_backend = cte.backend()
         assert cte_backend is original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_backend_in_connection_context(self, async_pool_and_model):
         """Test AsyncCTEQuery.backend() returns connection backend in context."""
         pool, model = async_pool_and_model
@@ -31,6 +35,7 @@ async def test_cte_query_backend_in_connection_context(self, async_pool_and_mode
             assert cte_backend is conn_backend
             assert cte_backend is not original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_backend_in_transaction_context(self, async_pool_and_model):
         """Test AsyncCTEQuery.backend() returns transaction backend in context."""
         pool, model = async_pool_and_model
@@ -43,6 +48,7 @@ async def test_cte_query_backend_in_transaction_context(self, async_pool_and_mod
             assert cte_backend is tx_backend
             assert cte_backend is not original_backend
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_nested_connection_contexts_reuse(self, async_pool_and_model):
         """Test nested async connection contexts reuse for CTE query."""
         pool, model = async_pool_and_model
@@ -58,6 +64,7 @@ async def test_nested_connection_contexts_reuse(self, async_pool_and_model):
                 assert inner_cte.backend() is outer_conn
                 assert inner_conn is outer_conn
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_nested_transaction_contexts_reuse(self, async_pool_and_model):
         """Test nested async transaction contexts reuse for CTE query."""
         pool, model = async_pool_and_model
@@ -73,6 +80,7 @@ async def test_nested_transaction_contexts_reuse(self, async_pool_and_model):
                 assert inner_cte.backend() is outer_tx
                 assert inner_tx is outer_tx
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_from_model_in_context(self, async_pool_and_model):
         """Test AsyncCTEQuery created from model query in context."""
         pool, model = async_pool_and_model
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/conftest.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/conftest.py
new file mode 100644
index 0000000..dcdd89e
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/conftest.py
@@ -0,0 +1,105 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/conftest.py
+"""Fixtures for cross-schema query tests.
+
+Provisioning is delegated to the backend provider through the OPTIONAL
+``setup_schema_fixtures`` method. Providers that do not implement it (or do
+not declare schema support) cause a clean skip — the testsuite itself makes
+no judgment about backend capabilities; that is declared by each backend's
+dialect (``supports_schema``) and consumed via ``@requires_protocol``.
+"""
+
+import pytest
+
+from rhosocial.activerecord.testsuite.core.registry import get_provider_registry
+from rhosocial.activerecord.testsuite.feature.query.conftest import (
+    PROVIDER_KEY_ASYNC,
+    PROVIDER_KEY_SYNC,
+    SCENARIO_PARAMS_ASYNC,
+    SCENARIO_PARAMS_SYNC,
+)
+
+
+def _provider_for(key):
+    registry = get_provider_registry()
+    provider_class = registry.get_provider(key)
+    return provider_class() if provider_class else None
+
+
+@pytest.fixture(scope="function", params=SCENARIO_PARAMS_SYNC)
+def schema_fixtures(request):
+    """(SchemaCustomer, SchemaOrder) bound to two distinct schemas."""
+    provider = _provider_for(PROVIDER_KEY_SYNC)
+    if provider is None:
+        pytest.skip("No testsuite scenarios found")
+    setup = getattr(provider, "setup_schema_fixtures", None)
+    if setup is None:
+        pytest.skip("Provider does not provide cross-schema fixtures")
+
+    scenario = request.param
+    models = setup(scenario)
+    try:
+        yield models
+    finally:
+        cleanup = getattr(provider, "cleanup_after_test", None)
+        if cleanup:
+            cleanup(scenario)
+
+
+@pytest.fixture(scope="function", params=SCENARIO_PARAMS_ASYNC)
+async def async_schema_fixtures(request):
+    """(AsyncSchemaCustomer, AsyncSchemaOrder) bound to two distinct schemas."""
+    provider = _provider_for(PROVIDER_KEY_ASYNC)
+    if provider is None:
+        pytest.skip("No async testsuite scenarios found")
+    setup = getattr(provider, "setup_schema_fixtures", None)
+    if setup is None:
+        pytest.skip("Provider does not provide cross-schema fixtures")
+
+    scenario = request.param
+    models = await setup(scenario)
+    try:
+        yield models
+    finally:
+        cleanup = getattr(provider, "cleanup_after_test", None)
+        if cleanup:
+            await cleanup(scenario)
+
+
+@pytest.fixture(scope="function", params=SCENARIO_PARAMS_SYNC)
+def mixed_schema_fixtures(request):
+    """(User, Order, MixedSchemaOrder): default users/orders plus orders in SCHEMA_A."""
+    provider = _provider_for(PROVIDER_KEY_SYNC)
+    if provider is None:
+        pytest.skip("No testsuite scenarios found")
+    setup = getattr(provider, "setup_mixed_schema_fixtures", None)
+    if setup is None:
+        pytest.skip("Provider does not provide mixed-schema fixtures")
+
+    scenario = request.param
+    models = setup(scenario)
+    try:
+        yield models
+    finally:
+        cleanup = getattr(provider, "cleanup_after_test", None)
+        if cleanup:
+            cleanup(scenario)
+
+
+@pytest.fixture(scope="function", params=SCENARIO_PARAMS_ASYNC)
+async def async_mixed_schema_fixtures(request):
+    """(AsyncUser, AsyncOrder, AsyncMixedSchemaOrder) with orders also in SCHEMA_A."""
+    provider = _provider_for(PROVIDER_KEY_ASYNC)
+    if provider is None:
+        pytest.skip("No async testsuite scenarios found")
+    setup = getattr(provider, "setup_mixed_schema_fixtures", None)
+    if setup is None:
+        pytest.skip("Provider does not provide mixed-schema fixtures")
+
+    scenario = request.param
+    models = await setup(scenario)
+    try:
+        yield models
+    finally:
+        cleanup = getattr(provider, "cleanup_after_test", None)
+        if cleanup:
+            await cleanup(scenario)
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/mixed_schema_models.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/mixed_schema_models.py
new file mode 100644
index 0000000..765a746
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/mixed_schema_models.py
@@ -0,0 +1,25 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/mixed_schema_models.py
+"""Fixture subclasses relocated into a non-default schema.
+
+The table DDL is intentionally identical to the base fixtures: providers
+provision the same ``orders`` shape in both the default namespace and
+``SCHEMA_A``, so only ``__schema_name__`` differs between ``Order`` and
+``MixedSchemaOrder``. Overriding the ``schema_name()`` classmethod would be
+equivalent; subclassing with the attribute keeps the intent declarative.
+"""
+
+from rhosocial.activerecord.testsuite.feature.query.fixtures.async_models import AsyncOrder
+from rhosocial.activerecord.testsuite.feature.query.fixtures.models import Order
+from rhosocial.activerecord.testsuite.feature.query.fixtures.schema_models import SCHEMA_A
+
+
+class MixedSchemaOrder(Order):
+    """Base ``orders`` fixture living in a non-default schema."""
+
+    __schema_name__ = SCHEMA_A
+
+
+class AsyncMixedSchemaOrder(AsyncOrder):
+    """Async variant of ``MixedSchemaOrder``."""
+
+    __schema_name__ = SCHEMA_A
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join.py
new file mode 100644
index 0000000..f7a6658
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join.py
@@ -0,0 +1,82 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join.py
+"""
+Cross-schema ActiveRecord / ActiveQuery tests.
+
+Two models live in two DIFFERENT schema namespaces (see fixtures/schema_models.py):
+``ar_crm.customers`` and ``ar_shop.orders``. Every generated statement must
+carry the model's own fully-qualified name, so joins, filters and writes work
+across schemas without any search_path manipulation.
+
+Backends without schema namespaces skip declaratively:
+@requires_protocol(SchemaSupport, "supports_schema").
+"""
+
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_cross_schema_join(schema_fixtures):
+    """JOIN across two schemas with WHERE + ORDER BY."""
+    Customer, Order = schema_fixtures
+
+    alice = Customer(name="alice")
+    bob = Customer(name="bob")
+    alice.save()
+    bob.save()
+
+    for cust, amount in ((alice, 100), (bob, 200), (alice, 300)):
+        Order(customer_id=cust.id, amount=amount).save()
+
+    rows = (
+        Order.query()
+        .join(Customer, on=Order.c.customer_id == Customer.c.id)
+        .select(Order.c.amount, Customer.c.name.as_("customer"))
+        .where(Customer.c.name == "alice")
+        .order_by(Order.c.amount)
+        .aggregate()
+    )
+    assert [(r["amount"], r["customer"]) for r in rows] == [(100, "alice"), (300, "alice")]
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_cross_schema_count_with_condition(schema_fixtures):
+    """Scalar COUNT over a cross-schema join."""
+    Customer, Order = schema_fixtures
+
+    cust = Customer(name="counted")
+    cust.save()
+    for i in range(3):
+        Order(customer_id=cust.id, amount=i).save()
+
+    total = (
+        Order.query()
+        .join(Customer, on=Order.c.customer_id == Customer.c.id)
+        .where(Customer.c.name == "counted")
+        .count()
+    )
+    assert total == 3
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_cross_schema_writes_are_scoped(schema_fixtures):
+    """save() lands in the owning schema; update/delete stay scoped by qualifiers."""
+    Customer, Order = schema_fixtures
+
+    cust = Customer(name="writer")
+    cust.save()
+    order = Order(customer_id=cust.id, amount=42)
+    order.save()
+
+    assert Order.query().where(Order.c.id == order.id).count() == 1
+    assert Customer.query().where(Customer.c.id == cust.id).count() == 1
+
+    Order.query().where(Order.c.id == order.id).update_all({"amount": 43})
+    refreshed = Order.query().where(Order.c.id == order.id).one()
+    assert refreshed.amount == 43
+
+    Order.query().where(Order.c.id == order.id).delete_all()
+    assert Order.query().where(Order.c.id == order.id).count() == 0
+    # The other schema is unaffected.
+    assert Customer.query().where(Customer.c.id == cust.id).count() == 1
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join_async.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join_async.py
new file mode 100644
index 0000000..9abc1dd
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join_async.py
@@ -0,0 +1,55 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_cross_schema_join_async.py
+"""Async variant of the cross-schema ActiveRecord / ActiveQuery tests."""
+
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+async def test_async_cross_schema_join(async_schema_fixtures):
+    """JOIN across two schemas with WHERE + ORDER BY (async)."""
+    AsyncCustomer, AsyncOrder = async_schema_fixtures
+
+    alice = AsyncCustomer(name="async_alice")
+    await alice.save()
+    bob = AsyncCustomer(name="async_bob")
+    await bob.save()
+
+    for cust, amount in ((alice, 100), (bob, 200), (alice, 300)):
+        await AsyncOrder(customer_id=cust.id, amount=amount).save()
+
+    rows = await (
+        AsyncOrder.query()
+        .join(AsyncCustomer, on=AsyncOrder.c.customer_id == AsyncCustomer.c.id)
+        .select(AsyncOrder.c.amount, AsyncCustomer.c.name.as_("customer"))
+        .where(AsyncCustomer.c.name == "async_alice")
+        .order_by(AsyncOrder.c.amount)
+        .aggregate()
+    )
+    assert [(r["amount"], r["customer"]) for r in rows] == [
+        (100, "async_alice"),
+        (300, "async_alice"),
+    ]
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+async def test_async_cross_schema_writes_are_scoped(async_schema_fixtures):
+    """Writes land in the owning schema and stay scoped by qualifiers (async)."""
+    AsyncCustomer, AsyncOrder = async_schema_fixtures
+
+    cust = AsyncCustomer(name="async_writer")
+    await cust.save()
+    order = AsyncOrder(customer_id=cust.id, amount=42)
+    await order.save()
+
+    assert await AsyncOrder.query().where(AsyncOrder.c.id == order.id).count() == 1
+    assert await AsyncCustomer.query().where(AsyncCustomer.c.id == cust.id).count() == 1
+
+    await AsyncOrder.query().where(AsyncOrder.c.id == order.id).update_all({"amount": 43})
+    refreshed = await AsyncOrder.query().where(AsyncOrder.c.id == order.id).one()
+    assert refreshed.amount == 43
+
+    await AsyncOrder.query().where(AsyncOrder.c.id == order.id).delete_all()
+    assert await AsyncOrder.query().where(AsyncOrder.c.id == order.id).count() == 0
+    assert await AsyncCustomer.query().where(AsyncCustomer.c.id == cust.id).count() == 1
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema.py
new file mode 100644
index 0000000..37f8852
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema.py
@@ -0,0 +1,91 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema.py
+"""Default-schema vs non-default-schema operations for identical tables.
+
+``Order`` (default namespace) and ``MixedSchemaOrder`` (``SCHEMA_A``) share
+the same table DDL; only ``__schema_name__`` differs. Providers provision
+both namespaces plus the default ``users`` table so joins can span the
+schema boundary.
+"""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_same_named_tables_coexist(mixed_schema_fixtures):
+    """Identical table names in default and SCHEMA_A hold independent rows."""
+    User, Order, MixedSchemaOrder = mixed_schema_fixtures
+
+    user = User(username="coexist", email="coexist@example.com", age=30)
+    user.save()
+    Order(user_id=user.id, order_number="def-1", total_amount=10).save()
+    Order(user_id=user.id, order_number="def-2", total_amount=20).save()
+    MixedSchemaOrder(user_id=user.id, order_number="sch-1", total_amount=100).save()
+    MixedSchemaOrder(user_id=user.id, order_number="sch-2", total_amount=200).save()
+    MixedSchemaOrder(user_id=user.id, order_number="sch-3", total_amount=300).save()
+
+    assert Order.query().count() == 2
+    assert MixedSchemaOrder.query().count() == 3
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_primary_key_lookup_is_namespace_scoped(mixed_schema_fixtures):
+    """The same PK value resolves to different rows per namespace."""
+    User, Order, MixedSchemaOrder = mixed_schema_fixtures
+
+    user = User(username="pkcheck", email="pkcheck@example.com", age=30)
+    user.save()
+    Order(user_id=user.id, order_number="default-row").save()
+    MixedSchemaOrder(user_id=user.id, order_number="schema-row").save()
+
+    assert Order.query().where(Order.c.id == 1).one().order_number == "default-row"
+    assert (
+        MixedSchemaOrder.query().where(MixedSchemaOrder.c.id == 1).one().order_number
+        == "schema-row"
+    )
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_update_all_and_delete_all_stay_scoped(mixed_schema_fixtures):
+    """Bulk writes on one namespace never leak into the other."""
+    User, Order, MixedSchemaOrder = mixed_schema_fixtures
+
+    user = User(username="scoped", email="scoped@example.com", age=30)
+    user.save()
+    Order(user_id=user.id, order_number="d1").save()
+    MixedSchemaOrder(user_id=user.id, order_number="s1").save()
+    MixedSchemaOrder(user_id=user.id, order_number="s2").save()
+
+    MixedSchemaOrder.query().where(MixedSchemaOrder.c.user_id == user.id).update_all(
+        {"status": "shipped"}
+    )
+    assert Order.query().where(Order.c.status == "pending").count() == 1
+    assert MixedSchemaOrder.query().where(MixedSchemaOrder.c.status == "shipped").count() == 2
+
+    MixedSchemaOrder.query().where(MixedSchemaOrder.c.order_number == "s1").delete_all()
+    assert MixedSchemaOrder.query().count() == 1
+    assert Order.query().count() == 1
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_join_default_user_with_schema_order(mixed_schema_fixtures):
+    """JOIN from the default namespace to a schema-qualified table."""
+    User, Order, MixedSchemaOrder = mixed_schema_fixtures
+
+    alice = User(username="alice", email="alice@example.com", age=30)
+    bob = User(username="bob", email="bob@example.com", age=30)
+    alice.save()
+    bob.save()
+    MixedSchemaOrder(user_id=alice.id, order_number="j1", total_amount=5).save()
+    MixedSchemaOrder(user_id=bob.id, order_number="j2", total_amount=7).save()
+    MixedSchemaOrder(user_id=alice.id, order_number="j3", total_amount=9).save()
+
+    rows = (
+        MixedSchemaOrder.query()
+        .join(User, on=MixedSchemaOrder.c.user_id == User.c.id)
+        .select(MixedSchemaOrder.c.order_number, User.c.username.as_("owner"))
+        .where(User.c.username == "alice")
+        .order_by(MixedSchemaOrder.c.total_amount)
+        .aggregate()
+    )
+    assert [(r["order_number"], r["owner"]) for r in rows] == [("j1", "alice"), ("j3", "alice")]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_async.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_async.py
new file mode 100644
index 0000000..ca8959f
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_async.py
@@ -0,0 +1,66 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_async.py
+"""Async default-schema vs non-default-schema operations for identical tables."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+async def test_async_same_named_tables_coexist(async_mixed_schema_fixtures):
+    """Identical table names in default and SCHEMA_A hold independent rows."""
+    AsyncUser, AsyncOrder, AsyncMixedSchemaOrder = async_mixed_schema_fixtures
+
+    user = AsyncUser(username="coexist", email="coexist@example.com", age=30)
+    await user.save()
+    await AsyncOrder(user_id=user.id, order_number="def-1", total_amount=10).save()
+    await AsyncMixedSchemaOrder(
+        user_id=user.id, order_number="sch-1", total_amount=100
+    ).save()
+
+    assert await AsyncOrder.query().count() == 1
+    assert await AsyncMixedSchemaOrder.query().count() == 1
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+async def test_async_update_all_stays_scoped(async_mixed_schema_fixtures):
+    """Bulk update on the schema-qualified namespace leaves defaults untouched."""
+    AsyncUser, AsyncOrder, AsyncMixedSchemaOrder = async_mixed_schema_fixtures
+
+    user = AsyncUser(username="scoped", email="scoped@example.com", age=30)
+    await user.save()
+    await AsyncOrder(user_id=user.id, order_number="d1").save()
+    await AsyncMixedSchemaOrder(user_id=user.id, order_number="s1").save()
+    await AsyncMixedSchemaOrder(user_id=user.id, order_number="s2").save()
+
+    await AsyncMixedSchemaOrder.query().where(
+        AsyncMixedSchemaOrder.c.user_id == user.id
+    ).update_all({"status": "shipped"})
+    assert await AsyncOrder.query().where(AsyncOrder.c.status == "pending").count() == 1
+    assert (
+        await AsyncMixedSchemaOrder.query()
+        .where(AsyncMixedSchemaOrder.c.status == "shipped")
+        .count()
+        == 2
+    )
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+async def test_async_join_default_user_with_schema_order(async_mixed_schema_fixtures):
+    """Async JOIN from the default namespace to a schema-qualified table."""
+    AsyncUser, AsyncOrder, AsyncMixedSchemaOrder = async_mixed_schema_fixtures
+
+    alice = AsyncUser(username="alice", email="alice@example.com", age=30)
+    bob = AsyncUser(username="bob", email="bob@example.com", age=30)
+    await alice.save()
+    await bob.save()
+    await AsyncMixedSchemaOrder(user_id=alice.id, order_number="j1", total_amount=5).save()
+    await AsyncMixedSchemaOrder(user_id=bob.id, order_number="j2", total_amount=7).save()
+
+    rows = (
+        await AsyncMixedSchemaOrder.query()
+        .join(AsyncUser, on=AsyncMixedSchemaOrder.c.user_id == AsyncUser.c.id)
+        .select(AsyncMixedSchemaOrder.c.order_number, AsyncUser.c.username.as_("owner"))
+        .where(AsyncUser.c.username == "alice")
+        .aggregate()
+    )
+    assert [(r["order_number"], r["owner"]) for r in rows] == [("j1", "alice")]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_qualifiers.py b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_qualifiers.py
new file mode 100644
index 0000000..6645bbc
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_qualifiers.py
@@ -0,0 +1,49 @@
+# src/rhosocial/activerecord/testsuite/feature/query/cross_schema/test_mixed_schema_qualifiers.py
+"""Qualifier-form contract for schema-bound models under plain and joined ranges."""
+import re
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+def _norm(sql: str) -> str:
+    """Normalise rendered SQL: strip identifier quotes, fold case/space."""
+    cleaned = sql.replace('"', "").replace("`", "").replace("[", "").replace("]", "")
+    return re.sub(r"\s+", " ", cleaned).lower()
+
+
+@requires_protocol("SchemaSupport", "supports_schema")
+def test_schema_model_qualifies_range_not_columns(mixed_schema_fixtures):
+    """A schema-bound model qualifies its FROM range, never three-part cols."""
+    _, _, MixedSchemaOrder = mixed_schema_fixtures
+
+    sql, _ = (
+        MixedSchemaOrder.query()
+        .select(MixedSchemaOrder.c.order_number)
+        .to_sql()
+    )
+    normed = _norm(sql)
+    # Range keeps the namespace...
+    assert "from ar_crm.orders" in normed
+    # ...while column references never grow a third (schema) part.
+    assert "ar_crm.orders.order_number" not in normed
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_schema_join_columns_stay_two_part(mixed_schema_fixtures):
+    """Even joined against default-schema tables, schema-model columns stay
+    two-part (TABLE.COLUMN) in ON/WHERE positions."""
+    User, _, MixedSchemaOrder = mixed_schema_fixtures
+
+    sql, _ = (
+        MixedSchemaOrder.query()
+        .join(User, on=MixedSchemaOrder.c.user_id == User.c.id)
+        .select(MixedSchemaOrder.c.order_number)
+        .where(User.c.username == "nobody")
+        .to_sql()
+    )
+    normed = _norm(sql)
+    assert "from ar_crm.orders" in normed
+    assert "ar_crm.orders.user_id" not in normed
+    assert "orders.user_id = users.id" in normed
+
+
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query.py b/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query.py
index 2b91861..2a5381d 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query.py
@@ -11,8 +11,13 @@
 import pytest
 from decimal import Decimal
 
+from rhosocial.activerecord.backend.dialect.protocols import CTESupport
 from rhosocial.activerecord.query import CTEQuery, AsyncCTEQuery
-from rhosocial.activerecord.testsuite.utils import requires_cte, requires_recursive_cte
+from rhosocial.activerecord.testsuite.utils import (
+    requires_cte,
+    requires_protocol,
+    requires_recursive_cte,
+)
 class TestCTEQueryActiveQuery:
     """Test CTE queries with ActiveQuery subqueries (synchronous)."""
 
@@ -102,6 +107,7 @@ def test_multiple_active_query_cte(self, tree_fixtures):
 class TestCTEQueryErrorHandling:
     """Test CTE query error handling and validation."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_with_wrong_paradigm_backend_raises_error(self, order_fixtures):
         """
         Test that CTEQuery raises TypeError when an async backend is provided.
@@ -126,6 +132,7 @@ def test_cte_query_with_wrong_paradigm_backend_raises_error(self, order_fixtures
         assert "StorageBackend" in str(exc_info.value)
         assert "Mock" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_with_mock_query_raises_error(self, order_fixtures):
         """
         Test that CTEQuery raises TypeError when an invalid query type (mock) is provided to with_cte.
@@ -152,6 +159,7 @@ def test_cte_query_with_mock_query_raises_error(self, order_fixtures):
         assert "not supported in CTE" in str(exc_info.value)
         assert "Only str, SQLQueryAndParams, IQuery, and QueryExpression" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_with_wrong_paradigm_query_raises_error(self, order_fixtures):
         """
         Test that CTEQuery raises TypeError when an async query is provided to with_cte.
@@ -186,6 +194,7 @@ def test_cte_query_with_wrong_paradigm_query_raises_error(self, order_fixtures):
         assert "CTEQuery (sync) cannot accept async query" in str(exc_info.value)
         assert "AsyncActiveQuery" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_with_invalid_query_type_raises_error(self, order_fixtures):
         """
         Test that CTEQuery raises TypeError when an unsupported query type is provided to with_cte.
@@ -362,6 +371,7 @@ def test_cte_with_joins(self, order_fixtures):
 class TestCTEQuerySyncErrorHandling:
     """Test CTE query error handling for edge cases (synchronous)."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_to_sql_with_empty_ctes_raises_error(self, order_fixtures):
         """
         Test that calling to_sql() on a CTEQuery with no CTEs defined raises ValueError.
@@ -381,6 +391,7 @@ def test_cte_query_to_sql_with_empty_ctes_raises_error(self, order_fixtures):
         # Verify the error message mentions the missing CTE requirement
         assert "CTEQuery must have at least one CTE defined" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     def test_cte_query_to_sql_with_nonexistent_main_cte_name_raises_error(self, order_fixtures):
         """
         Test that calling to_sql() with a _main_cte_name that doesn't exist raises ValueError.
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query_async.py b/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query_async.py
index 5457209..c205e60 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/cte/test_cte_query_active_query_async.py
@@ -11,8 +11,13 @@
 import pytest
 from decimal import Decimal
 
+from rhosocial.activerecord.backend.dialect.protocols import CTESupport
 from rhosocial.activerecord.query import CTEQuery, AsyncCTEQuery
-from rhosocial.activerecord.testsuite.utils import requires_cte, requires_recursive_cte
+from rhosocial.activerecord.testsuite.utils import (
+    requires_cte,
+    requires_protocol,
+    requires_recursive_cte,
+)
 class TestAsyncCTEQueryActiveQuery:
     """Test CTE queries with ActiveQuery subqueries (asynchronous)."""
 
@@ -102,6 +107,7 @@ async def test_multiple_active_query_cte(self, async_tree_fixtures):
 class TestAsyncCTEQueryErrorHandling:
     """Test CTE query error handling and validation for async."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_with_wrong_paradigm_backend_raises_error(self, async_order_fixtures):
         """
         Test that AsyncCTEQuery raises TypeError when a sync backend is provided.
@@ -123,6 +129,7 @@ async def test_cte_query_with_wrong_paradigm_backend_raises_error(self, async_or
         assert "AsyncCTEQuery requires an AsyncStorageBackend" in str(exc_info.value)
         assert "StorageBackend" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_with_mock_query_raises_error(self, async_order_fixtures):
         """
         Test that AsyncCTEQuery raises TypeError when an invalid query type (mock) is provided to with_cte.
@@ -142,6 +149,7 @@ async def test_cte_query_with_mock_query_raises_error(self, async_order_fixtures
         assert "not supported in CTE" in str(exc_info.value)
         assert "Only str, SQLQueryAndParams, IQuery, and QueryExpression" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_with_wrong_paradigm_query_raises_error(self, async_order_fixtures):
         """
         Test that AsyncCTEQuery raises TypeError when a sync query is provided to with_cte.
@@ -167,6 +175,7 @@ async def test_cte_query_with_wrong_paradigm_query_raises_error(self, async_orde
         assert "AsyncCTEQuery (async) cannot accept sync query" in str(exc_info.value)
         assert "ActiveQuery" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_with_invalid_query_type_raises_error(self, async_order_fixtures):
         """
         Test that CTEQuery raises TypeError when an unsupported query type is provided to with_cte.
@@ -338,6 +347,7 @@ async def test_cte_with_joins(self, async_order_fixtures):
 class TestAsyncCTEQuerySyncErrorHandling:
     """Test CTE query error handling for edge cases (asynchronous)."""
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_to_sql_with_empty_ctes_raises_error(self, async_order_fixtures):
         """
         Test that calling to_sql() on an AsyncCTEQuery with no CTEs defined raises ValueError.
@@ -353,6 +363,7 @@ async def test_cte_query_to_sql_with_empty_ctes_raises_error(self, async_order_f
 
         assert "CTEQuery must have at least one CTE defined" in str(exc_info.value)
 
+    @requires_protocol(CTESupport, "supports_basic_cte")
     async def test_cte_query_to_sql_with_nonexistent_main_cte_name_raises_error(self, async_order_fixtures):
         """
         Test that calling to_sql() with a _main_cte_name that doesn't exist raises ValueError.
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/fixtures/schema_models.py b/src/rhosocial/activerecord/testsuite/feature/query/fixtures/schema_models.py
new file mode 100644
index 0000000..3b34ecb
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/fixtures/schema_models.py
@@ -0,0 +1,61 @@
+# src/rhosocial/activerecord/testsuite/feature/query/fixtures/schema_models.py
+"""Models bound to explicit, NON-default schema namespaces.
+
+Used by the cross-schema query tests. The schema names are constants shared
+with backend providers: the provider provisions ``SCHEMA_A`` / ``SCHEMA_B``
+and the tables inside them, then returns these configured model classes.
+
+Only backends whose dialect declares ``supports_schema()`` will run these
+tests; everyone else skips via ``@requires_protocol``.
+"""
+
+from typing import ClassVar, Optional
+
+from rhosocial.activerecord.base.field_proxy import FieldProxy
+from rhosocial.activerecord.model import ActiveRecord, AsyncActiveRecord
+
+# Non-default schema namespaces provisioned by supporting backends.
+SCHEMA_A = "ar_crm"    # customers live here
+SCHEMA_B = "ar_shop"   # orders live here
+
+
+class SchemaCustomer(ActiveRecord):
+    __table_name__ = "customers"
+    __schema_name__ = SCHEMA_A
+    __primary_key__ = "id"
+    c: ClassVar[FieldProxy] = FieldProxy()
+
+    id: Optional[int] = None
+    name: str
+
+
+class SchemaOrder(ActiveRecord):
+    __table_name__ = "orders"
+    __schema_name__ = SCHEMA_B
+    __primary_key__ = "id"
+    c: ClassVar[FieldProxy] = FieldProxy()
+
+    id: Optional[int] = None
+    customer_id: int
+    amount: int
+
+
+class AsyncSchemaCustomer(AsyncActiveRecord):
+    __table_name__ = "customers"
+    __schema_name__ = SCHEMA_A
+    __primary_key__ = "id"
+    c: ClassVar[FieldProxy] = FieldProxy()
+
+    id: Optional[int] = None
+    name: str
+
+
+class AsyncSchemaOrder(AsyncActiveRecord):
+    __table_name__ = "orders"
+    __schema_name__ = SCHEMA_B
+    __primary_key__ = "id"
+    c: ClassVar[FieldProxy] = FieldProxy()
+
+    id: Optional[int] = None
+    customer_id: int
+    amount: int
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate.py
new file mode 100644
index 0000000..43dbb56
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate.py
@@ -0,0 +1,61 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate.py
+"""JOIN combined with aggregation: GROUP BY / HAVING over joined ranges."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+from rhosocial.activerecord.backend.expression import functions
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_join_group_by_count(order_fixtures):
+    """Count orders per user across the join."""
+    User, Order, _ = order_fixtures
+    dialect = User.__backend__.dialect
+
+    alice = User(username="agg_alice", email="aa@example.com", age=30)
+    bob = User(username="agg_bob", email="ab@example.com", age=31)
+    alice.save()
+    bob.save()
+    Order(user_id=alice.id, order_number="g1").save()
+    Order(user_id=alice.id, order_number="g2").save()
+    Order(user_id=bob.id, order_number="g3").save()
+
+    rows = (
+        User.query()
+        .join(Order, on=User.c.id == Order.c.user_id)
+        .select(User.c.username, functions.count(dialect, Order.c.id).as_("n_orders"))
+        .group_by(User.c.username)
+        .aggregate()
+    )
+    counts = {r["username"]: r["n_orders"] for r in rows}
+    assert counts == {"agg_alice": 2, "agg_bob": 1}
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_join_group_by_sum_having(order_fixtures):
+    """HAVING filters aggregated groups after a join."""
+    User, Order, _ = order_fixtures
+    dialect = User.__backend__.dialect
+
+    big = User(username="big_spender", email="bs@example.com", age=50)
+    small = User(username="small_fry", email="sf@example.com", age=22)
+    big.save()
+    small.save()
+    for i in range(3):
+        Order(user_id=big.id, order_number=f"b{i}", total_amount=100 + i).save()
+    Order(user_id=small.id, order_number="s1", total_amount=10).save()
+
+    rows = (
+        User.query()
+        .join(Order, on=User.c.id == Order.c.user_id)
+        .select(
+            User.c.username,
+            functions.sum_(dialect, Order.c.total_amount).as_("spent"),
+            functions.count(dialect, Order.c.id).as_("n"),
+        )
+        .group_by(User.c.username)
+        .having(functions.count(dialect, Order.c.id) > 1)
+        .aggregate()
+    )
+    assert [(r["username"], r["spent"], r["n"]) for r in rows] == [("big_spender", 303, 3)]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate_async.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate_async.py
new file mode 100644
index 0000000..cc4b1fe
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate_async.py
@@ -0,0 +1,59 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_aggregate_async.py
+"""Async JOIN + aggregation coverage."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+from rhosocial.activerecord.backend.expression import functions
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+async def test_join_group_by_count(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+    dialect = AsyncUser.__backend__.dialect
+
+    alice = AsyncUser(username="agg_alice", email="aa@example.com", age=30)
+    bob = AsyncUser(username="agg_bob", email="ab@example.com", age=31)
+    await alice.save()
+    await bob.save()
+    await AsyncOrder(user_id=alice.id, order_number="g1").save()
+    await AsyncOrder(user_id=alice.id, order_number="g2").save()
+    await AsyncOrder(user_id=bob.id, order_number="g3").save()
+
+    rows = (
+        await AsyncUser.query()
+        .join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .select(AsyncUser.c.username, functions.count(dialect, AsyncOrder.c.id).as_("n_orders"))
+        .group_by(AsyncUser.c.username)
+        .aggregate()
+    )
+    counts = {r["username"]: r["n_orders"] for r in rows}
+    assert counts == {"agg_alice": 2, "agg_bob": 1}
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+async def test_join_group_by_sum_having(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+    dialect = AsyncUser.__backend__.dialect
+
+    big = AsyncUser(username="big_spender", email="bs@example.com", age=50)
+    small = AsyncUser(username="small_fry", email="sf@example.com", age=22)
+    await big.save()
+    await small.save()
+    for i in range(3):
+        await AsyncOrder(user_id=big.id, order_number=f"b{i}", total_amount=100 + i).save()
+    await AsyncOrder(user_id=small.id, order_number="s1", total_amount=10).save()
+
+    rows = (
+        await AsyncUser.query()
+        .join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .select(
+            AsyncUser.c.username,
+            functions.sum_(dialect, AsyncOrder.c.total_amount).as_("spent"),
+            functions.count(dialect, AsyncOrder.c.id).as_("n"),
+        )
+        .group_by(AsyncUser.c.username)
+        .having(functions.count(dialect, AsyncOrder.c.id) > 1)
+        .aggregate()
+    )
+    assert [(r["username"], r["spent"], r["n"]) for r in rows] == [("big_spender", 303, 3)]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_qualifiers.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_qualifiers.py
new file mode 100644
index 0000000..152e9b4
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_qualifiers.py
@@ -0,0 +1,72 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_qualifiers.py
+"""Column-qualifier resolution contract under JOINs.
+
+Locks the generated SQL *forms* (after quote/case normalisation) so dialect
+regressions surface locally instead of as live-server errors:
+
+- JOIN ranges keep schema qualifiers; column references stay two-part.
+- Aliased self-join ranges are referenced by alias only.
+"""
+import re
+
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+def _norm(sql: str) -> str:
+    """Normalise rendered SQL: strip identifier quotes, fold case/space."""
+    cleaned = sql.replace('"', "").replace("`", "").replace("[", "").replace("]", "")
+    return re.sub(r"\s+", " ", cleaned).lower()
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_join_columns_are_table_qualified(order_fixtures):
+    """ON predicates and select lists qualify columns with their table."""
+    User, Order, _ = order_fixtures
+
+    user = User(username="qual_user", email="qu@example.com", age=30)
+    user.save()
+    Order(user_id=user.id, order_number="q1").save()
+
+    query = (
+        User.query()
+        .join(Order, on=User.c.id == Order.c.user_id)
+        .select(User.c.username, Order.c.order_number)
+        .where(User.c.username == "qual_user")
+    )
+    sql, params = query.to_sql()
+    normed = _norm(sql)
+    assert "users.id = orders.user_id" in normed
+    assert "orders.user_id" in normed.split(" where ")[0]
+    assert "users.username" in normed
+    assert params == ("qual_user",) or params == ["qual_user"]
+
+    rows = query.aggregate()
+    assert [r["order_number"] for r in rows] == ["q1"]
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+async def test_aliased_self_join_references_alias_only(tree_fixtures):
+    """Self-join columns address the runtime alias, not the base table."""
+    NodeModel = tree_fixtures[0]
+
+    root = NodeModel(name="alias_root", value=1)
+    root.save()
+    kid = NodeModel(name="alias_kid", value=2)
+    kid.parent_id = root.id
+    kid.save()
+
+    child = NodeModel.c.with_table_alias("child")
+    query = NodeModel.query().join(
+        NodeModel, on=child.parent_id == NodeModel.c.id, alias="child"
+    ).select(child.name.as_("child_name"), NodeModel.c.name.as_("parent_name"))
+    sql, _ = query.to_sql()
+    normed = _norm(sql)
+
+    assert "nodes as child" in normed or "nodes child" in normed
+    assert "child.parent_id = nodes.id" in normed
+    assert "child.name" in normed
+
+    rows = query.aggregate()
+    assert [(r["child_name"], r["parent_name"]) for r in rows] == [("alias_kid", "alias_root")]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_subquery.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_subquery.py
new file mode 100644
index 0000000..6e3d1da
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_subquery.py
@@ -0,0 +1,52 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_subquery.py
+"""JOIN combined with subqueries: IN-filters over joined ranges and
+aggregate-subquery comparisons."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_join_filtered_by_in_subquery(order_fixtures):
+    """Only users having a 'big' order survive an IN-subquery filter."""
+    User, Order, _ = order_fixtures
+
+    rich = User(username="sub_rich", email="sr@example.com", age=60)
+    poor = User(username="sub_poor", email="sp@example.com", age=61)
+    rich.save()
+    poor.save()
+    Order(user_id=rich.id, order_number="r1", total_amount=999).save()
+    Order(user_id=poor.id, order_number="p1", total_amount=1).save()
+
+    big_orders = Order.query().select(Order.c.user_id).where(Order.c.total_amount > 100)
+    rows = (
+        User.query()
+        .join(Order, on=User.c.id == Order.c.user_id)
+        .where(User.c.id.in_(big_orders))
+        .select(User.c.username)
+        .aggregate()
+    )
+    assert [r["username"] for r in rows] == ["sub_rich"]
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_three_table_join_with_in_subquery(order_fixtures):
+    """Chained joins remain composable with a subquery predicate."""
+    User, Order, OrderItem = order_fixtures
+
+    user = User(username="chain_sub", email="cs@example.com", age=35)
+    user.save()
+    order = Order(user_id=user.id, order_number="c1")
+    order.save()
+    OrderItem(order_id=order.id, product_name="widget", quantity=5, unit_price=20).save()
+
+    target_items = OrderItem.query().select(OrderItem.c.order_id).where(OrderItem.c.quantity >= 5)
+    rows = (
+        User.query()
+        .join(Order, on=User.c.id == Order.c.user_id)
+        .inner_join(OrderItem, on=Order.c.id == OrderItem.c.order_id)
+        .where(Order.c.id.in_(target_items))
+        .select(User.c.username, Order.c.order_number)
+        .aggregate()
+    )
+    assert [(r["username"], r["order_number"]) for r in rows] == [("chain_sub", "c1")]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types.py
new file mode 100644
index 0000000..618fb40
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types.py
@@ -0,0 +1,122 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types.py
+"""Outer, cross and mixed-type join coverage for ActiveQuery.
+
+Verifies the row-preservation semantics of each JOIN type (the most common
+regression area) rather than only the happy-path matched rows.
+"""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+def _user_without_orders(users):
+    """Create a user that places no orders; returns the model instance."""
+    lonely = users[0](username="lonely", email="lonely@example.com", age=40)
+    lonely.save()
+    return lonely
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+def test_left_join_preserves_unmatched_left_rows(order_fixtures):
+    """Rows without a match still appear, right-side columns being NULL."""
+    User, Order, _ = order_fixtures
+    _user_without_orders([User])
+
+    rows = (
+        User.query()
+        .left_join(Order, on=User.c.id == Order.c.user_id)
+        .select(User.c.username, Order.c.id.as_("order_id"))
+        .aggregate()
+    )
+    lonely_rows = [r for r in rows if r["username"] == "lonely"]
+    assert len(lonely_rows) == 1
+    assert lonely_rows[0]["order_id"] is None
+
+
+@requires_protocol("JoinSupport", "supports_right_join")
+def test_right_join_preserves_right_side(order_fixtures):
+    """RIGHT JOIN keeps every right-hand row even when left side has gaps."""
+    User, Order, _ = order_fixtures
+    _user_without_orders([User])
+
+    total_orders = Order.query().count()
+    joined = (
+        Order.query()
+        .right_join(User, on=Order.c.user_id == User.c.id)
+        .select(Order.c.id.as_("order_id"))
+        .aggregate()
+    )
+    assert len(joined) == total_orders
+
+
+@requires_protocol("JoinSupport", "supports_full_join")
+def test_full_outer_join_covers_both_sides(order_fixtures):
+    """FULL OUTER JOIN yields unmatched rows of both sides exactly once."""
+    User, Order, _ = order_fixtures
+    lonely = _user_without_orders([User])
+
+    orders_total = Order.query().count()
+    users_total = User.query().count()
+    users_with_orders = (
+        User.query().join(Order, on=User.c.id == Order.c.user_id).aggregate()
+    )
+
+    rows = (
+        User.query()
+        .full_join(Order, on=User.c.id == Order.c.user_id)
+        .select(User.c.username, Order.c.id.as_("order_id"))
+        .aggregate()
+    )
+    # Every order appears; every user without orders adds one NULL-extended row.
+    matched_users = len({r["username"] for r in rows if r["order_id"] is not None})
+    null_extended = [r for r in rows if r["order_id"] is None]
+    assert len(rows) == orders_total + (users_total - matched_users)
+    assert [r["username"] for r in null_extended] == ["lonely"]
+
+
+@requires_protocol("JoinSupport", "supports_cross_join")
+def test_cross_join_is_cartesian(order_fixtures):
+    """CROSS JOIN produces |left| x |right| rows without an ON clause."""
+    User, Order, _ = order_fixtures
+
+    users_total = User.query().count()
+    orders_total = Order.query().count()
+
+    rows = User.query().cross_join(Order).select(User.c.username).aggregate()
+    assert len(rows) == users_total * orders_total
+
+
+@requires_protocol("JoinSupport", "supports_straight_join")
+def test_straight_join_returns_inner_equivalent(order_fixtures):
+    """STRAIGHT_JOIN is result-equivalent to INNER JOIN (MySQL optimizer hint)."""
+    User, Order, _ = order_fixtures
+
+    inner = User.query().join(Order, on=User.c.id == Order.c.user_id).aggregate()
+    straight = (
+        User.query()
+        .straight_join(Order, on=User.c.id == Order.c.user_id)
+        .aggregate()
+    )
+    assert len(straight) == len(inner)
+
+
+def test_mixed_type_join_chain(order_fixtures):
+    """LEFT and INNER joins can be chained with preserved per-hop semantics."""
+    User, Order, OrderItem = order_fixtures
+
+    lonely = _user_without_orders([User])
+    buyer = User(username="buyer", email="buyer@example.com", age=31)
+    buyer.save()
+    order = Order(user_id=buyer.id, order_number="mix-1", total_amount=50)
+    order.save()
+
+    rows = (
+        User.query()
+        .left_join(Order, on=User.c.id == Order.c.user_id)
+        .inner_join(OrderItem, on=Order.c.id == OrderItem.c.order_id)
+        .select(User.c.username, OrderItem.c.id.as_("item_id"))
+        .aggregate()
+    )
+    # Only users whose orders have items survive the trailing INNER hop;
+    # the lonely user's NULL-extended row cannot satisfy it.
+    assert all(r["username"] != "lonely" for r in rows)
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types_async.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types_async.py
new file mode 100644
index 0000000..f6cde46
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types_async.py
@@ -0,0 +1,102 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_types_async.py
+"""Async variants for outer, cross and mixed-type join coverage."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+async def _user_without_orders(users):
+    lonely = users[0](username="lonely", email="lonely@example.com", age=40)
+    await lonely.save()
+    return lonely
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+async def test_left_join_preserves_unmatched_left_rows(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+    await _user_without_orders([AsyncUser])
+
+    rows = (
+        await AsyncUser.query()
+        .left_join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .select(AsyncUser.c.username, AsyncOrder.c.id.as_("order_id"))
+        .aggregate()
+    )
+    lonely_rows = [r for r in rows if r["username"] == "lonely"]
+    assert len(lonely_rows) == 1
+    assert lonely_rows[0]["order_id"] is None
+
+
+@requires_protocol("JoinSupport", "supports_right_join")
+async def test_right_join_preserves_right_side(async_order_fixtures):
+    AsyncOrder, AsyncUser = async_order_fixtures[1], async_order_fixtures[0]
+    await _user_without_orders([async_order_fixtures[0]])
+
+    total_orders = await AsyncOrder.query().count()
+    joined = (
+        await AsyncOrder.query()
+        .right_join(AsyncUser, on=AsyncOrder.c.user_id == AsyncUser.c.id)
+        .select(AsyncOrder.c.id.as_("order_id"))
+        .aggregate()
+    )
+    assert len(joined) == total_orders
+
+
+@requires_protocol("JoinSupport", "supports_full_join")
+async def test_full_outer_join_covers_both_sides(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+    await _user_without_orders([AsyncUser])
+
+    orders_total = await AsyncOrder.query().count()
+    users_total = await AsyncUser.query().count()
+
+    rows = (
+        await AsyncUser.query()
+        .full_join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .select(AsyncUser.c.username, AsyncOrder.c.id.as_("order_id"))
+        .aggregate()
+    )
+    matched_users = len({r["username"] for r in rows if r["order_id"] is not None})
+    assert len(rows) == orders_total + (users_total - matched_users)
+
+
+@requires_protocol("JoinSupport", "supports_cross_join")
+async def test_cross_join_is_cartesian(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+
+    users_total = await AsyncUser.query().count()
+    orders_total = await AsyncOrder.query().count()
+
+    rows = await (AsyncUser.query().cross_join(AsyncOrder).select(AsyncUser.c.username)).aggregate()
+    assert len(rows) == users_total * orders_total
+
+
+@requires_protocol("JoinSupport", "supports_straight_join")
+async def test_straight_join_returns_inner_equivalent(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+
+    inner = await (AsyncUser.query().join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)).aggregate()
+    straight = (
+        await AsyncUser.query()
+        .straight_join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .aggregate()
+    )
+    assert len(straight) == len(inner)
+
+
+async def test_mixed_type_join_chain(async_order_fixtures):
+    AsyncUser, AsyncOrder, AsyncOrderItem = async_order_fixtures
+
+    buyer = AsyncUser(username="buyer", email="buyer@example.com", age=31)
+    await buyer.save()
+    order = AsyncOrder(user_id=buyer.id, order_number="mix-1", total_amount=50)
+    await order.save()
+
+    rows = (
+        await AsyncUser.query()
+        .left_join(AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id)
+        .inner_join(AsyncOrderItem, on=AsyncOrder.c.id == AsyncOrderItem.c.order_id)
+        .select(AsyncUser.c.username, AsyncOrderItem.c.id.as_("item_id"))
+        .aggregate()
+    )
+    assert all(r["username"] != "lonely" for r in rows)
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using.py
new file mode 100644
index 0000000..da05dfd
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using.py
@@ -0,0 +1,66 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using.py
+"""USING-clause join coverage.
+
+``JoinExpression`` renders ``... JOIN ... USING ("col")`` when ``using``
+columns are given instead of an ON predicate; the ActiveQuery convenience
+methods forward the ``using`` keyword.
+"""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_inner_join_with_using_column(order_fixtures):
+    """JOIN ... USING ("id") pairs the tables' identical surrogate keys.
+
+    Fresh per-test tables make autoincrement ids deterministic (both start
+    at 1), giving an artificial but fully valid USING pairing.
+    """
+    User, Order, _ = order_fixtures
+
+    user = User(username="using_user", email="using@example.com", age=33)
+    user.save()
+    Order(user_id=user.id, order_number="u1", total_amount=11).save()
+
+    rows = (
+        User.query()
+        .inner_join(Order, using=["id"])
+        .select(User.c.username, Order.c.order_number)
+        .aggregate()
+    )
+    assert [r["order_number"] for r in rows] == ["u1"]
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+def test_left_join_with_using(order_fixtures):
+    """LEFT JOIN ... USING keeps unmatched left rows like its ON counterpart."""
+    User, Order, _ = order_fixtures
+
+    lonely = User(username="lonely_using", email="lu@example.com", age=44)
+    lonely.save()
+    buyer = User(username="buyer_using", email="bu@example.com", age=45)
+    buyer.save()
+    # Advance the orders sequence so this order's id lands on buyer.id,
+    # letting USING ("id") pair them while lonely keeps no match.
+    Order(user_id=lonely.id, order_number="tmp").save()
+    Order.query().where(Order.c.order_number == "tmp").delete_all()
+    Order(user_id=buyer.id, order_number="u2", total_amount=12).save()
+
+    rows = (
+        User.query()
+        .left_join(Order, using=["id"])
+        .select(User.c.username, Order.c.id.as_("order_id"))
+        .aggregate()
+    )
+    by_name = {r["username"]: r["order_id"] for r in rows}
+    assert by_name["lonely_using"] is None
+    assert by_name["buyer_using"] is not None
+
+
+def test_on_and_using_are_mutually_exclusive(order_fixtures):
+    """Passing both raises instead of generating ambiguous SQL."""
+    User, Order, _ = order_fixtures
+
+    with pytest.raises(ValueError):
+        User.query().join(Order, on=User.c.id == Order.c.user_id, using=["user_id"])
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using_async.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using_async.py
new file mode 100644
index 0000000..2786ab8
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using_async.py
@@ -0,0 +1,56 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_join_using_async.py
+"""Async USING-clause join coverage."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+async def test_inner_join_with_using_column(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+
+    user = AsyncUser(username="using_user", email="using@example.com", age=33)
+    await user.save()
+    await AsyncOrder(user_id=user.id, order_number="u1", total_amount=11).save()
+
+    rows = (
+        await AsyncUser.query()
+        .inner_join(AsyncOrder, using=["id"])
+        .select(AsyncUser.c.username, AsyncOrder.c.order_number)
+        .aggregate()
+    )
+    assert [r["order_number"] for r in rows] == ["u1"]
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+async def test_left_join_with_using(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+
+    lonely = AsyncUser(username="lonely_using", email="lu@example.com", age=44)
+    await lonely.save()
+    buyer = AsyncUser(username="buyer_using", email="bu@example.com", age=45)
+    await buyer.save()
+    # Advance the orders sequence so this order's id lands on buyer.id,
+    # letting USING ("id") pair them while lonely keeps no match.
+    await AsyncOrder(user_id=lonely.id, order_number="tmp").save()
+    await AsyncOrder.query().where(AsyncOrder.c.order_number == "tmp").delete_all()
+    await AsyncOrder(user_id=buyer.id, order_number="u2", total_amount=12).save()
+
+    rows = (
+        await AsyncUser.query()
+        .left_join(AsyncOrder, using=["id"])
+        .select(AsyncUser.c.username, AsyncOrder.c.id.as_("order_id"))
+        .aggregate()
+    )
+    by_name = {r["username"]: r["order_id"] for r in rows}
+    assert by_name["lonely_using"] is None
+    assert by_name["buyer_using"] is not None
+
+
+async def test_on_and_using_are_mutually_exclusive(async_order_fixtures):
+    AsyncUser, AsyncOrder, _ = async_order_fixtures
+
+    with pytest.raises(ValueError):
+        AsyncUser.query().join(
+            AsyncOrder, on=AsyncUser.c.id == AsyncOrder.c.user_id, using=["user_id"]
+        )
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join.py
new file mode 100644
index 0000000..9909f35
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join.py
@@ -0,0 +1,85 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join.py
+"""True self-join coverage: one table joined to itself under two aliases.
+
+The ActiveRecord-level mechanism is a runtime-derived second proxy:
+``Node.c.with_table_alias(alias)`` yields a static ``alias.column``
+accessor that pairs with ``join(Node, on=..., alias=alias)``.
+"""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+from rhosocial.activerecord.backend.expression import functions
+
+from rhosocial.activerecord.testsuite.feature.query.fixtures.cte_models import Node
+
+
+def _seed_tree(tree_fixtures):
+    """root -> (a, b); a -> a1. Returns nodes keyed by name."""
+    NodeModel = tree_fixtures[0]
+    named = {}
+    for name, parent, value in (
+        ("root", None, 1),
+        ("a", "root", 10),
+        ("b", "root", 20),
+        ("a1", "a", 100),
+    ):
+        node = NodeModel(name=name, value=value)
+        if parent is not None:
+            node.parent_id = named[parent].id
+        node.save()
+        named[name] = node
+    return named
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_self_join_child_to_parent(tree_fixtures):
+    _seed_tree(tree_fixtures)
+
+    child = Node.c.with_table_alias("child")
+    rows = (
+        Node.query()
+        .join(Node, on=child.parent_id == Node.c.id, alias="child")
+        .select(child.name.as_("child_name"), Node.c.name.as_("parent_name"))
+        .where(Node.c.name == "root")
+        .order_by(child.name)
+        .aggregate()
+    )
+    assert [(r["child_name"], r["parent_name"]) for r in rows] == [("a", "root"), ("b", "root")]
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+def test_self_join_left_keeps_roots(tree_fixtures):
+    """LEFT self-join keeps parents without parents of their own."""
+    _seed_tree(tree_fixtures)
+
+    child = Node.c.with_table_alias("kid")
+    rows = (
+        Node.query()
+        .left_join(Node, on=child.parent_id == Node.c.id, alias="kid")
+        .select(Node.c.name.as_("node"), child.name.as_("child_name"))
+        .where(Node.c.parent_id.is_null())
+        .aggregate()
+    )
+    assert [(r["node"], r["child_name"]) for r in rows] == [("root", "a"), ("root", "b")]
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+def test_self_join_aggregates_children_per_parent(tree_fixtures):
+    _seed_tree(tree_fixtures)
+
+    child = Node.c.with_table_alias("kid")
+    dialect = Node.__backend__.dialect
+    rows = (
+        Node.query()
+        .join(Node, on=child.parent_id == Node.c.id, alias="kid")
+        .select(
+            Node.c.name.as_("parent"),
+            functions.count(dialect, child.value).as_("n_children"),
+            functions.sum_(dialect, child.value).as_("value_sum"),
+        )
+        .group_by(Node.c.name)
+        .having(functions.count(dialect, child.value) > 1)
+        .aggregate()
+    )
+    assert [(r["parent"], r["n_children"], r["value_sum"]) for r in rows] == [("root", 2, 30)]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join_async.py b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join_async.py
new file mode 100644
index 0000000..5ba7081
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join_async.py
@@ -0,0 +1,54 @@
+# src/rhosocial/activerecord/testsuite/feature/query/joins/test_self_join_async.py
+"""Async self-join coverage via runtime-derived aliased proxies."""
+import pytest
+
+from rhosocial.activerecord.testsuite.utils import requires_protocol
+
+from rhosocial.activerecord.testsuite.feature.query.fixtures.async_cte_models import AsyncNode
+
+
+async def _seed_tree(async_tree_fixtures):
+    NodeModel = async_tree_fixtures[0]
+    named = {}
+    for name, parent, value in (
+        ("root", None, 1),
+        ("a", "root", 10),
+        ("b", "root", 20),
+    ):
+        node = NodeModel(name=name, value=value)
+        if parent is not None:
+            node.parent_id = named[parent].id
+        await node.save()
+        named[name] = node
+    return named
+
+
+@requires_protocol("JoinSupport", "supports_inner_join")
+async def test_async_self_join_child_to_parent(async_tree_fixtures):
+    await _seed_tree(async_tree_fixtures)
+
+    child = AsyncNode.c.with_table_alias("child")
+    rows = (
+        await AsyncNode.query()
+        .join(AsyncNode, on=child.parent_id == AsyncNode.c.id, alias="child")
+        .select(child.name.as_("child_name"), AsyncNode.c.name.as_("parent_name"))
+        .where(AsyncNode.c.name == "root")
+        .order_by(child.name)
+        .aggregate()
+    )
+    assert [(r["child_name"], r["parent_name"]) for r in rows] == [("a", "root"), ("b", "root")]
+
+
+@requires_protocol("JoinSupport", "supports_left_join")
+async def test_async_self_join_left_keeps_roots(async_tree_fixtures):
+    await _seed_tree(async_tree_fixtures)
+
+    child = AsyncNode.c.with_table_alias("kid")
+    rows = (
+        await AsyncNode.query()
+        .left_join(AsyncNode, on=child.parent_id == AsyncNode.c.id, alias="kid")
+        .select(AsyncNode.c.name.as_("node"), child.name.as_("child_name"))
+        .where(AsyncNode.c.parent_id.is_null())
+        .aggregate()
+    )
+    assert [(r["node"], r["child_name"]) for r in rows] == [("root", "a"), ("root", "b")]
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries.py b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries.py
index 75d64c3..7404959 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries.py
@@ -13,6 +13,9 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries_async.py b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries_async.py
index 4c8debd..3104fce 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_parallel_queries_async.py
@@ -13,6 +13,9 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation.py b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation.py
index 8d38cc7..073d398 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation.py
@@ -17,6 +17,9 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation_async.py b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation_async.py
index f676ff6..ee61124 100644
--- a/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation_async.py
+++ b/src/rhosocial/activerecord/testsuite/feature/query/worker/test_transaction_isolation_async.py
@@ -17,6 +17,9 @@
 import pytest
 from rhosocial.activerecord.worker import WorkerPool, TaskContext
 
+pytestmark = pytest.mark.serial
+
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Synchronous Task Functions
diff --git a/src/rhosocial/activerecord/testsuite/utils/__init__.py b/src/rhosocial/activerecord/testsuite/utils/__init__.py
index 0c64fce..a10a129 100644
--- a/src/rhosocial/activerecord/testsuite/utils/__init__.py
+++ b/src/rhosocial/activerecord/testsuite/utils/__init__.py
@@ -48,6 +48,16 @@
     assert_datetime_close,
 )
 
+from .expression import (
+    assert_params_equal,
+    collect_expression_classes,
+    make_instance,
+    register_all,
+    register_special_constructor,
+    roundtrip_expression,
+    sql_consistent,
+)
+
 from .fixture_selector import select_fixture
 
 __all__ = [
@@ -90,4 +100,12 @@
 
     # Fixture selection
     'select_fixture',
+    # Expression round-trip/contract helpers
+    'assert_params_equal',
+    'collect_expression_classes',
+    'make_instance',
+    'register_all',
+    'register_special_constructor',
+    'roundtrip_expression',
+    'sql_consistent',
 ]
diff --git a/src/rhosocial/activerecord/testsuite/utils/expression.py b/src/rhosocial/activerecord/testsuite/utils/expression.py
new file mode 100644
index 0000000..ce4692a
--- /dev/null
+++ b/src/rhosocial/activerecord/testsuite/utils/expression.py
@@ -0,0 +1,398 @@
+# src/rhosocial/activerecord/testsuite/utils/expression.py
+"""
+Expression serialization round-trip / contract test helpers.
+
+This module is the test-only home (NOT shipped as part of the core library's
+runtime API) for constructing expression instances and asserting they survive
+all three serializations. It is shared across the core library tests and all
+backend test suites (MySQL / PostgreSQL / MariaDB / SQL Server / Oracle).
+
+After registering a class into the ExpressionRegistry, expression instances
+must satisfy:
+
+- ``get_params()`` reflects every ``__init__`` parameter
+- A round-trip through the three encoding channels returns a structurally
+  equivalent instance (``get_params()`` equal)
+- When the chosen backend's dialect supports ``to_sql()``, the SQL keyword
+  rendering is identical
+"""
+
+import dataclasses
+import inspect
+import typing
+import warnings
+from typing import Any, Callable, Dict, List, Optional, Type
+
+from rhosocial.activerecord.backend.expression.bases import BaseExpression
+
+
+def _placeholder_for(param: inspect.Parameter, dialect: Any = None):
+    """Return a heuristic construction value for a required parameter."""
+    annotation = param.annotation
+    if annotation is not inspect.Parameter.empty:
+        origin = typing.get_origin(annotation)
+        raw = annotation if origin is None else origin
+        if isinstance(annotation, str):
+            text = annotation
+        else:
+            text = getattr(raw, "__name__", "")
+        if raw in (list, tuple, set):
+            return []
+        if raw is dict:
+            return {}
+        if raw is str or "str" in text:
+            return "x"
+        if raw is int:
+            return 1
+        if raw is float:
+            return 1.0
+        if raw is bool:
+            return True
+        if "Expression" in text or "Subquery" in text:
+            return _literal(dialect)
+
+    name = param.name
+    if name in ("value",):
+        return 1
+    if name in ("values", "columns", "args", "params", "predicates", "expressions"):
+        return []
+    if name in ("left", "right") or "predicate" in name or "condition" in name:
+        return _comparison(dialect)
+    if "expression" in name or name in ("expr", "operand", "subquery", "query"):
+        return _literal(dialect)
+    return "x"
+
+
+def _literal(dialect):
+    from rhosocial.activerecord.backend.expression.core import Literal
+
+    return Literal(dialect, 1)
+
+
+def _comparison(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.predicates import ComparisonPredicate
+
+    return ComparisonPredicate(dialect, "=", Column(dialect, "a"), _literal(dialect))
+
+
+def _try_construct(cls, dialect):
+    """Construct an instance of cls with heuristic arguments; None on failure."""
+    sig = inspect.signature(cls.__init__)
+    # DataType-family declares dialect as an optional trailing keyword.
+    has_dialect = "dialect" in sig.parameters
+    first_param = next((p for p in sig.parameters if p != "self"), None)
+    dialect_as_kw = first_param != "dialect"
+    args = []
+    kwargs = {}
+    skipped_defaulted_positional = False
+    for pname, param in sig.parameters.items():
+        if pname in ("self", "dialect"):
+            continue
+        if param.kind == inspect.Parameter.VAR_KEYWORD:
+            continue
+        if param.kind == inspect.Parameter.VAR_POSITIONAL:
+            if not skipped_defaulted_positional:
+                args.append(_literal(dialect))
+            continue
+        if param.default is not inspect.Parameter.empty:
+            if param.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD:
+                skipped_defaulted_positional = True
+            continue
+        placeholder = _placeholder_for(param, dialect)
+        # placeholder None → default case; real value provided otherwise
+        if param.kind == inspect.Parameter.KEYWORD_ONLY:
+            kwargs[pname] = placeholder
+        else:
+            args.append(placeholder)
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            if not has_dialect:
+                return cls(*args, **kwargs)
+            if dialect_as_kw:
+                return cls(*args, dialect=dialect, **kwargs)
+            return cls(dialect, *args, **kwargs)
+    except Exception:
+        return None
+
+
+def special_constructors():
+    """Explicit constructors for classes whose ``__init__`` rejects heuristic
+    placeholder values. Keyed by a suffix that matches the class's full
+    module.ClassName. Registered by name only — this is a static registry and
+    backends may register their own entries for backend-specific expressions.
+    """
+    return {
+        "advanced_functions.JSONExpression": _json_expr,
+        "query_parts.JoinExpression": _join_expr,
+        "statements.ddl_partition.PartitionClause": _partition_clause,
+        "statements.dml.OnConflictClause": _on_conflict,
+        "statements.dml.UpdateExpression": _update_expr,
+        "statements.ddl_view.CreateViewExpression": _create_view_expr,
+        "datetime.ExtractExpression": _extract_expr,
+        "datetime.DatePartExpression": _datepart_expr,
+        "datetime.DateTruncExpression": _datetrunc_expr,
+        "datetime.IntervalExpression": _interval_expr,
+        "datetime.DateTimeDiffExpression": _datetime_diff_expr,
+        "datetime.DateTimeSubtractExpression": _datetime_sub_expr,
+        "datetime.DateTimeAddExpression": _datetime_add_expr,
+    }
+
+
+def _json_expr(dialect):
+    from rhosocial.activerecord.backend.expression.advanced_functions import JSONExpression
+    from rhosocial.activerecord.backend.expression.core import Literal
+
+    return JSONExpression(dialect, Literal(dialect, '{"a": 1}'), path="$.a")
+
+
+def _table(dialect, name="t"):
+    from rhosocial.activerecord.backend.expression.core import TableExpression
+
+    return TableExpression(dialect, name)
+
+
+def _join_expr(dialect):
+    from rhosocial.activerecord.backend.expression.query_parts import JoinExpression
+    from rhosocial.activerecord.backend.expression.predicates import ComparisonPredicate
+    from rhosocial.activerecord.backend.expression.core import Column
+
+    return JoinExpression(
+        dialect,
+        left_table=_table(dialect, "a"),
+        right_table=_table(dialect, "b"),
+        condition=ComparisonPredicate(dialect, "=", Column(dialect, "a"), Column(dialect, "b")),
+    )
+
+
+def _partition_clause(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.statements.ddl_partition import (
+        PartitionClause,
+        PartitionStrategy,
+    )
+
+    return PartitionClause(dialect, method=PartitionStrategy.RANGE, keys=[Column(dialect, "created_at")])
+
+
+def _on_conflict(dialect):
+    from rhosocial.activerecord.backend.expression.statements.dml import OnConflictClause
+
+    return OnConflictClause(dialect, do_nothing=True, conflict_target=["id"])
+
+
+def _update_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Literal
+    from rhosocial.activerecord.backend.expression.statements.dml import UpdateExpression
+
+    return UpdateExpression(dialect, table=_table(dialect), assignments={"a": Literal(dialect, 1)})
+
+
+def _create_view_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.statements.ddl_view import CreateViewExpression
+    from rhosocial.activerecord.backend.expression.statements.dql import QueryExpression
+
+    query = QueryExpression(dialect, select=[Column(dialect, "id")], from_=_table(dialect, "t"))
+    return CreateViewExpression(dialect, view_name="v", query=query)
+
+
+def _extract_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import ExtractExpression
+
+    return ExtractExpression(dialect, "YEAR", Column(dialect, "created_at"))
+
+
+def _datepart_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import DatePartExpression
+
+    return DatePartExpression(dialect, "YEAR", Column(dialect, "created_at"))
+
+
+def _datetrunc_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import DateTruncExpression
+
+    return DateTruncExpression(dialect, "year", Column(dialect, "created_at"))
+
+
+def _interval_expr(dialect):
+    from rhosocial.activerecord.backend.expression.datetime import IntervalExpression
+
+    return IntervalExpression(dialect, 7, "day")
+
+
+def _datetime_diff_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import DateTimeDiffExpression
+
+    return DateTimeDiffExpression(dialect, "day", Column(dialect, "a"), Column(dialect, "b"))
+
+
+def _datetime_sub_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import (
+        DateTimeSubtractExpression,
+        IntervalExpression,
+    )
+
+    return DateTimeSubtractExpression(dialect, Column(dialect, "a"), IntervalExpression(dialect, 1, "day"))
+
+
+def _datetime_add_expr(dialect):
+    from rhosocial.activerecord.backend.expression.core import Column
+    from rhosocial.activerecord.backend.expression.datetime import (
+        DateTimeAddExpression,
+        IntervalExpression,
+    )
+
+    return DateTimeAddExpression(dialect, Column(dialect, "a"), IntervalExpression(dialect, 1, "day"))
+
+
+_SPECIAL_REGISTRY: Dict[str, Callable] = special_constructors()
+
+
+def register_special_constructor(fqn_suffix: str, factory: Callable) -> None:
+    """Register an explicit constructor for an expression class.
+
+    ``fqn_suffix`` must match the tail of the class's fully qualified name
+    (e.g. ``"json.MySQLJSONObjectExpression"``). ``factory(dialect)`` must
+    return an expression instance.
+    """
+    _SPECIAL_REGISTRY[fqn_suffix] = factory
+
+
+def make_instance(cls: Type[BaseExpression], dialect: Any):
+    """Best-effort construction for an expression class.
+
+    Returns (instance, source) where source is "special", "heuristic", or
+    (None, "failed").
+    """
+    full_name = f"{cls.__module__}.{cls.__name__}"
+    special = next((fn for s, fn in _SPECIAL_REGISTRY.items() if full_name.endswith(s)), None)
+    if special is not None:
+        try:
+            return special(dialect), "special"
+        except Exception:
+            return None, "special-failed"
+    instance = _try_construct(cls, dialect)
+    if instance is None:
+        return None, "failed"
+    return instance, "heuristic"
+
+
+def collect_expression_classes(package_path: str) -> Dict[str, Type[BaseExpression]]:
+    """Walk a package subtree and collect every defined BaseExpression subclass."""
+    import importlib
+    import pkgutil
+
+    pkg = importlib.import_module(package_path)
+    classes: Dict[str, Type[BaseExpression]] = {}
+    for _, modname, _ in pkgutil.walk_packages(pkg.__path__, pkg.__name__ + "."):
+        try:
+            mod = importlib.import_module(modname)
+        except Exception:
+            continue
+        for name in dir(mod):
+            obj = getattr(mod, name, None)
+            if (
+                isinstance(obj, type)
+                and obj.__module__ == modname
+                and issubclass(obj, BaseExpression)
+                and obj is not BaseExpression
+                and not inspect.isabstract(obj)
+            ):
+                classes[f"{modname}.{name}"] = obj
+    return classes
+
+
+def register_all(classes: Dict[str, Type[BaseExpression]]) -> None:
+    from rhosocial.activerecord.backend.expression.serialization import ExpressionRegistry
+
+    for cls in classes.values():
+        ExpressionRegistry.register(cls)
+
+
+def assert_params_equal(a: Any, b: Any, path: str = "params") -> None:
+    """Deep-compare two get_params() outputs, treating nested BaseExpression
+    instances as structurally equal and nested dataclasses as recursively compared.
+    """
+    if a is b:
+        return
+    if isinstance(a, BaseExpression) and isinstance(b, BaseExpression):
+        assert_params_equal(a.get_params(), b.get_params(), path + ".<expr>")
+        return
+    if dataclasses.is_dataclass(a) and not isinstance(a, type):
+        assert dataclasses.is_dataclass(b) and not isinstance(b, type), (
+            f"{path}: expected dataclass instance, got {type(b).__name__}"
+        )
+        for f in dataclasses.fields(a):
+            assert_params_equal(getattr(a, f.name), getattr(b, f.name), f"{path}.{f.name}")
+        return
+    assert type(a) is type(b) or (
+        isinstance(a, (list, tuple, dict)) and isinstance(b, (list, tuple, dict))
+    ), f"{path}: type mismatch {type(a).__name__} vs {type(b).__name__}"
+    if isinstance(a, dict):
+        assert set(a) == set(b), f"{path}: keys differ {set(a) ^ set(b)}"
+        for k in a:
+            assert_params_equal(a[k], b[k], f"{path}.{k}")
+        return
+    if isinstance(a, (list, tuple)):
+        assert len(a) == len(b), f"{path}: length differ"
+        for i, (x, y) in enumerate(zip(a, b)):
+            assert_params_equal(x, y, f"{path}[{i}]")
+        return
+    assert a == b, f"{path}: {a!r} != {b!r}"
+
+
+def roundtrip_expression(fqn: str, instance: BaseExpression, dialect: Any) -> None:
+    """Assert an instance round-trips losslessly through dict / JSON / XML."""
+    original = instance.get_params()
+    for restored in (
+        _rt_dict(instance, dialect),
+        _rt_json(instance, dialect),
+        _rt_xml(instance, dialect),
+    ):
+        assert_params_equal(restored.get_params(), original, fqn)
+
+
+def _rt_dict(instance, dialect):
+    from rhosocial.activerecord.backend.expression.serialization import deserialize, serialize
+
+    return deserialize(serialize(instance), dialect)
+
+
+def _rt_json(instance, dialect):
+    from rhosocial.activerecord.backend.expression.serialization import deserialize_json, serialize_json
+
+    return deserialize_json(serialize_json(instance), dialect)
+
+
+def _rt_xml(instance, dialect):
+    from rhosocial.activerecord.backend.expression.serialization import deserialize_xml, serialize_xml
+
+    return deserialize_xml(serialize_xml(instance), dialect)
+
+
+def sql_consistent(fqn: str, instance: BaseExpression, dialect: Any) -> None:
+    """Assert to_sql is identical after round-trip, when the dialect supports it."""
+    try:
+        expected = instance.to_sql()
+    except Exception:
+        return
+    for restored in (_rt_dict(instance, dialect), _rt_json(instance, dialect), _rt_xml(instance, dialect)):
+        assert restored.to_sql() == expected, fqn
+
+
+__all__ = [
+    "assert_params_equal",
+    "collect_expression_classes",
+    "make_instance",
+    "register_all",
+    "register_special_constructor",
+    "roundtrip_expression",
+    "sql_consistent",
+]
\ No newline at end of file