fix: eliminate three thread-safety bugs in telemetry delta reporting

src/tracer/IOTracer.py

@@ -168,7 +168,7 @@ def __init__(
                 self.automatic_upload = False
 
 
-        self.writer             = WriteManager(output_dir, self.upload_manager, automatic_upload)
+        self.writer             = WriteManager(output_dir, self.upload_manager, self.automatic_upload)
         if self.automatic_upload:
             self.upload_manager.on_upload_success = self._post_telemetry_delta
         self.fs_snapper         = FilesystemSnapper(self.writer, anonymous)
@@ -294,20 +294,28 @@ def _activity_detector(self) -> None:
             with self._counter_lock:
                 count = self._disk_event_counter
                 self._disk_event_counter = 0
-            if count > _ACTIVITY_THRESHOLD:
-                self.active_duration_s += 1
+                if count > _ACTIVITY_THRESHOLD:
+                    self.active_duration_s += 1
 
     def _post_telemetry_delta(self) -> None:
+        with self._counter_lock:
+            current_active_s = self.active_duration_s
+
         with self._telemetry_lock:
-            duration_delta = self.active_duration_s - self._last_posted_active_s
-            self._last_posted_active_s = self.active_duration_s
+            duration_delta = current_active_s - self._last_posted_active_s
+
+            current_rows = self.writer.get_rows_written_snapshot()
 
-            current_rows = dict(self.writer.rows_written)
             events_delta = {
                 k.lower().replace(" ", "_"): current_rows.get(k, 0) - self._last_posted_rows.get(k, 0)
                 for k in current_rows
                 if current_rows.get(k, 0) > self._last_posted_rows.get(k, 0)
             }
+
+            if duration_delta == 0 and not events_delta:
+                return
+
+            self._last_posted_active_s = current_active_s
             self._last_posted_rows = current_rows
 
         self.upload_manager.post_telemetry(
@@ -1351,6 +1359,7 @@ def _write_manifest(self, started_at, stopped_at=None):
         """
         manifest = schema.schema_for_manifest()
         duration = (stopped_at - started_at).total_seconds() if stopped_at else None
+        rows_written_snap, write_dropped_snap = self.writer.get_counters_snapshot()
         manifest.update({
             "tracer": {"version": self.version},
             "machine_id": capture_machine_id(),
@@ -1375,8 +1384,8 @@ def _write_manifest(self, started_at, stopped_at=None):
             "diagnostics": {
                 "attached_probes": self._attached_probes(),
                 "lost_events": dict(self._lost_counts),
-                "rows_written": dict(self.writer.rows_written),
-                "write_dropped": dict(getattr(self.writer, "write_dropped", {})),
+                "rows_written": rows_written_snap,
+                "write_dropped": write_dropped_snap,
                 "block": self._block_stats(),
             },
         })

src/tracer/WriterManager.py

@@ -84,6 +84,7 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         # Total rows written to disk per stream (keyed by buffer label), for the
         # session manifest — lets a consumer spot a stream whose probes attached
         # but produced no events.
+        self._rows_written_lock = threading.Lock()
         self.rows_written = {}
         # Rows that could NOT be persisted (write/flush raised — e.g. ENOSPC /
         # EDQUOT / closed handle) and were dropped after exceeding the on-error
@@ -951,16 +952,27 @@ def _write_buffer_to_file(self, buffer, file_handle, buffer_name: str):
                 buffer.popleft()
                 dropped += 1
             if dropped:
-                self.write_dropped[buffer_name] = (
-                    self.write_dropped.get(buffer_name, 0) + dropped
-                )
+                with self._rows_written_lock:
+                    self.write_dropped[buffer_name] = (
+                        self.write_dropped.get(buffer_name, 0) + dropped
+                    )
             logger("error", f"Error writing {buffer_name} buffer: {e}"
                    + (f" — dropped {dropped} rows past retry cap" if dropped else ""))
             return
 
-        self.rows_written[buffer_name] = (
-            self.rows_written.get(buffer_name, 0) + len(drained)
-        )
+        with self._rows_written_lock:
+            self.rows_written[buffer_name] = (
+                self.rows_written.get(buffer_name, 0) + len(drained)
+            )
+
+    def get_rows_written_snapshot(self) -> dict[str, int]:
+        with self._rows_written_lock:
+            return dict(self.rows_written)
+
+    def get_counters_snapshot(self) -> tuple[dict[str, int], dict[str, int]]:
+        """Return (rows_written, write_dropped) snapshots under one lock."""
+        with self._rows_written_lock:
+            return dict(self.rows_written), dict(self.write_dropped)
 
     def write_to_disk(self):
         """Write all buffered data to disk using parallel threads."""