Address PR review: testable age flush, thread-safe tests, accurate docstring

- Extract the age-based bundle flush from _periodic_flush into
  _maybe_flush_bundle_by_age(now=...), with an injectable clock reading so
  the 20-minute time trigger can be unit tested without sleeping.
- Add tests covering the age-triggered merge/upload, the below-threshold
  no-op, and the empty-buffer no-op.
- Run bundle tests through a SilentWriteManager subclass that neuters the
  adaptive-sizing and periodic-flush background threads, so they don't
  linger across runs or fire timers mid-assertion.
- Fix the _add_to_bundle docstring to note a size-triggered flush/merge/
  upload can happen from that method.

Author: claude

src/tracer/WriterManager.py

@@ -290,18 +290,34 @@ def _periodic_flush(self):
                 except Exception as e:
                     logger("error", f"Error in periodic flush: {e}")
 
-            # Age-based bundle flush: merge and upload any locally buffered
-            # files once the oldest has waited bundle_max_interval (20 min),
-            # even if the 100 MB size threshold was never reached.
-            if self.automatic_upload:
-                with self._bundle_lock:
-                    has_pending = bool(self._pending_bundle)
-                    bundle_age = time.monotonic() - self._bundle_window_start
-                if has_pending and bundle_age >= self.bundle_max_interval:
-                    try:
-                        self._flush_bundle()
-                    except Exception as e:
-                        logger("error", f"Error in periodic bundle flush: {e}")
+            try:
+                self._maybe_flush_bundle_by_age()
+            except Exception as e:
+                logger("error", f"Error in periodic bundle flush: {e}")
+
+    def _maybe_flush_bundle_by_age(self, now: float | None = None) -> bool:
+        """Merge and upload buffered files if the oldest has waited too long.
+
+        Triggers a bundle flush once the buffering window has been open for at
+        least ``bundle_max_interval`` (20 min), even if the ``bundle_max_bytes``
+        size threshold was never reached. ``now`` is an injectable
+        ``time.monotonic()`` reading so the age trigger can be unit tested
+        without sleeping for the full interval.
+
+        Returns:
+            bool: True if a flush was triggered, False otherwise.
+        """
+        if not self.automatic_upload:
+            return False
+        if now is None:
+            now = time.monotonic()
+        with self._bundle_lock:
+            has_pending = bool(self._pending_bundle)
+            bundle_age = now - self._bundle_window_start
+        if has_pending and bundle_age >= self.bundle_max_interval:
+            self._flush_bundle()
+            return True
+        return False
 
     def _reset_flush_timer(self):
         """Reset the periodic flush timer (called after manual flushes)."""
@@ -963,9 +979,10 @@ def _add_to_bundle(self, file_path: str):
         """Buffer a compressed file locally for a later merged upload.
 
         Files accumulate on local disk until the buffered size reaches
-        ``bundle_max_bytes``; the age-based flush (``bundle_max_interval``) is
-        handled separately by the periodic flush thread. Nothing is uploaded
-        here, only queued for the eventual merge.
+        ``bundle_max_bytes``, at which point this method triggers
+        ``_flush_bundle`` to merge them into a single tar and queue it for
+        upload. The age-based flush (``bundle_max_interval``) is handled
+        separately by the periodic flush thread.
         """
         try:
             file_size = os.path.getsize(file_path)

tests/test_writer_bundle.py

@@ -43,18 +43,31 @@ def append_object(self, file_path):
         self.uploaded.append(file_path)
 
 
+class SilentWriteManager(WriteManager):
+    """WriteManager with its background threads neutered for tests.
+
+    ``__init__`` still spins up the adaptive-sizing and periodic-flush threads,
+    but overriding their targets with no-ops makes them exit immediately so
+    they neither linger across test runs nor fire timers during assertions.
+    """
+
+    def _adaptive_sizing(self):
+        return
+
+    def _periodic_flush(self):
+        return
+
+
 class BundleBufferingTests(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.mkdtemp()
         self.output_dir = os.path.join(self.tmp, "trace")
         self.upload = FakeUploadManager()
-        self.wm = WriteManager(
+        self.wm = SilentWriteManager(
             output_dir=self.output_dir,
             upload_manager=self.upload,
             automatic_upload=True,
         )
-        # Stop the background threads so they don't interfere with assertions.
-        self.wm._periodic_flush_active = False
 
     def tearDown(self):
         import shutil
@@ -97,6 +110,45 @@ def test_size_threshold_triggers_merge_and_upload(self):
         with tarfile.open(bundle_path) as tar:
             self.assertEqual(len(tar.getmembers()), 3)
 
+    def test_age_threshold_triggers_merge_and_upload(self):
+        # Keep the size threshold high so only the age trigger can fire.
+        self.wm.bundle_max_bytes = 10_000
+        self.wm.bundle_max_interval = 1200  # 20 minutes
+        f = self._make_file("a.csv.gz", 100)
+        self.wm._add_to_bundle(f)
+        self.assertEqual(self.upload.uploaded, [])  # not old enough yet
+
+        # Inject a monotonic reading just past the interval.
+        now = self.wm._bundle_window_start + 1201
+        flushed = self.wm._maybe_flush_bundle_by_age(now=now)
+
+        self.assertTrue(flushed)
+        self.assertEqual(len(self.upload.uploaded), 1)
+        bundle_path = self.upload.uploaded[0]
+        self.assertTrue(bundle_path.endswith(".tar"))
+        with tarfile.open(bundle_path) as tar:
+            self.assertEqual(len(tar.getmembers()), 1)
+        self.assertEqual(self.wm._pending_bundle, [])
+        self.assertEqual(self.wm._pending_bundle_bytes, 0)
+        self.assertFalse(os.path.exists(f))
+
+    def test_age_below_threshold_does_not_flush(self):
+        self.wm.bundle_max_bytes = 10_000
+        self.wm.bundle_max_interval = 1200
+        f = self._make_file("a.csv.gz", 100)
+        self.wm._add_to_bundle(f)
+
+        now = self.wm._bundle_window_start + 5  # well under the interval
+        self.assertFalse(self.wm._maybe_flush_bundle_by_age(now=now))
+        self.assertEqual(self.upload.uploaded, [])
+        self.assertEqual(self.wm._pending_bundle, [f])
+
+    def test_age_flush_is_noop_when_empty(self):
+        # No buffered files: even a huge age must not flush or upload.
+        now = self.wm._bundle_window_start + 999_999
+        self.assertFalse(self.wm._maybe_flush_bundle_by_age(now=now))
+        self.assertEqual(self.upload.uploaded, [])
+
     def test_window_start_resets_on_new_buffer(self):
         self.wm.bundle_max_bytes = 10_000
         old = time.monotonic() - 9999