NVIDIA-NeMo · eric-tramel · May 21, 2026 · May 21, 2026 · May 21, 2026 · May 21, 2026
@@ -24,6 +24,7 @@ class JinjaRenderingEngine(StrEnum):
     "RunConfig.throttle and ThrottleConfig are deprecated. Use RunConfig.request_admission with "
     "RequestAdmissionTuningConfig for supported advanced request-admission tuning."
 )
+_PROGRESS_BAR_DEPRECATION_MESSAGE = "RunConfig.progress_bar is deprecated. Use RunConfig.display_tui instead."
 
 
 class RequestAdmissionTuningConfig(ConfigBase):
@@ -142,9 +143,9 @@ class RunConfig(ConfigBase):
             Default is 0.
         async_trace: If True, collect per-task tracing data when using the async engine
             (DATA_DESIGNER_ASYNC_ENGINE=1). Has no effect on the sync path. Default is False.
-        progress_bar: If True, display sticky ANSI progress bars instead of periodic log lines
-            during generation. Requires a TTY; falls back to log lines in non-TTY environments.
-            Default is False.
+        display_tui: If True, display the terminal throughput TUI instead of periodic
+            log lines during generation. Requires a TTY; falls back to log lines in
+            non-TTY environments. Default is True.
         progress_interval: How often (in seconds) the async progress reporter emits a
             consolidated log block. Must be > 0. Default is 5.0.
         preserve_dropped_columns: If True, write columns removed by drop processors to
@@ -172,7 +173,7 @@ class RunConfig(ConfigBase):
     max_conversation_restarts: int = Field(default=5, ge=0)
     max_conversation_correction_steps: int = Field(default=0, ge=0)
     async_trace: bool = False
