diff --git a/CHANGELOG.md b/CHANGELOG.md
index b969539..0259672 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `django_ox.testing.assert_task_count()`, a read-only assertion that reads
+  the `OxTask` table directly, matching by backend alias, queue, task path,
+  and `args`/`kwargs` compared the same way `enqueue()` stores them. A
+  rolled-back enqueue is not counted. `docs/patterns.md` also documents an
+  isolated-queue `run_tasks()` pytest fixture recipe, with a repository
+  test, not a shipped fixture API.
 - `ox_retry` and `ox_discard` management commands for scriptable recovery
   of tasks. Both commands take one or more task UUIDs, deduplicate them, and
   delegate to `django_ox.actions.retry_many` or `discard_many`. Both support
diff --git a/docs/llms-full.txt b/docs/llms-full.txt
index 7749461..d472634 100644
--- a/docs/llms-full.txt
+++ b/docs/llms-full.txt
@@ -4654,6 +4654,86 @@ Use a real worker to test timeout enforcement and worker infrastructure. `run_ta
 
 Timeouts are inert. `deadline()` and `remaining()` return `None`. A configured timeout produces `run_tasks_timeout_inert` once per task path per call. A hanging body or callback hangs the test.
 
+### Assert on the queue directly
+
+`django_ox.testing.assert_task_count()` reads the `OxTask` table itself, so
+it works whether a task is still queued or has already run and left its
+row behind. It matches by backend alias, queue and task path, with `args`
+and `kwargs` compared the same way `enqueue()` stores them:
+
+```python
+from django_ox.testing import assert_task_count
+
+send_welcome_email.enqueue(user.pk)
+assert_task_count(
+    path="myapp.tasks.send_welcome_email", queue="default", args=[user.pk], count=1
+)
+```
+
+A rolled-back enqueue is not counted, since the row behind it never
+existed. `args` and `kwargs` are matched after the same JSON round trip
+`enqueue()` applies before storing them, so a tuple in `args` matches a
+stored list, and a raw `datetime`, `UUID` or `Decimal` is rejected with
+`TypeError`: supply that value's own stored JSON representation instead.
+
+### An isolated queue for one test
+
+`run_tasks()` claims whatever its backend's worker would, which includes
+rows other tests, fixtures or `setUpTestData` left behind on a shared
+queue. To run only a task under test, reserve a queue nothing else writes
+to, and check it is still empty before draining it. This is a documented
+recipe, not a shipped fixture or pytest plugin:
+
+```python
+import pytest
+from django_ox.models import OxTask
+from django_ox.testing import run_tasks
+
+# Must appear in this backend's own QUEUES list (a sibling of OPTIONS,
+# not inside it): TASKS["default"]["QUEUES"].
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    # Checked once, here, before the test body enqueues anything: a row
+    # from setUpTestData or an earlier fixture is caught before it is ever
+    # claimed. The returned `run` does not repeat the check, since by then
+    # the only row it is meant to find is the one the test just enqueued.
+    if OxTask.objects.filter(
+        backend_name="default", queue_name=ISOLATED_QUEUE
+    ).exists():
+        raise ValueError(
+            f"queue {ISOLATED_QUEUE!r} already holds rows; reserve a queue "
+            "no other test or fixture writes to"
+        )
+
+    def run(**kwargs):
+        return run_tasks(queues=[ISOLATED_QUEUE], **kwargs)
+
+    return run
+```
+
+```python
+@task(queue_name=ISOLATED_QUEUE)
+def convert_video(path): ...
+
+
+def test_convert_video(isolated_run_tasks):
+    convert_video.enqueue("clip.mp4")
+    [result] = isolated_run_tasks()
+    assert result.return_value is not None
+```
+
+Fixture setup runs before the test body, so the emptiness check sees only
+what was there beforehand, never the task the test is about to enqueue.
+A queue that already holds work, including a row `setUpTestData` left
+there, fails fast with `ValueError` rather than draining someone else's
+task. Passing `queues=` to the wrapped `run_tasks()` call, not a model
+filter, is what keeps the drain from touching unrelated rows:
+`tests/test_testing_assertions.py` tests this recipe's isolation and
+failure behaviour directly.
+
 ## Optional coordination features
 
 [Unique tasks](features/unique-tasks.md), [batches](features/batches.md),
@@ -8524,11 +8604,15 @@ export names this page does not list; those names are not public.
   is always `null`. On `worker_claim_recovery_failed` and
   `worker_claim_recovery_expired`, it is always `"expired"`.
 - **The testing helpers** `django_ox.testing.ImmediateBackend`,
-  `django_ox.testing.DummyBackend` and `django_ox.testing.run_tasks`.
-  The backends accept policy declarations but do not enforce retries,
-  backoff or timeouts. `run_tasks()` drains due queued tasks through the
-  configured worker class, including retries and backoff. It does not
-  enforce timeouts. `run_tasks()` is public and provisional.
+  `django_ox.testing.DummyBackend`, `django_ox.testing.run_tasks` and
+  `django_ox.testing.assert_task_count`. The backends accept policy
+  declarations but do not enforce retries, backoff or timeouts.
+  `run_tasks()` drains due queued tasks through the configured worker
+  class, including retries and backoff. It does not enforce timeouts.
+  `run_tasks()` is public and provisional. `assert_task_count()` is
+  public and stable: its keyword-only signature
+  (`path`, `queue`, `count`, `backend`, `args`, `kwargs`) does not carry
+  `run_tasks()`'s provisional status.
 - **The database schema** of `OxTask` and `OxScheduleTick`, evolved only
   through shipped migrations.
 - **`django_ox.__version__`.**
@@ -9384,6 +9468,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `django_ox.testing.assert_task_count()`, a read-only assertion that reads
+  the `OxTask` table directly, matching by backend alias, queue, task path,
+  and `args`/`kwargs` compared the same way `enqueue()` stores them. A
+  rolled-back enqueue is not counted. `docs/patterns.md` also documents an
+  isolated-queue `run_tasks()` pytest fixture recipe, with a repository
+  test, not a shipped fixture API.
 - `ox_retry` and `ox_discard` management commands for scriptable recovery
   of tasks. Both commands take one or more task UUIDs, deduplicate them, and
   delegate to `django_ox.actions.retry_many` or `discard_many`. Both support
diff --git a/docs/patterns.md b/docs/patterns.md
index e83599f..a0aa561 100644
--- a/docs/patterns.md
+++ b/docs/patterns.md
@@ -447,6 +447,86 @@ Use a real worker to test timeout enforcement and worker infrastructure. `run_ta
 
 Timeouts are inert. `deadline()` and `remaining()` return `None`. A configured timeout produces `run_tasks_timeout_inert` once per task path per call. A hanging body or callback hangs the test.
 
