comet-ml
diff --git a/‎sdks/python/src/opik/__init__.py‎
Lines changed: 10 additions & 0 deletions b/‎sdks/python/src/opik/__init__.py‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎sdks/python/src/opik/api_objects/connection_resources.py‎
Lines changed: 42 additions & 14 deletions b/‎sdks/python/src/opik/api_objects/connection_resources.py‎
Lines changed: 42 additions & 14 deletions
diff --git a/‎sdks/python/src/opik/api_objects/opik_client.py‎
Lines changed: 88 additions & 7 deletions b/‎sdks/python/src/opik/api_objects/opik_client.py‎
Lines changed: 88 additions & 7 deletions
@@ -24,6 +24,12 @@
 from .api_objects.trace import Trace
 from .configurator.configure import configure
 from .decorator.tracker import flush_tracker, track
+from .message_processing.data_loss import (
+    ErrorsReport,
+    FailedMessageInfo,
+    FailureReason,
+    FlushResult,
+)
 from .evaluation import (
     evaluate,
     evaluate_experiment,
@@ -68,6 +74,10 @@
     "ExperimentItemReferences",
     "track",
     "flush_tracker",
+    "FlushResult",
+    "FailedMessageInfo",
+    "FailureReason",
+    "ErrorsReport",
     "Opik",
     "get_global_client",
     "set_global_client",
 
@@ -30,6 +30,8 @@
 from ..file_upload import upload_manager
 from ..healthcheck import connection_monitor, connection_probe
 from ..message_processing import (
+    data_loss,
+    flush_reporter,
     message_queue,
     permissions,
     streamer,
@@ -61,6 +63,7 @@ def __init__(
         file_upload_manager: upload_manager.FileUploadManager,
         replay_manager: replay_manager.ReplayManager,
         streamer: streamer.Streamer,
+        data_loss_tracker: data_loss.DataLossTracker,
         flush_timeout: Optional[int],
     ) -> None:
         self.httpx_client = httpx_client
@@ -69,14 +72,18 @@ def __init__(
         self.file_upload_manager = file_upload_manager
         self.replay_manager = replay_manager
         self.streamer = streamer
+        self.flush_reporter = flush_reporter.FlushReporter(
+            streamer=streamer,
+            data_loss_tracker=data_loss_tracker,
+        )
         self.flush_timeout = flush_timeout
 
-    def close(self, timeout: Optional[int], *, flush: bool) -> None:
+    def close(self, timeout: Optional[int], *, flush: bool) -> bool:
         # Drain/stop the streamer (consumer threads, replay, batch preprocessor);
         # on flush=True it also flushes pending file uploads.
         # Closing the streamer also stops and joins the replay manager (its own
         # daemon thread), so there is no separate replay teardown to do here.
-        self.streamer.close(timeout, flush=flush)
+        flushed = self.streamer.close(timeout, flush=flush)
         # Stop the upload worker pool too, so eviction doesn't leave its threads
         # running. wait=flush mirrors the streamer: block for in-flight uploads
         # on a durable close, return immediately on fire-and-forget teardown.
@@ -90,19 +97,23 @@ def close(self, timeout: Optional[int], *, flush: bool) -> None:
             # daemon threads to finish in-flight requests, so closing the pool
             # here would race them — leave it for GC / process-exit close_all.
             self.httpx_client.close()
+        return flushed
 
-    def flush(self, timeout: Optional[int]) -> None:
+    def flush(self, timeout: Optional[int]) -> bool:
         """Drain the shared message queue without tearing the bundle down.
 
         Used when a handle releases with ``flush=True`` while other handles still
         share the bundle: the queued data is persisted now, but the transport
         stays alive for the remaining handles.
+
+        Returns whether the queue drained fully within ``timeout``.
         """
-        self.streamer.flush(timeout)
+        return self.streamer.flush(timeout)
 
 
 def _create_replay_manager(
-    config: opik_config.OpikConfig, httpx_client: httpx.Client
+    config: opik_config.OpikConfig,
+    httpx_client: httpx.Client,
 ) -> replay_manager.ReplayManager:
     probe = connection_probe.ConnectionProbe(
         base_url=config.url_override,
@@ -156,6 +167,8 @@ def create_connection_resources(
         worker_count=config.file_upload_background_workers,
     )
 
+    data_loss_tracker = data_loss.DataLossTracker()
+
     fallback_replay = _create_replay_manager(config, httpx_client_)
 
     message_processor = message_processors_chain.create_message_processors_chain(
@@ -166,6 +179,7 @@ def create_connection_resources(
             retry_interval_seconds=config.unauthorized_message_type_retry_interval,
             max_retry_count=config.unauthorized_message_type_max_retry_count,
         ),
+        data_loss_tracker=data_loss_tracker,
     )
     streamer_ = streamer_constructors.construct_online_streamer(
         file_uploader=file_uploader,
@@ -186,6 +200,7 @@ def create_connection_resources(
         file_upload_manager=file_uploader,
         replay_manager=fallback_replay,
         streamer=streamer_,
+        data_loss_tracker=data_loss_tracker,
         flush_timeout=config.default_flush_timeout,
     )
 
@@ -236,12 +251,16 @@ def __init__(
 
     def release(
         self, timeout: Optional[int], *, flush: bool = True, close_on_zero: bool
-    ) -> None:
+    ) -> Optional[bool]:
+        """Release this handle's reference. Returns the authoritative flush
+        outcome when this release performed the drain (an explicit
+        ``flush=True`` release), ``None`` otherwise (already released, or a GC
+        finalizer that does no network I/O)."""
         with self._once_lock:
             if self._released:
-                return
+                return None
             self._released = True
-        self._manager.release(
+        return self._manager.release(
             self._key, timeout, flush=flush, close_on_zero=close_on_zero
         )
 
@@ -326,7 +345,7 @@ def release(
         *,
         flush: bool = True,
         close_on_zero: bool,
-    ) -> None:
+    ) -> Optional[bool]:
         # Durability under sharing: an explicit ``end(flush=True)`` on a handle
         # that still shares its bundle must drain the shared queue *before* this
         # handle gives up its reference. Flushing while our reference is still
@@ -337,6 +356,12 @@ def release(
         # bundle; a sole holder's ``close(flush=True)`` below already drains
         # durably. A GC finalizer (``close_on_zero=False``) never does network
         # I/O, so it never pre-flushes.
+        # Authoritative flush outcome for the caller: set by whichever branch
+        # below actually drains — the shared pre-flush or the last-reference
+        # close. Stays None when this release did no draining (GC finalizer, or
+        # a bundle already released elsewhere), so the caller can tell "not
+        # confirmed" apart from "confirmed not flushed".
+        flushed: Optional[bool] = None
         if flush and close_on_zero:
             with self._lock:
                 entry = self._entries.get(key)
@@ -346,7 +371,7 @@ def release(
                     else None
                 )
             if shared_bundle is not None:
-                shared_bundle.flush(timeout)
+                flushed = shared_bundle.flush(timeout)
 
         # Now drop our reference. Because we only decrement here — after any
         # pre-flush above has completed — a close can never run while another
@@ -355,10 +380,10 @@ def release(
         with self._lock:
             entry = self._entries.get(key)
             if entry is None:
-                return
+                return flushed
             entry.refcount -= 1
             if entry.refcount > 0:
-                return
+                return flushed
             if not close_on_zero:
                 # The last reference was dropped by a GC finalizer (see
                 # ``Opik._acquire_shared_resources``). Only the refcount
@@ -367,13 +392,16 @@ def release(
                 # never happen inside garbage collection. Leave the bundle
                 # cached so a later same-identity ``acquire`` reuses it, or
                 # ``close_all`` disposes it at process exit.
-                return
+                return flushed
             # Evict before close, under the lock, so a concurrent acquire never
             # receives a bundle that is being torn down.
             del self._entries[key]
             bundle = entry.resources
 
-        bundle.close(timeout, flush=flush)
+        closed_flushed = bundle.close(timeout, flush=flush)
+        # A non-draining teardown has no flush outcome to report — return None
+        # (not close()'s bool) so the result stays "no drain happened here".
+        return closed_flushed if flush else None
 
     def close_all(self, *, flush: bool = True) -> None:
         """Close and evict every cached bundle. Registered as the process
 
@@ -71,6 +71,7 @@
     url_helpers,
 )
 from ..message_processing import (
+    data_loss,
     messages,
 )
 from ..message_processing.batching import sequence_splitter
@@ -217,6 +218,8 @@ def _bind_resources(self) -> None:
         self._rest_client = self._resources.rest_client
         self.__internal_api__message_processor__ = self._resources.message_processor
         self._streamer = self._resources.streamer
+        self._flush_reporter = self._resources.flush_reporter
+        self._last_flush_result: Optional[data_loss.FlushResult] = None
 
     def _display_trace_url(self, trace_id: str, project_name: str) -> None:
         project_url = url_helpers.get_project_url_by_trace_id(
@@ -1927,9 +1930,13 @@ def get_experiment_by_id(self, id: str) -> experiment.Experiment:
             project_name=experiment_public.project_name,
         )
 
-    def end(self, timeout: Optional[int] = None, *, flush: bool = True) -> None:
+    def end(
+        self, timeout: Optional[int] = None, *, flush: bool = True
+    ) -> Optional[data_loss.FlushResult]:
         """
-        End the Opik session and submit all pending messages.
+        End the Opik session, releasing this client's connection reference. When
+        ``flush`` is True (the default), all pending messages are submitted
+        first; when ``flush`` is False, anything still queued is dropped.
 
         Connection resources are shared and ref-counted across clients with a
         matching configuration: this releases the current client's reference.
@@ -1957,28 +1964,102 @@ def end(self, timeout: Optional[int] = None, *, flush: bool = True) -> None:
         is shared — it may still succeed by riding another live client's
         resources. Do not rely on either outcome; create a new client instead.
 
+        The outcome is also available afterwards via :attr:`last_flush_result`.
+
         Returns:
-            None
+            The flush outcome (including any data-loss detail) when ``flush`` is
+            True; ``None`` when ``flush`` is False (nothing was flushed).
         """
         timeout = timeout if timeout is not None else self._flush_timeout
+        marker = self._flush_reporter.marker()
         # Explicit teardown on a user thread, so close on the last reference
         # (close_on_zero=True). Releasing is idempotent, so the detached GC
-        # finalizer cannot double-decrement.
-        self._lease.release(timeout, flush=flush, close_on_zero=True)
+        # finalizer cannot double-decrement. release() returns the authoritative
+        # flush outcome computed inside the drain (streamer.flush) — the same
+        # source flush() uses — rather than the weaker queue_size()==0 proxy,
+        # which can read empty on the pop-vs-processed race and while file
+        # uploads are still in flight.
+        flushed = self._lease.release(timeout, flush=flush, close_on_zero=True)
         self._finalizer.detach()
+        if not flush:
+            return None
+        if flushed is None:
+            # No drain ran on this call — e.g. a repeated end() after the client
+            # was already released. Keep the outcome from the release that did
+            # the work rather than overwriting it with a spurious not-flushed
+            # result, so end() is idempotent.
+            return self._last_flush_result
+        self._last_flush_result = self._flush_reporter.build_result(
+            marker, flushed=flushed
+        )
+        return self._last_flush_result
 
     def flush(self, timeout: Optional[int] = None) -> bool:
         """
         Flush the streamer to ensure all messages are sent.
 
+        Attachment/file upload *failures* are not counted in the data-loss
+        detail (``dropped_*`` / ``failures``), but an incomplete upload still
+        makes the flush report as not fully flushed — so ``flushed`` (and hence
+        the returned bool) does reflect uploads. Never raises and never blocks
+        beyond ``timeout``: an observability SDK must not disrupt the app it
+        instruments. Detailed outcome — including any data that was dropped — is
+        available via :attr:`last_flush_result`.
+
         Args:
             timeout (Optional[int]): The timeout for flushing the streamer. Once the timeout is reached, the flush method will return regardless of whether all messages have been sent.
 
         Returns:
-            True if all messages have been sent within specified timeout, False otherwise.
+            True if all messages were delivered within the timeout with no data
+            loss; False if the timeout was hit or any message was dropped.
         """
         timeout = timeout if timeout is not None else self._flush_timeout
-        return self._streamer.flush(timeout)
+        try:
+            marker = self._flush_reporter.marker()
+            flushed = self._streamer.flush(timeout)
+            self._last_flush_result = self._flush_reporter.build_result(
+                marker, flushed=flushed
+            )
+            return self._last_flush_result.success
+        except Exception:
+            # An observability SDK must not disrupt the app it instruments: a
+            # failure inside flush is reported as "not flushed", never raised.
+            # Record a failed outcome so last_flush_result reflects this attempt
+            # rather than keeping a stale prior success. Built directly (not via
+            # build_result, which may itself be what raised) so it cannot re-raise.
+            LOGGER.error("Opik flush failed unexpectedly", exc_info=True)
+            self._last_flush_result = data_loss.FlushResult(
+                flushed=False,
+                remaining_queue_size=0,
+                dropped_messages=0,
+                dropped_items=0,
+                failures=(),
+            )
+            return False
+
+    @property
+    def last_flush_result(self) -> Optional[data_loss.FlushResult]:
+        """Outcome of the most recent ``flush()``/``end()`` on this client.
+
+        ``None`` until the first flush.
+        """
+        return self._last_flush_result
+
+    def get_errors_report(self) -> data_loss.ErrorsReport:
+        """Report of messages the background sender terminally dropped.
+
+        Unlike :attr:`last_flush_result`, which is scoped to a single flush, this
+        reports the sender's retained data-loss history — including drops that
+        happened before or between flushes.
+
+        The report is **capped**: the total counts are exact, but the per-drop
+        ``failures`` list keeps only the most recent entries (bounded, drop-oldest)
+        so it never grows without bound — see :class:`~opik.ErrorsReport`.
+
+        The sender is shared across clients with a matching configuration, so
+        the report may include drops from sibling clients on the same connection.
+        """
+        return self._flush_reporter.build_errors_report()
 
     def __internal_api__drain_to_processors__(
         self, timeout: Optional[float] = None