test(metacog): drop redundant valid_until test + tighten concurrency bound + cover is_watch_active

Closes round-2 review findings q-1, q-2, q-6.

* **q-1:** ``test_valid_until_drops_when_watch_missing`` collapsed to the
  same code path as ``test_valid_until_drops_when_watch_inactive`` after
  the apply-pass switched the predicate from ``get_watch[active]`` to
  ``is_watch_active`` (both stubbed via ``patch_session_storage(active=False)``).
  The "missing" case has no distinguishable branch at the dispatch
  layer, so dropping it removes a tautological duplicate.  The
  missing-row mapping moves to the storage layer (q-2 below) where it
  IS distinguishable.

* **q-2:** ``is_watch_active`` was a new public storage primitive with
  zero direct backend coverage — only via-session-via-stub coverage.
  New ``TestIsWatchActive`` in ``tests/test_watch_storage.py`` covers
  active row → True, inactive row → False, missing row → False.
  Pinned at the storage boundary so future backend changes fail loudly
  there instead of in the dispatch tests.

* **q-6:** Concurrency test had ``n_threads = 2`` alongside two literal
  Thread objects and a tautological ``assert len(threads) == n_threads``.
  Threads are now built from a labels tuple, so ``len(threads)`` drives
  the slack bound; the redundant assertion is gone.

tests/test_watch_dispatch.py

@@ -243,19 +243,6 @@ class TestValidUntil:
         # Predicate ran once with the dispatched watch_id.
         assert is_active_calls == ["watch-1"]
 
-    def test_valid_until_drops_when_watch_missing(self, tmp_db, monkeypatch):
-        session = _make_session_for_dispatch()
-        _runner, dispatch = _register_runner(session)
-
-        # Missing-row case collapses to the same False return shape under
-        # the ``is_watch_active`` API — pinned separately to make the
-        # missing-vs-inactive intent explicit at the call site.
-        patch_session_storage(monkeypatch, active=False)
-
-        dispatch("body", "watch-1")
-        out = session._nudge_queue.drain({"any"})
-        assert out == []
-
     def test_valid_until_drops_when_storage_raises(self, tmp_db, monkeypatch):
         """The closure's broad-except in the predicate translates a
         storage-layer exception to ``False`` so the drain doesn't
@@ -310,18 +297,14 @@ class TestConcurrency:
         # behaviour, not storage round-trips.
         patch_session_storage(monkeypatch, active=True)
 
-        n_threads = 2
         per_thread = 100
+        labels = ("a", "b")
 
         def fire(label: str) -> None:
             for i in range(per_thread):
                 dispatch(f"{label}-{i}", f"watch-{label}")
 
-        threads = [
-            threading.Thread(target=fire, args=("a",), daemon=True),
-            threading.Thread(target=fire, args=("b",), daemon=True),
-        ]
-        assert len(threads) == n_threads
+        threads = [threading.Thread(target=fire, args=(label,), daemon=True) for label in labels]
         for t in threads:
             t.start()
         for t in threads:
@@ -329,16 +312,12 @@ class TestConcurrency:
         for t in threads:
             assert not t.is_alive(), "dispatch thread did not finish in time"
 
-        # Two threads × 100 dispatches → at most 200 entries; the cap
-        # bounds the actual count; no exception was raised across the
-        # 3-acquisition window per dispatch.
+        # The non-atomic count-then-drop window admits at most one "slip"
+        # per concurrent thread above the cap (each thread can observe a
+        # sub-cap count and append before another thread's drop runs).
         depth = len(session._nudge_queue)
-        assert depth <= n_threads * per_thread
-        # The non-atomic count-then-drop window admits at most one
-        # "slip" per concurrent thread above the cap (each thread can
-        # observe a sub-cap count and append before another thread's
-        # drop runs).  Bound the slack to N_THREADS, not 2 * per_thread.
-        assert depth <= _WATCH_QUEUE_SOFT_CAP + n_threads
+        assert depth <= len(threads) * per_thread
+        assert depth <= _WATCH_QUEUE_SOFT_CAP + len(threads)
 
 
 # ---------------------------------------------------------------------------

tests/test_watch_storage.py

@@ -72,6 +72,20 @@ class TestWatchCRUD:
         assert db.delete_watch("nope") is False
 
 
+class TestIsWatchActive:
+    def test_active_row_returns_true(self, db):
+        db.create_watch(**_make_watch_kwargs())
+        assert db.is_watch_active("watch_001") is True
+
+    def test_inactive_row_returns_false(self, db):
+        db.create_watch(**_make_watch_kwargs())
+        db.update_watch("watch_001", active=False)
+        assert db.is_watch_active("watch_001") is False
+
+    def test_missing_row_returns_false(self, db):
+        assert db.is_watch_active("nope") is False
+
+
 class TestWatchListQueries:
     def test_list_for_ws(self, db):
         db.create_watch(**_make_watch_kwargs(watch_id="w1", ws_id="ws-1", name="a"))