+### Assert on the queue directly
+
+`django_ox.testing.assert_task_count()` reads the `OxTask` table itself, so
+it works whether a task is still queued or has already run and left its
+row behind. It matches by backend alias, queue and task path, with `args`
+and `kwargs` compared the same way `enqueue()` stores them:
+
+```python
+from django_ox.testing import assert_task_count
+
+send_welcome_email.enqueue(user.pk)
+assert_task_count(
+    path="myapp.tasks.send_welcome_email", queue="default", args=[user.pk], count=1
+)
+```
+
+A rolled-back enqueue is not counted, since the row behind it never
+existed. `args` and `kwargs` are matched after the same JSON round trip
+`enqueue()` applies before storing them, so a tuple in `args` matches a
+stored list, and a raw `datetime`, `UUID` or `Decimal` is rejected with
+`TypeError`: supply that value's own stored JSON representation instead.
+
+### An isolated queue for one test
+
+`run_tasks()` claims whatever its backend's worker would, which includes
+rows other tests, fixtures or `setUpTestData` left behind on a shared
+queue. To run only a task under test, reserve a queue nothing else writes
+to, and check it is still empty before draining it. This is a documented
+recipe, not a shipped fixture or pytest plugin:
+
+```python
+import pytest
+from django_ox.models import OxTask
+from django_ox.testing import run_tasks
+
+# Must appear in this backend's own QUEUES list (a sibling of OPTIONS,
+# not inside it): TASKS["default"]["QUEUES"].
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    # Checked once, here, before the test body enqueues anything: a row
+    # from setUpTestData or an earlier fixture is caught before it is ever
+    # claimed. The returned `run` does not repeat the check, since by then
+    # the only row it is meant to find is the one the test just enqueued.
+    if OxTask.objects.filter(
+        backend_name="default", queue_name=ISOLATED_QUEUE
+    ).exists():
+        raise ValueError(
+            f"queue {ISOLATED_QUEUE!r} already holds rows; reserve a queue "
+            "no other test or fixture writes to"
+        )
+
+    def run(**kwargs):
+        return run_tasks(queues=[ISOLATED_QUEUE], **kwargs)
+
+    return run
+```
+
+```python
+@task(queue_name=ISOLATED_QUEUE)
+def convert_video(path): ...
+
+
+def test_convert_video(isolated_run_tasks):
+    convert_video.enqueue("clip.mp4")
+    [result] = isolated_run_tasks()
+    assert result.return_value is not None
+```
+
+Fixture setup runs before the test body, so the emptiness check sees only
+what was there beforehand, never the task the test is about to enqueue.
+A queue that already holds work, including a row `setUpTestData` left
+there, fails fast with `ValueError` rather than draining someone else's
+task. Passing `queues=` to the wrapped `run_tasks()` call, not a model
+filter, is what keeps the drain from touching unrelated rows:
+`tests/test_testing_assertions.py` tests this recipe's isolation and
+failure behaviour directly.
+
 ## Optional coordination features
 
 [Unique tasks](features/unique-tasks.md), [batches](features/batches.md),