-    progress_bar: bool = False
+    display_tui: bool = True
     progress_interval: float = Field(default=5.0, gt=0.0)
     preserve_dropped_columns: bool = Field(
         default=True,
@@ -191,9 +192,22 @@ class RunConfig(ConfigBase):
 
     @model_validator(mode="before")
     @classmethod
-    def translate_deprecated_throttle_config(cls, data: Any) -> Any:
-        if isinstance(data, dict) and "throttle" in data:
-            normalized = dict(data)
+    def translate_deprecated_fields(cls, data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+
+        normalized = dict(data)
+
+        if "progress_bar" in normalized:
+            progress_bar = normalized.pop("progress_bar")
+            normalized.setdefault("display_tui", progress_bar)
+            warnings.warn(
+                _PROGRESS_BAR_DEPRECATION_MESSAGE,
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        if "throttle" in normalized:
             throttle = normalized.pop("throttle")
             if normalized.get("request_admission") is not None:
                 raise ValueError(
@@ -211,7 +225,25 @@ def translate_deprecated_throttle_config(cls, data: Any) -> Any:
                 stacklevel=2,
             )
             return normalized
-        return data
+        return normalized
+
+    @property
+    def progress_bar(self) -> bool:
+        warnings.warn(
+            _PROGRESS_BAR_DEPRECATION_MESSAGE,
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return self.display_tui
+
+    @progress_bar.setter
+    def progress_bar(self, value: bool) -> None:
+        warnings.warn(
+            _PROGRESS_BAR_DEPRECATION_MESSAGE,
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        self.display_tui = value
 
     @model_validator(mode="after")
     def normalize_shutdown_settings(self) -> Self:

@@ -24,6 +24,37 @@ def test_run_config_accepts_native_renderer() -> None:
     assert JinjaRenderingEngine(run_config.jinja_rendering_engine) == JinjaRenderingEngine.NATIVE
 
 
+def test_run_config_defaults_to_display_tui_enabled() -> None:
+    assert RunConfig().display_tui is True
+
+
+def test_run_config_accepts_display_tui() -> None:
+    assert RunConfig(display_tui=False).display_tui is False
+
+
+def test_run_config_progress_bar_shim_translates_to_display_tui() -> None:
+    with pytest.warns(DeprecationWarning, match="RunConfig.progress_bar.*RunConfig.display_tui"):
+        run_config = RunConfig(progress_bar=False)
+
+    assert run_config.display_tui is False
+
+
+def test_run_config_progress_bar_property_getter_warns() -> None:
+    run_config = RunConfig(display_tui=False)
+
+    with pytest.warns(DeprecationWarning, match="RunConfig.progress_bar.*RunConfig.display_tui"):
+        assert run_config.progress_bar is False
+
+
+def test_run_config_progress_bar_property_setter_warns() -> None:
+    run_config = RunConfig(display_tui=False)
+
+    with pytest.warns(DeprecationWarning, match="RunConfig.progress_bar.*RunConfig.display_tui"):
+        run_config.progress_bar = True
+
+    assert run_config.display_tui is True
+
+
 def test_run_config_preserves_dropped_columns_by_default() -> None:
     assert RunConfig().preserve_dropped_columns is True
 

@@ -33,6 +33,7 @@ bump = true
 [tool.hatch.metadata.hooks.uv-dynamic-versioning]
 dependencies = [
     "anyascii>=0.3.3,<1",
+    "asciichartpy>=1.5.25,<2",
     "chardet>=3.0.2,<6",  # Pulled in by sqlfluff
     "cryptography>=46.0.7,<47", # 46.0.7 fixes CVE-2026-39892 pulled in by mcp
     "data-designer-config=={{ version }}",

@@ -7,6 +7,7 @@
 
 import asyncio
 import concurrent.futures
+import contextlib
 import inspect
 import logging
 from typing import TYPE_CHECKING, Any
@@ -86,6 +87,11 @@ def generate(self, *args: Any, **kwargs: Any) -> tuple[Any, list]:
                 "Use 'await model.agenerate()' in async custom columns."
             )
 
+        from data_designer.engine.context import (
+            current_generation_column,
+            current_run_cancel_event,
+            is_run_cancellation_requested,
+        )
         from data_designer.engine.dataset_builders.utils.async_concurrency import ensure_async_engine_loop
 
         # Honor a per-call ``timeout=`` override (passed straight through to the
@@ -99,10 +105,41 @@ def generate(self, *args: Any, **kwargs: Any) -> tuple[Any, list]:
         conversation_restarts = int(kwargs.get("max_conversation_restarts", 0) or 0)
         bridge_timeout = _compute_bridge_timeout(per_request_timeout, correction_steps, conversation_restarts)
 
+        if is_run_cancellation_requested():
+            raise asyncio.CancelledError
+
+        column = current_generation_column.get()
+        cancel_event = current_run_cancel_event.get()
+
+        async def agenerate_with_bridge_context() -> tuple[Any, list]:
+            column_token = current_generation_column.set(column)
+            cancel_token = current_run_cancel_event.set(cancel_event)
+            try:
+                if is_run_cancellation_requested():
+                    raise asyncio.CancelledError
+                return await facade.agenerate(*args, **kwargs)
+            finally:
+                # Cross-thread cancellation can close the coroutine from a
+                # different context after it has started. The task context is
+                # being discarded either way, so avoid an unraisable reset error.
+                with contextlib.suppress(ValueError):
+                    current_run_cancel_event.reset(cancel_token)
+                with contextlib.suppress(ValueError):
+                    current_generation_column.reset(column_token)
+
         loop = ensure_async_engine_loop()
-        future = asyncio.run_coroutine_threadsafe(facade.agenerate(*args, **kwargs), loop)
+        coro = agenerate_with_bridge_context()
+        try:
+            future = asyncio.run_coroutine_threadsafe(coro, loop)
+        except RuntimeError as exc:
+            coro.close()
+            if is_run_cancellation_requested() or "interpreter shutdown" in str(exc):
+                raise asyncio.CancelledError from exc
+            raise
         try:
             return future.result(timeout=bridge_timeout)
+        except concurrent.futures.CancelledError as exc:
+            raise asyncio.CancelledError from exc
         except concurrent.futures.TimeoutError as exc:
             future.cancel()
             # Demoted to debug: the raised ModelTimeoutError already surfaces

@@ -4,11 +4,26 @@
 from __future__ import annotations
 
 from contextvars import ContextVar
+from threading import Event
 
 # Set by the async scheduler before executing each task.
 # Value: (current_rg_index, total_rg_count) or None.
 current_row_group: ContextVar[tuple[int, int] | None] = ContextVar("current_row_group", default=None)
 
+# Set by the async scheduler before executing each task so model usage can be
+# attributed even when scheduler telemetry context is not available.
+current_generation_column: ContextVar[str | None] = ContextVar("current_generation_column", default=None)
+
+# Shared cancellation signal for sync generator work running in thread-pool
+# workers. Context variables copy the Event object into worker threads, and the
+# scheduler flips the Event on cancellation.
+current_run_cancel_event: ContextVar[Event | None] = ContextVar("current_run_cancel_event", default=None)
+
+
+def is_run_cancellation_requested() -> bool:
+    cancel_event = current_run_cancel_event.get()
+    return cancel_event.is_set() if cancel_event is not None else False
+
 
 def format_row_group_tag() -> str:
     """Return a '(x/X) ' prefix if a row group context is active, else ''."""