Add Daq(scan_monitoring=) to opt out of the namespace-wide recording

start_scan_monitoring/end_scan_monitoring were unconditionally wired
into callbacks_start_scan/_end_scan with no way to turn them off short
of editing the callback lists directly. Needed now because
RecordingSession.start() attaches a CA monitor to every monitorable
channel in a single-threaded, unthreaded loop (namespace_store.py:1108),
and start_recording()'s HTTP call only gets the general
status_server_timeout (10s default) rather than the much larger
status_server_snapshot_timeout (180s) the analogous whole-namespace
snapshot/capture call gets -- reliably too slow at bernina's namespace
size, producing a "start monitoring failed ... falling back to the
local namespace mechanism" warning on every scan, easily mistaken for
the whole status-server integration being broken (it isn't -- that's
an unrelated, independently-succeeding call).

scan_monitoring defaults to True (unchanged behavior); set
bernina.daq.scan_monitoring = False on an already-running session for
an immediate, no-restart opt-out, or pass scan_monitoring=False at
construction. Checked first in both methods, before touching
status_client at all, so disabling it can't itself produce a warning.

Proper fix (making /recording/start async like /status/capture already
is) tracked separately, not done here.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

eco/acquisition/daq_client.py

@@ -148,6 +148,7 @@ class Daq(Assembly):
         status_server_stale_action="ask",
         status_server_stale_timeout=20.0,
         status_server_async=True,
+        scan_monitoring=True,
     ):
         super().__init__(name=name)
         self.channels = {}
@@ -286,6 +287,22 @@ class Daq(Assembly):
         # to the run in the background, instead of making the scan wait for
         # all three. See Daq.write_status / POST /status/capture.
         self.status_server_async = status_server_async
+        # Whether callbacks_start_scan/_end_scan attempt the namespace-wide
+        # recording at all (start_scan_monitoring/end_scan_monitoring) -
+        # live-toggleable (bernina.daq.scan_monitoring = False), not just a
+        # constructor default, since this is exactly the kind of thing you
+        # want to switch off mid-session without restarting the namespace.
+        # Currently a plain, synchronous, per-channel attach loop server-side
+        # (RecordingSession.start(), no thread pool) against only a 10 s
+        # client timeout (status_server_timeout, not the more generous
+        # status_server_snapshot_timeout that whole-namespace snapshot/
+        # capture calls get) - reliably too slow at bernina's namespace size,
+        # which is why this defaults to True but is meant to be turned off
+        # until that's fixed properly (make /recording/start async, like
+        # /status/capture already is) rather than spamming "start monitoring
+        # failed ... falling back to the local namespace mechanism" every
+        # single scan.
+        self.scan_monitoring = scan_monitoring
         if not rate_multiplicator == "auto":
             print(
                 "warning: rate multiplicator automatically determined from event_master!"
@@ -2287,6 +2304,8 @@ class Daq(Assembly):
         and `append_start_status_to_scan` (shares its per-scan
         server-in-use decision, cached on the scan).
         """
+        if not self.scan_monitoring:
+            return None
         if self.status_client is None:
             return None
         if not self._status_server_ok_for_this_scan(scan):
@@ -2343,6 +2362,8 @@ class Daq(Assembly):
         (which needs the "monitors" scan parameter already set to write it
         into scan_info_rel.json).
         """
+        if not self.scan_monitoring:
+            return None
         if self.status_client is None:
             return None
         recording_id = scan.counter_scratch(self.name).get(

tests/test_daq_status_server.py

@@ -88,6 +88,7 @@ def _daq(status_server=None, namespace=None, strict=False):
     # the pre-existing tests below cover the synchronous path; the async one
     # has its own set further down
     daq.status_server_async = False
+    daq.scan_monitoring = True
     daq.aux_calls = []
     daq.append_aux = lambda *a, **kw: daq.aux_calls.append((a, kw))
     return daq
@@ -640,6 +641,34 @@ def test_start_scan_monitoring_warns_about_failed_required_channels(capsys):
     assert "REQUIRED" in out and "daq.something" in out
 
 
+def test_start_scan_monitoring_is_a_noop_when_disabled():
+    """scan_monitoring=False (bernina.daq.scan_monitoring = False) -- the
+    opt-out for RecordingSession.start()'s current per-channel, unthreaded
+    attach loop being reliably slower than status_server_timeout at
+    bernina's namespace size. Checked before even touching status_client,
+    so this also can't produce a "start monitoring failed" warning."""
+    client = HealthClient()
+    daq = _daq_fresh(client)
+    daq.scan_monitoring = False
+    scan = FakeScan(runno=42)
+    assert daq.start_scan_monitoring(scan) is None
+    assert client.recording_starts == []
+    assert "monitoring_recording_id" not in scan.counter_scratch("daq")
+
+
+def test_end_scan_monitoring_is_a_noop_when_disabled_even_with_a_recording_id():
+    """Belt-and-suspenders: even if monitoring_recording_id somehow got set
+    (e.g. scan_monitoring flipped off mid-scan), end_scan_monitoring must
+    not try to capture it."""
+    client = HealthClient()
+    daq = _daq_fresh(client)
+    scan = FakeScan(runno=42)
+    scan.counter_scratch("daq")["monitoring_recording_id"] = "p12345_run0042"
+    daq.scan_monitoring = False
+    assert daq.end_scan_monitoring(scan) is None
+    assert client.recording_captures == []
+
+
 def test_start_scan_monitoring_is_a_noop_without_a_status_server():
     daq = _daq_fresh(None)
     daq._status_server = None