diff --git a/docs/stability.md b/docs/stability.md
index 2677677..78ec63a 100644
--- a/docs/stability.md
+++ b/docs/stability.md
@@ -112,11 +112,15 @@ export names this page does not list; those names are not public.
   is always `null`. On `worker_claim_recovery_failed` and
   `worker_claim_recovery_expired`, it is always `"expired"`.
 - **The testing helpers** `django_ox.testing.ImmediateBackend`,
-  `django_ox.testing.DummyBackend` and `django_ox.testing.run_tasks`.
-  The backends accept policy declarations but do not enforce retries,
-  backoff or timeouts. `run_tasks()` drains due queued tasks through the
-  configured worker class, including retries and backoff. It does not
-  enforce timeouts. `run_tasks()` is public and provisional.
+  `django_ox.testing.DummyBackend`, `django_ox.testing.run_tasks` and
+  `django_ox.testing.assert_task_count`. The backends accept policy
+  declarations but do not enforce retries, backoff or timeouts.
+  `run_tasks()` drains due queued tasks through the configured worker
+  class, including retries and backoff. It does not enforce timeouts.
+  `run_tasks()` is public and provisional. `assert_task_count()` is
+  public and stable: its keyword-only signature
+  (`path`, `queue`, `count`, `backend`, `args`, `kwargs`) does not carry
+  `run_tasks()`'s provisional status.
 - **The database schema** of `OxTask` and `OxScheduleTick`, evolved only
   through shipped migrations.
 - **`django_ox.__version__`.**
diff --git a/src/django_ox/testing.py b/src/django_ox/testing.py
index fc42485..40ef58c 100644
--- a/src/django_ox/testing.py
+++ b/src/django_ox/testing.py
@@ -25,11 +25,17 @@
 enqueued the way a worker runs it, on the test's own thread and connection,
 so it works inside a TestCase's transaction; its docstring says what it
 does and does not reproduce.
+
+assert_task_count() is a read-only assertion against the OxTask table
+itself, for either kind of test: whether a task is still queued, or ran and
+left its row behind.
 """
 
 from __future__ import annotations
 
+import json
 import logging
+from collections.abc import Mapping, Sequence
 from threading import Lock
 from typing import Any
 
@@ -38,7 +44,7 @@
 from .compat import ImmediateBackend as _StockImmediateBackend
 from .tasks import PolicyTask, task_policy, validate_policy
 
-__all__ = ["DummyBackend", "ImmediateBackend", "run_tasks"]
+__all__ = ["DummyBackend", "ImmediateBackend", "assert_task_count", "run_tasks"]
 
 logger = logging.getLogger("django_ox")
 
@@ -232,3 +238,80 @@ def run_tasks(
         max_tasks=max_tasks,
         raise_failures=raise_failures,
     )
+
+
+def _json_round_trip(label: str, value: Any) -> Any:
+    try:
+        return json.loads(json.dumps(value))
+    except TypeError as exc:
+        raise TypeError(
+            f"assert_task_count() {label} must be values a standard JSON round "
+            f"trip preserves ({exc}); supply the task's stored JSON "
+            f"representation instead."
+        ) from exc
+
+
+def assert_task_count(
+    *,
+    path: str,
+    queue: str,
+    count: int,
+    backend: str = DEFAULT_TASK_BACKEND_ALIAS,
+    args: Sequence[Any] = (),
+    kwargs: Mapping[str, Any] | None = None,
+) -> None:
+    """
+    Assert that exactly `count` OxTask rows in `backend`'s `queue` have
+    `path` as their stored task path and match `args` and `kwargs`.
+
+    This reads database rows, not a separate enqueue record: a task whose
+    enqueue was rolled back is not counted, and the count includes a row
+    that has since run, retried or finished, since its path, args and
+    kwargs are unchanged by running. Narrow with `run_tasks(queues=...)` or
+    a status check first if only queued work should count.
+
+    `args` and `kwargs` are compared to the stored values after a JSON
+    round trip, the same lossy conversion `enqueue()` itself applies before
+    storing them: a tuple in `args` matches a stored list, but a type a
+    standard `json.dumps()` call cannot handle, such as a raw `datetime`,
+    `UUID` or `Decimal`, is rejected with `TypeError` rather than silently
+    miscompared. Pass that value's own stored JSON representation (a
+    string, an int or a plain list or dict) instead.
+
+    Raises `AssertionError`, naming `backend`, `queue`, `path`, the
+    normalised `args` and `kwargs`, and the expected and observed counts,
+    when the observed count differs.
+    """
+    if not isinstance(path, str):
+        raise TypeError(f"assert_task_count() path must be a string, got {path!r}.")
+    if not isinstance(queue, str):
+        raise TypeError(f"assert_task_count() queue must be a string, got {queue!r}.")
+    if not isinstance(backend, str):
+        raise TypeError(
+            f"assert_task_count() backend must be a string, got {backend!r}."
+        )
+    if isinstance(count, bool) or not isinstance(count, int):
+        raise TypeError(
+            f"assert_task_count() count must be a whole number, got {count!r}."
+        )
+    if count < 0:
+        raise ValueError(f"assert_task_count() count must be 0 or more, got {count!r}.")
+
+    want_args = _json_round_trip("args", list(args))
+    want_kwargs = _json_round_trip("kwargs", dict(kwargs) if kwargs else {})
+
+    # Imported here, not at the top: this module is named in TASKS settings,
+    # and the worker imports the models.
+    from .models import OxTask
+
+    rows = OxTask.objects.filter(backend_name=backend, queue_name=queue, task_path=path)
+    observed = sum(
+        1 for row in rows if row.args == want_args and row.kwargs == want_kwargs
+    )
+    if observed != count:
+        raise AssertionError(
+            f"expected {count} task(s) matching backend={backend!r} "
+            f"queue={queue!r} path={path!r} args={want_args!r} "
+            f"kwargs={want_kwargs!r}, found {observed} (of {rows.count()} row(s) "
+            f"stored for that path in that backend and queue)"
+        )
diff --git a/tests/settings.py b/tests/settings.py
index 5d0ca3b..55ff0a0 100644
--- a/tests/settings.py
+++ b/tests/settings.py
@@ -72,7 +72,9 @@
 TASKS: dict[str, dict[str, Any]] = {
     "default": {
         "BACKEND": "django_ox.backend.OxBackend",
-        "QUEUES": ["default", "emails"],
+        # "isolated-recipe" is reserved for the docs/patterns.md isolated-queue
+        # fixture recipe and its test; nothing else enqueues onto it.
+        "QUEUES": ["default", "emails", "isolated-recipe"],
         "OPTIONS": {"MAX_ATTEMPTS": 3},
     }
 }
diff --git a/tests/test_testing_assertions.py b/tests/test_testing_assertions.py
new file mode 100644
index 0000000..7cff76e
--- /dev/null
+++ b/tests/test_testing_assertions.py
@@ -0,0 +1,186 @@
+"""
+django_ox.testing.assert_task_count(): reading OxTask rows directly, with
+JSON-normalised matching on args and kwargs.
+"""
+
+import pytest
+from django.db import transaction
+from django.test import TestCase, TransactionTestCase
+
+from django_ox.models import OxTask
+from django_ox.testing import assert_task_count, run_tasks
+
+from . import testing_assertions_tasks as t
+
+pytestmark = pytest.mark.django_db
+
+PATH = "tests.testing_assertions_tasks.note"
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    """docs/patterns.md's isolated-queue recipe, reproduced to test it."""
+    if OxTask.objects.filter(
+        backend_name="default", queue_name=ISOLATED_QUEUE
+    ).exists():
+        raise ValueError(
+            f"queue {ISOLATED_QUEUE!r} already holds rows; reserve a queue "
+            "no other test or fixture writes to"
+        )
+
+    def run(**kwargs):
+        return run_tasks(queues=[ISOLATED_QUEUE], **kwargs)
+
+    return run
+
+
+class TestAssertTaskCount(TestCase):
+    def test_matches_an_enqueued_task_by_path_args_and_kwargs(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+    def test_zero_when_nothing_matches(self):
+        assert_task_count(path=PATH, queue="default", count=0)
+
+    def test_tuple_args_match_the_stored_list(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=("hello",), count=1)
+
+    def test_kwargs_are_matched_too(self):
+        t.greet.enqueue(name="Ada", loud=True)
+        assert_task_count(
+            path="tests.testing_assertions_tasks.greet",
+            queue="default",
+            kwargs={"name": "Ada", "loud": True},
+            count=1,
+        )
+
+    def test_wrong_args_do_not_match(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["goodbye"], count=0)
+
+    def test_count_mismatch_reports_expected_and_observed(self):
+        t.note.enqueue("hello")
+        t.note.enqueue("hello")
+        with pytest.raises(AssertionError) as excinfo:
+            assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+        message = str(excinfo.value)
+        assert "expected 1" in message
+        assert "found 2" in message
+        assert PATH in message
+        assert "default" in message
+
+    def test_a_rolled_back_enqueue_is_not_counted(self):
+        with pytest.raises(ValueError, match="roll back"), transaction.atomic():
+            t.note.enqueue("hello")
+            raise ValueError("roll back")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+    def test_scoped_to_its_own_queue(self):
+        t.note.using(queue_name="emails").enqueue("hello")
+        assert_task_count(path=PATH, queue="emails", args=["hello"], count=1)
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+    def test_scoped_to_its_own_backend(self):
+        t.note.enqueue("hello")
+        assert_task_count(
+            path=PATH, queue="default", args=["hello"], backend="other", count=0
+        )
+
+    def test_a_finished_rows_path_args_and_kwargs_still_count(self):
+        from django_ox.testing import run_tasks
+
+        t.note.enqueue("hello")
+        run_tasks()
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+
+class TestAssertTaskCountUnderATransactionTestCase(TransactionTestCase):
+    """
+    TestCase above wraps every test in a transaction it rolls back; a plain
+    enqueue() there never truly commits. TransactionTestCase wraps nothing,
+    so the same assertion has to read a row that is actually committed, and
+    still has to exclude one that a nested atomic() rolled back.
+    """
+
+    def test_a_committed_enqueue_is_counted(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+    def test_a_rolled_back_nested_atomic_is_not_counted(self):
+        with pytest.raises(ValueError, match="roll back"), transaction.atomic():
+            t.note.enqueue("hello")
+            raise ValueError("roll back")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+
+class TestAssertTaskCountArguments:
+    def test_rejects_a_non_string_path(self):
+        with pytest.raises(TypeError, match="path"):
+            assert_task_count(path=123, queue="default", count=0)
+
+    def test_rejects_a_non_string_queue(self):
+        with pytest.raises(TypeError, match="queue"):
+            assert_task_count(path=PATH, queue=123, count=0)
+
+    def test_rejects_a_non_integer_count(self):
+        with pytest.raises(TypeError, match="count"):
+            assert_task_count(path=PATH, queue="default", count="1")
+
+    def test_rejects_a_negative_count(self):
+        with pytest.raises(ValueError, match="count"):
+            assert_task_count(path=PATH, queue="default", count=-1)
+
+    def test_rejects_a_datetime_argument(self):
+        from datetime import datetime
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", args=[datetime.now()], count=0
+            )
+
+    def test_rejects_a_uuid_keyword_argument(self):
+        import uuid
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", kwargs={"id": uuid.uuid4()}, count=0
+            )
+
+    def test_rejects_a_decimal_argument(self):
+        from decimal import Decimal
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", args=[Decimal("1.5")], count=0
+            )
+
+
+def test_isolated_run_tasks_runs_only_the_reserved_queue(isolated_run_tasks):
+    """docs/patterns.md's "An isolated queue for one test" recipe."""
+    t.isolated_task.enqueue("clip.mp4")
+    t.note.enqueue("unrelated")
+    [result] = isolated_run_tasks()
+    assert result.return_value == "clip.mp4"
+    assert_task_count(
+        path="tests.testing_assertions_tasks.note",
+        queue="default",
+        args=["unrelated"],
+        count=1,
+    )
+
+
+def test_isolated_run_tasks_refuses_a_queue_already_holding_a_row(request):
+    # Simulates what setUpTestData or an earlier fixture would leave behind:
+    # a row in the reserved queue before the recipe's own check ever runs.
+    t.isolated_task.enqueue("leftover")
+    with pytest.raises(ValueError, match="already holds"):
+        request.getfixturevalue("isolated_run_tasks")
+    # Refused before any claim: the leftover row is untouched.
+    assert_task_count(
+        path="tests.testing_assertions_tasks.isolated_task",
+        queue=ISOLATED_QUEUE,
+        args=["leftover"],
+        count=1,
+    )
diff --git a/tests/testing_assertions_tasks.py b/tests/testing_assertions_tasks.py
new file mode 100644
index 0000000..777892d
--- /dev/null
+++ b/tests/testing_assertions_tasks.py
@@ -0,0 +1,18 @@
+"""Module-level tasks for tests/test_testing_assertions.py."""
+
+from django_ox.compat import task
+
+
+@task
+def note(label):
+    pass
+
+
+@task
+def greet(name, *, loud=False):
+    pass
+
+
+@task(queue_name="isolated-recipe")
+def isolated_task(label):
+    return label