feat: add OpenTelemetry metrics for create jobs

[eric-tramel]

📋 Summary

What. Adds default, pull-based OpenTelemetry metrics for DataDesigner.create() on a loopback Prometheus endpoint controlled by RunConfig.

Why. Gives operators current create progress, throughput, scheduler outcomes, model activity, request completion/latency, errors, and safe log-volume signals without making task code depend on OpenTelemetry or adding parallel instrumentation paths.

The interface layer owns a private OTel meter provider, registry, and HTTP listener. Existing generic engine scheduler and request-event seams feed the exporter, while a metric-only logging handler tees severity counts without exporting log bodies. Telemetry failures are contained and cannot change create results.

🔗 Related Issue

None.

🔄 Changes

• Adds RunConfig.otel_metrics_port, defaulting to loopback port 9464; None disables telemetry for an invocation.
• Exposes seven instruments at /metrics:
  • data_designer.create.duration
  • data_designer.dataset.records
  • data_designer.dataset.progress
  • data_designer.scheduler.events
  • data_designer.model.request.active
  • gen_ai.client.operation.duration
  • data_designer.log.records
• Reports lifecycle-aware current progress and active model requests, including standard gen_ai.operation.name, gen_ai.provider.name, and gen_ai.request.model attributes.
• Extends the existing generic scheduler/request event seams with cheap event-interest filtering, terminal scheduler outcomes, request timing, and production sink wiring.
• Guarantees sync/async request-lease cleanup if started-event delivery is interrupted, and applies request interest gates before expensive attempt snapshots are built.
• Reports early shutdown as a distinct terminal outcome while preserving the scheduler completion diagnostics contract.
• Keeps OTel SDK/exporter ownership in the interface package and preserves config → engine dependency boundaries and lazy imports.
• Serializes run teardown with event admission and state cleanup so delayed events or overlapping creates cannot publish stale gauges.
• Retires an old listener after releasing the runtime lock so a concurrent scrape cannot deadlock an idle port rebind.
• Bounds metric attributes and excludes prompts, responses, request/run/task identifiers, exception text, environment resource attributes, and log bodies.
• Documents scrape configuration, metric semantics, checkpoint-granularity progress, listener lifecycle, security posture, and raw-log limitations.

📖 Usage

The default create path starts http://127.0.0.1:9464/metrics:

import data_designer.config as dd
from data_designer.interface import DataDesigner

designer = DataDesigner()
designer.create(config_builder)

Choose another loopback port or disable telemetry:

designer.set_run_config(dd.RunConfig(otel_metrics_port=9465))
# designer.set_run_config(dd.RunConfig(otel_metrics_port=None))

Prometheus can scrape 127.0.0.1:9464 at /metrics. The listener is process-wide and remains available after a create job so final cumulative metrics can be collected. Generated-record totals and progress advance at durable row-group checkpoints; the active-request gauge and request-duration count provide live activity between checkpoints.

📊 External dashboard validation

A local external Prometheus/Grafana stack scraped the listener every 2 seconds and rendered current listener health, active jobs, durable progress, generated records, active requests by exact model, attempts/second, p95 latency, scheduler totals, failures, and log volume.

The end-to-end demonstration generated 512 records through 1,536 logical model tasks using the nvidia-internal provider and three non-8B models, each configured with max_tokens=1024:

• nvidia/openai/gpt-oss-20b
• nvidia/nvidia/nemotron-nano-31b-v3
• nvidia/openai/gpt-oss-120b

All 1,536 logical tasks completed successfully. The external collector observed two transient internal_server attempts on the 120B model, two matching task.retry_deferred scheduler events, and final job success. Dashboard queries use raw counters for event/failure totals, short rate windows aligned with the 2-second scrape interval, a lifecycle-aware progress gauge, and model-separated GenAI series so the panels represent current state instead of long-window residual activity.

The original POC capture is retained below; the validated dashboard now includes the live-state and 2-second refresh refinements described above.

📝 Changelog

• Adds default local OpenTelemetry/Prometheus metrics for DataDesigner create jobs with a RunConfig port/disable option.

🧪 Testing

• make check-all — passed

• make test — 3,907 passed, 1 existing skip

• Focused scheduler, request-admission, custom-bridge, and OpenTelemetry review-fix suite — 302 passed

• Seven-persona delegated code review — no actionable findings remain after targeted follow-up

• Deterministic delayed-event, overlapping-teardown, and unlocked-rebind regressions — passed

• Isolated package-install suites — passed during feature validation

• Import-performance and lazy-import checks — passed

• Fern validation — 0 errors, 1 existing warning

• Live nvidia-internal multi-model run — 512 records and 1,536 logical tasks completed successfully while an external Prometheus/Grafana stack observed execution

• make test passes

• Unit tests added/updated

• E2E behavior validated

👀 Review areas

• RunConfig surface
• Generic engine observability seam
• OpenTelemetry adapter and listener
• User documentation

✅ Checklist

• Follows commit message conventions
• Commits are signed off (DCO)
• Architecture docs updated (if applicable)

---

Description updated with AI

[github-actions]

Fern preview: https://nvidia-preview-pr-802.docs.buildwithfern.com/nemo/datadesigner

Fern previews include the docs-website version archive with PR changes synced into latest. Notebook tutorials are rendered without execution outputs in previews.

[github-actions]

Code Review: PR #802 — feat: add OpenTelemetry metrics for create jobs

Author: eric-tramel · Base: main ← codex/otel-create-observability
Size: +1,461 / −47 across 23 files · State: OPEN

Summary

This PR adds default, pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create(). A process-owned OpenTelemetryRuntime (in the interface package) owns the meter provider, Prometheus CollectorRegistry, and a loopback HTTP listener bound to 127.0.0.1:9464 (configurable via RunConfig.otel_metrics_port, None to disable). Five cumulative instruments are exposed: create duration, dataset records, scheduler events, GenAI request duration, and log-severity counts.

The design correctly respects the package layering. OTel SDK/exporter ownership stays in data-designer (interface); the engine gains only generic event-sink seams (accepts_scheduler_event interest filtering, terminal scheduler outcomes, request timing), and config gains only a plain integer field. The OpenTelemetryRuntime is passed into the engine as an anonymous SchedulerAdmissionEventSink / RequestAdmissionEventSink implementer — the engine never imports the interface. This is the right seam.

The change is unusually well-tested: privacy (secret-sentinel scrubbing), lifecycle/rebind, concurrent creates, exact-once emission, init-failure cleanup, real HTTP scrape, and a subprocess test asserting the OTel SDK is not imported when disabled. Failure containment is a clear design goal and is consistently applied.

Findings

Correctness

• _non_negative_int rejects bool but silently drops floats (minor). packages/data-designer/src/data_designer/integrations/opentelemetry.py:1298 — max(0, value) if isinstance(value, int) else 0. surviving_rows/dropped_rows are ints today so this is fine, but note isinstance(True, int) is True, so a stray bool would count as 1/0. Non-blocking; current callers only pass ints.

• Import-direction violations in structural analysis are false positives. The graphify report flags 27 "config → engine .set()" violations. These are misattributions: the .set() calls are runtime_correlation_provider.set() (a ContextVar) and OTel histogram/counter .set()/.add(), not config importing engine. I verified run_config.py imports nothing from engine (the only config change is a plain int | None Pydantic field). No real layering violation exists. See Structural Impact section below.

• run refactor preserves the finally-emit-once contract correctly. async_scheduler.py:1082+ restructures the try/with into try/except/finally so the terminal scheduler_job_completed event fires exactly once with outcome in {success, cancelled, error} and a failure-only error_type. The previous duplicate emit (health snapshot + separate job-completed inside the body) is removed. CancelledError is re-raised without being swallowed. Test test_scheduler_emits_one_terminal_event_without_replacing_failure covers both the error and cancelled paths and asserts the original exception is re-raised (exc_info.value is raised). Good.

• model_request_executor timing refactor is sound. Both sync and async paths move to a single finally that emits model_request_completed once with outcome + duration_seconds. The nested inner try/finally around call() ensures duration_seconds is captured even when the call raises, before the outer handler classifies the outcome. outcome is initialized to "unexpected_exception" so an unexpected escape still emits a defined value. Correct.

Design / Architecture

• Ambient run-id inheritance is the linchpin — worth calling out. observe_create sets a runtime_correlation_provider ContextVar with a fresh run-<uuid>; the scheduler now inherits that ambient run_id when no explicit run_id is passed (async_scheduler.py:230), and _event_is_active / record_log gate on membership in _active_run_ids. This is how metrics get scoped to a specific create invocation across threads. The concurrency test (test_concurrent_data_designer_creates_share_one_exporter) and the log-scoping test exercise this. The mechanism is subtle but correct and well-covered.

• Repeated self._open_telemetry if self._run_config.otel_metrics_port is not None else None (3×) in data_designer.py (lines 745, 746, 754). Minor DRY nit — a small _active_event_sink() helper would remove the triplication and keep the enable/disable rule in one place. Non-blocking.

• getattr(self._resource_provider, "scheduler_event_sink", None) in dataset_builder.py:976 uses a defensive getattr even though ResourceProvider now declares the field. Since it's a typed model attribute, self._resource_provider.scheduler_event_sink would be safe and clearer. Presumably defensive for stub/mock providers in tests — acceptable, but a direct access would be more honest about the contract.

• Log handler attaches to the data_designer root logger process-wide. _initialize adds _MetricLogHandler to logging.getLogger("data_designer"). record_log correctly gates on the active-run set, so counts are scoped, but the handler stays attached for the process lifetime (removed only on shutdown/atexit). This is intentional (documented "process-wide listener remains available"), and the handler is cheap + exception-swallowing. Fine, but reviewers should be aware every data_designer.* log record now passes through this handler once metrics have been initialized once.

Security / Privacy

• Privacy posture is strong and explicitly tested. _bounded_attribute whitelists [alnum . _ -], truncates to 128 chars, and collapses to _OTHER; unknown GenAI operations and domains collapse to _OTHER; model IDs, request/attempt IDs, and raw log bodies are never exported. test_runtime_exports_bounded_metrics_without_log_bodies asserts no SECRET-* sentinel (including OTEL_RESOURCE_ATTRIBUTES injection) leaks into the scrape, and target_info is restricted to service_name/service_version. Error text is excluded — only error.type (class name) is recorded. This is exactly right for a metrics endpoint.

• Loopback-only bind + documented warning. Binds to 127.0.0.1 only, with a docs <Warning> that there is no auth and it's not for remote exposure. Reasonable for the stated same-host-Prometheus use case. Defaulting the endpoint on (port 9464) is a deliberate product choice; operators who object can set None. Worth confirming 9464 (the OTel Prometheus convention) doesn't collide with anything else expected on target hosts — documented as configurable, so acceptable.

Performance

• Import cost contained. OTel/prometheus SDK imports are deferred into _initialize (called lazily on first activation), and a subprocess test asserts opentelemetry.sdk / opentelemetry.exporter.prometheus are absent from sys.modules when disabled. Consistent with the project's fast-import invariant. The PR notes import-perf/lazy-import checks pass.

• Per-event overhead is minimal. accepts_scheduler_event short-circuits on an empty active-run set before any event construction, and scheduler_event_sink_accepts avoids task_admission_snapshot()/health-diagnostics work when the sink rejects the kind (verified by test_scheduler_filter_skips_snapshot_and_health_diagnostics). Good — this avoids paying telemetry cost on the hot path when disabled.

Testing

• Comprehensive: config bounds/serialization, engine sink forwarding, filtering, ambient correlation, terminal-event exactness, request timing, and interface-level end-to-end scrapes (including concurrent and repeated creates proving cumulative accumulation and exact-once export). Failure-containment paths (port in use, init failure, log-handler failure, idempotent shutdown) are all covered.
• One gap worth noting: no test asserts behavior when two different processes contend for port 9464 (only in-process rebind/reuse). This is inherently hard to test and the _activate_run exception handler covers bind failure (test_runtime_port_failure_does_not_escape), so acceptable.

Docs

• The new "Local OpenTelemetry Metrics" section in architecture-and-performance.mdx is accurate against the implementation: default port, disable semantics, the five instruments, listener lifecycle, the loopback/no-auth warning, and the multiprocessing limitation. The RunConfig.otel_metrics_port docstring and Pydantic description are consistent with each other and with the field bounds (ge=1, le=65535).

Nits

• opentelemetry.py:1298 _non_negative_int — see bool note above.
• data_designer.py — 3× duplicated enable-check ternary; consider a helper.
• dataset_builder.py:976 — getattr where a typed attribute access would do.

Structural Impact

Pre-computed by graphify; annotated with reviewer verification.

Risk: HIGH (1 core abstraction(s) modified; 27 import direction violation(s))

• 20 Python files, 341 AST entities, 7/79 clusters

Import Direction Violations (27)

Legal direction: interface -> engine -> config

• ._is_missing_value() (config) --calls--> .set() (engine)
• column_types() (config) --calls--> .set() (engine)
• generate_analysis_report() (config) --calls--> .set() (engine)
• _validate_skip_scope() (config) --calls--> .set() (engine)
• required_columns() (config) --calls--> .set() (engine)
• +22 more

Core Abstractions Modified

• AsyncTaskScheduler - Add support for images in preview_result.display_sample_record() #5 most connected (146 deps)

High-Connectivity Changes

• AsyncTaskScheduler (146 deps) in packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py
• .set() (116 deps) in packages/data-designer-engine/src/data_designer/engine/observability.py
• DataDesigner (86 deps) in packages/data-designer/src/data_designer/interface/data_designer.py
• DatasetBuilder (73 deps) in packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py
• ResourceProvider (73 deps) in packages/data-designer-engine/src/data_designer/engine/resources/resource_provider.py
• JinjaRenderingEngine (66 deps) in packages/data-designer-config/src/data_designer/config/run_config.py
• RunConfig (59 deps) in packages/data-designer-config/src/data_designer/config/run_config.py
• ModelRequestExecutor (51 deps) in packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py
• +209 more

Cross-Package Dependencies

• mcp_command() (interface) --calls--> .run() (engine)
• models_command() (interface) --calls--> .run() (engine)
• providers_command() (interface) --calls--> .run() (engine)
• tools_command() (interface) --calls--> .run() (engine)
• .run_create() (interface) --calls--> .model_copy() (config)
• .__init__() (interface) --calls--> .set() (engine)
• +1070 more

Reviewer note on the 27 import-direction violations: these are false positives from call-graph attribution. The flagged config --calls--> .set() (engine) edges resolve to ContextVar.set() (runtime_correlation_provider) and OTel counter/histogram .set()/.add(), not config importing engine. I verified packages/data-designer-config/src/data_designer/config/run_config.py imports nothing from data_designer.engine — the only config change is a plain int | None Pydantic field. No real layering violation is introduced. The AsyncTaskScheduler and observability changes are additive event-sink seams; the OpenTelemetryRuntime is injected into the engine as an anonymous sink so the engine never imports the interface. The HIGH risk rating is driven by the god-node blast radius (scheduler, observability), which warrants the extra scrutiny applied above — but the changes to those nodes are backward-compatible additions, not signature-breaking edits.

Verdict

Approve (with minor, non-blocking nits). This is a high-quality, well-architected PR that respects the interface → engine → config layering, contains all telemetry failures so they cannot alter create results, and has strong privacy guarantees backed by explicit tests. The scheduler run() and request-executor timing refactors preserve their exactly-once emission contracts and re-raise the original exceptions. The structural-analysis HIGH rating reflects god-node connectivity, not a real regression: the flagged import-direction violations are call-graph false positives (ContextVar/OTel .set()), and the god-node edits are backward-compatible additions.

Suggested (optional) follow-ups: dedupe the 3× enable-check ternary in data_designer.py; drop the defensive getattr in dataset_builder.py now that the field is typed; consider guarding _non_negative_int against bool. None block merge.

[greptile-apps]

Greptile Summary

This PR adds pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create(), exposing seven instruments at a configurable loopback /metrics endpoint controlled by a new RunConfig.otel_metrics_port field (default 9464, None to disable). The interface layer owns a process-wide OpenTelemetryRuntime singleton that acts as both a SchedulerAdmissionEventSink and a RequestAdmissionEventSink, wiring existing engine seams to OTel counters, gauges, and histograms without requiring task code to depend on OTel.

• Adds otel_metrics_port to RunConfig and wraps DataDesigner.create() with an @_observe_create decorator that gates telemetry on per-run lifecycle, serializes teardown under a runtime lock, and calls _stop_server outside the lock to avoid a scrape-handler deadlock on port rebind.
• Extends async_scheduler.py to capture the ambient RuntimeCorrelation (set by observe_create) at scheduler init time and to emit scheduler_job_completed in a finally block with outcome, error-type, and (for success/early-shutdown only) full health diagnostics.
• Refactors model_request_executor.py to time the actual model call in an inner try/finally and always emit model_request_completed in the outer finally, consolidating outcome assignment and eliminating the previous per-branch duplicate emission.

Confidence Score: 5/5

Safe to merge. The new telemetry path is fully contained — failures are swallowed and cannot affect create results. Thread-safety is well-handled throughout with an RLock and careful sequencing of server shutdown outside the lock.

The implementation correctly handles all tricky concurrency corners: the scrape-handler deadlock on port rebind is fixed by stopping the old server after releasing the lock, gauge state is correctly cleaned up on both normal and error teardown paths, and the scheduler captures its run_id at init time on the calling thread rather than relying on async context propagation. The 28-test integration suite covers regression cases for delayed events, overlapping teardown, port rebind, and failure containment. No logic errors or correctness issues were found.

No files require special attention.

Important Files Changed

Filename	Overview
packages/data-designer/src/data_designer/integrations/opentelemetry.py	New file: process-owned OpenTelemetryRuntime singleton with per-run lifecycle management, event to metric translation, thread-safe gauge/counter state, and lazy OTel SDK initialization. _stop_server is correctly called outside self._lock to prevent a scrape-handler deadlock.
packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py	Captures ambient RuntimeCorrelation at scheduler init time for run_id propagation; moves scheduler_job_completed emission to the finally block with outcome/error_type; replaces null-sink checks with scheduler_event_sink_accepts() interest filtering.
packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py	Consolidates outcome tracking and model_request_completed emission into a single outer finally block; times only the actual model call (inner try/finally); eliminates duplicate per-branch emit_model_event calls.
packages/data-designer-config/src/data_designer/config/run_config.py	Adds otel_metrics_port: int
packages/data-designer-engine/src/data_designer/engine/observability.py	Adds scheduler_event_sink_accepts() and request_event_sink_accepts() helper functions with exception-safe interest filtering; adds @runtime_checkable to SchedulerAdmissionEventSink Protocol.
packages/data-designer/src/data_designer/interface/data_designer.py	Adds @_observe_create decorator to create() and wires _open_telemetry as request/scheduler event sink when otel_metrics_port is not None; reads run_config at call time, ensuring port and sink assignment are consistent.
packages/data-designer/tests/integrations/test_opentelemetry.py	28-test integration suite covering metric values, attribute bounding, scrape content, active-request gauge balance, port rebind, unlock-before-stop regression, log scoping, overlapping teardown, and failure containment.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Caller as Caller Thread
    participant OTel as OpenTelemetryRuntime
    participant Scheduler as AsyncTaskScheduler
    participant Executor as ModelRequestExecutor
    participant Prom as Prometheus HTTP

    Caller->>OTel: observe_create(port) sets RuntimeCorrelation(run_id)
    OTel->>OTel: _activate_run adds run_id to _active_run_ids
    Caller->>Scheduler: create() via run_coroutine_threadsafe
    Note over Scheduler: __init__ captures ambient run_id
    Scheduler->>OTel: emit scheduler_job_started
    loop per row-group
        Scheduler->>OTel: emit row_group_checkpointed
        Executor->>OTel: emit model_request_started (+1 gauge)
        Executor->>OTel: emit model_request_completed (-1 gauge + histogram)
    end
    Prom-->>OTel: GET /metrics scrape
    Scheduler->>OTel: emit scheduler_job_completed
    Caller->>OTel: observe_create finally block cleanup
    OTel->>OTel: _stop_server called OUTSIDE _lock

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Caller as Caller Thread
    participant OTel as OpenTelemetryRuntime
    participant Scheduler as AsyncTaskScheduler
    participant Executor as ModelRequestExecutor
    participant Prom as Prometheus HTTP

    Caller->>OTel: observe_create(port) sets RuntimeCorrelation(run_id)
    OTel->>OTel: _activate_run adds run_id to _active_run_ids
    Caller->>Scheduler: create() via run_coroutine_threadsafe
    Note over Scheduler: __init__ captures ambient run_id
    Scheduler->>OTel: emit scheduler_job_started
    loop per row-group
        Scheduler->>OTel: emit row_group_checkpointed
        Executor->>OTel: emit model_request_started (+1 gauge)
        Executor->>OTel: emit model_request_completed (-1 gauge + histogram)
    end
    Prom-->>OTel: GET /metrics scrape
    Scheduler->>OTel: emit scheduler_job_completed
    Caller->>OTel: observe_create finally block cleanup
    OTel->>OTel: _stop_server called OUTSIDE _lock

Loading

_{Reviews (4): Last reviewed commit: "fix: address observability review feedba..." | Re-trigger Greptile}

[andreatgretel]

One inline finding.

[andreatgretel]

[P2] Keep started-event emission inside the lease cleanup scope

The admission lease is already acquired here, but this emission occurs before the try that guarantees its release. If KeyboardInterrupt occurs while the sink is running, the coroutine exits with the lease still active. I reproduced active_lease_count == 1 on this head, while the same forced path on main releases it with the local_cancelled outcome. Initialize the timing and outcome variables first, then emit inside the protected try. The synchronous attempt has the same regression.

[nabinchha]

Nice work on this one, @eric-tramel — this is a well-structured, privacy-conscious observability layer with impressively thorough tests.

Summary

Adds default, pull-based OpenTelemetry/Prometheus metrics for DataDesigner.create() on a loopback endpoint controlled by RunConfig.otel_metrics_port (default 9464, None to disable). The interface package owns the OTel SDK, meter provider, and HTTP listener; existing engine scheduler/request event seams feed the exporter, and a metric-only log handler tees severity counts. The implementation matches the PR description, keeps the config → engine → interface layering intact, defers heavy imports into _initialize(), and contains telemetry failures so they can't change create results.

Findings

Warnings — Worth addressing

packages/data-designer/src/data_designer/interface/data_designer.py:173 + run_config.py:194 — Telemetry is on by default, opening a listening socket on every create()

• What: otel_metrics_port defaults to 9464, so every DataDesigner.create() in every environment now binds a TCP listener (loopback) and keeps it open until process exit. This is a new, always-on side effect for existing users who never asked for metrics.
• Why: In multi-instance hosts, CI, or notebook workflows this can surprise operators, and security scanners routinely flag open ports even on loopback. The graceful-degradation path (bind failure → warn → continue) is good, but the default posture is still "open a socket unless told not to."
• Suggestion: Worth a quick team confirmation that default-on is the intended contract vs. opt-in (e.g. default None, enabled explicitly). If default-on is deliberate, consider calling it out prominently in the release notes/changelog so operators aren't caught off guard. Not blocking if the team already signed off on this.

packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py:125,176 — Confirming the existing P2 (started-event emitted before the release-guarding try)

• What: I'm not re-raising new feedback here, just corroborating @andreatgretel's inline comment: _emit_model_event("model_request_started", ...) runs after the lease is acquired but before the try that guarantees release. _emit_model_event swallows Exception but not BaseException, so a KeyboardInterrupt during the started emit leaks the lease. This still reproduces on the current head.
• Why: It's a genuine (if narrow) resource-leak window, and the fix is cheap.
• Suggestion: Move the started emission inside the protected try (initialize started_at/duration_seconds/outcome first, then emit), as suggested in the inline thread. Same for both the sync and async attempt paths.

packages/data-designer-engine/src/data_designer/engine/models/clients/model_request_executor.py:272 — Request correlation relies on the contextvar being present on the calling thread

• What: _item() reads runtime_correlation_provider.current(), and emit_request_event drops any event whose captured_correlation has no run_id. The active-request gauge and gen_ai.client.operation.duration therefore only populate when the model call executes in a context that inherited the contextvar set by observe_create.
• Why: This is fine for the async path (tasks inherit the context), but for any sync/thread-pool execution mode that runs client calls on worker threads that didn't copy the context, current() returns None and those requests silently won't appear in the model-activity metrics. Want to make sure that's understood/acceptable.
• Suggestion: Could we confirm the concurrency modes that route through ModelRequestExecutor all preserve the contextvar (or add a targeted test for the sync/threaded path)? If sync-mode gaps are expected, a one-line note in the docs/metric description would set expectations.

packages/data-designer/src/data_designer/integrations/opentelemetry.py — OpenTelemetryRuntime is carrying four responsibilities (SRP)

• What: One ~500-line class owns (a) HTTP server + meter-provider lifecycle, (b) instrument definitions, (c) a domain state machine (active runs, dataset-progress math, in-flight request accounting), and (d) event adaptation for three sinks. The progress aggregation (_start_dataset_progress/_advance_dataset_progress/_finish_active_dataset/_active_dataset_progress, ~404-435) is a self-contained, OTel-agnostic state machine.
• Why: The progress semantics can only be tested by standing up a Prometheus server, and every concern shares the same _lock. This will get harder to reason about as instruments are added.
• Suggestion: Pull the progress math into a small DatasetProgressTracker (no OTel deps, directly unit-testable) and consider a frozen _MetricInstruments container for the instruments; leave OpenTelemetryRuntime as orchestration + lifecycle. Fine as a follow-up if it's too much for this PR. (Naming: DatasetProgressTracker stays un-prefixed since the whole point is direct testability — STYLEGUIDE prefers public over private for testable helpers — whereas _MetricInstruments is a dumb container tested through the runtime, so the underscore fits.)

packages/data-designer/src/data_designer/integrations/opentelemetry.py:84-96 — ~11 attributes typed Any sidestep the "typed code" invariant

• What: _registry, _meter_provider, _server, _create_duration, _dataset_records, … are all Any because the OTel types aren't imported at module scope.
• Why: AGENTS.md/STYLEGUIDE call out avoiding overly broad Any; as-is the checker can't catch e.g. calling .add() on a histogram vs .record() on a counter.
• Suggestion: Import the OTel/prometheus types under TYPE_CHECKING (the same pattern used for lazy heavy libs) and annotate precisely (Histogram, Counter, UpDownCounter, ObservableGauge, MeterProvider, CollectorRegistry, HTTPServer). This is a low-risk, high-clarity win and pairs naturally with the split above.

model_request_executor.py:287 + data_designer.py:745 — Hot-path events are built before the gate, and preview() pays for events it always drops

• What: _emit_model_event unconditionally builds the full RequestAdmissionEvent — including self._request_admission.pressure.snapshot(...) and the recursive _json_safe normalization in __post_init__ — whenever _event_sink is not None, then dispatches to emit_request_event, which drops it if the run isn't active. Unlike the scheduler path there's no accepts_request_event fast-path filter. Because _create_resource_provider wires the sink for both create and preview (only create activates a run), every model call on the preview path does this build-then-drop.
• Why: Model requests are the highest-volume events, and preview is latency-sensitive; the per-call snapshot + JSON normalization is wasted work there.
• Suggestion: Add an accepts_request_event(event_kind) gate mirroring scheduler_event_sink_accepts and check it before building the event / taking the pressure snapshot. That both restores symmetry with the scheduler path and makes the inactive-run/preview case exit cheaply (cleaner than special-casing preview at the wiring site).

Engine observability now has three inconsistent emission seams (strategic design debt) — observability.py, request_admission/controller.py:770-777, async_scheduler.py, model_request_executor.py

• What: The DIP itself is clean (engine defines the sink Protocols; the interface's OpenTelemetryRuntime implements them; correlation flows the other way via the runtime_correlation_provider contextvar). But the emission side now has three different styles side by side:
  • AsyncScheduler → single direct sink with an accepts_scheduler_event interest filter.
  • RequestAdmissionController._emit_events → direct sink and the global emit_request_admission_event callback registry (the latter is how progress/reporter.py consumes events), no filter.
  • ModelRequestExecutor → single direct sink only (so model_request_started/completed reach OTel but not the progress reporter), no filter.
• Why: There's no single answer to "how do I observe request activity?" The controller's dual fan-out also has mismatched reliability — the direct-sink call is try/except-guarded but the emit_request_admission_event(event) call on line 777 is not. As more consumers appear, this divergence compounds.
• Suggestion: Converge on one seam over time — e.g. make the progress reporter just another registered subscriber/sink, and give both scheduler and request paths the same optional accepts_* filter, so the model is uniformly "producer emits event → fan-out to N sinks." Bigger than this PR should take on, but worth capturing as the design debt this feature adds.

packages/data-designer-engine/src/data_designer/engine/observability.py:227 — accepts_scheduler_event is a phantom (undeclared) protocol method

• What: scheduler_event_sink_accepts probes getattr(sink, "accepts_scheduler_event", None), but that method isn't on the SchedulerAdmissionEventSink Protocol (which only declares emit_scheduler_event).
• Why: This is a latent bug, not just a style nit — a rename/typo of the method silently degrades to "accept everything" (the filter becomes a no-op), and neither the type checker nor the contract would catch it. Given the filter exists partly for hot-path cost, a silent no-op has real consequences.
• Suggestion: Declare accepts_scheduler_event on the Protocol (or a FilteringSchedulerAdmissionEventSink sub-Protocol checked via isinstance) so the filter is part of the contract instead of duck-typed.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:66-70 — Swallowed exceptions make telemetry bugs undiagnosable (error handling)

• What: The "never break create" policy is applied via bare swallows — e.g. _MetricLogHandler.emit does except Exception: pass with no logging at all — so a genuine bug inside the instrumentation is completely invisible in the field.
• Why: STYLEGUIDE flags swallowed exceptions as an error-handling problem. The intent (don't break create) is right, but "silently drop with zero trace" means a broken metric would never be noticed or reproducible.
• Suggestion: Keep the swallow, but log at debug with exc_info=True instead of bare pass, and optionally gate a strict re-raise behind an env var (e.g. DATA_DESIGNER_OTEL_STRICT=1) for tests/field debugging.

packages/data-designer-config/src/data_designer/config/run_config.py:194 — otel_metrics_port=None reads like a kill switch but isn't

• What: None means "don't activate for this invocation," yet a listener started by a prior run keeps serving. The field overloads "enabled" and "which port."
• Why: This is a user-facing footgun with a security angle — someone who sets None specifically to stop metrics being served will still have the prior run's endpoint open. Combined with the default-on Warning above, the mismatch between expectation and behavior could bite.
• Suggestion: At minimum, clarify in the docstring that None disables activation for the invocation only (not a process-wide stop), and document how to actually stop the listener. Splitting into enabled: bool + port: int would remove the overload but is probably more than needed.

Suggestions — Take it or leave it

Four optional polish items (click to expand)

packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py:976 — Defensive getattr on a field that always exists

• What: scheduler_event_sink=getattr(self._resource_provider, "scheduler_event_sink", None). Now that ResourceProvider declares scheduler_event_sink: SchedulerAdmissionEventSink | None = None, the attribute always exists.
• Why: The getattr fallback reads as "this might not be there," which is misleading for a declared Pydantic field.
• Suggestion: Simplify to self._resource_provider.scheduler_event_sink unless there's a compat reason (e.g. externally-constructed providers) — in which case a short comment would help.

packages/data-designer/src/data_designer/interface/data_designer.py:743-757 — The "is telemetry on?" rule is duplicated four times (DRY)

• What: self._open_telemetry if self._run_config.otel_metrics_port is not None else None appears at three call sites (request sink, scheduler sink, request-admission controller), plus the port passed into observe_create.
• Why: The enablement invariant is smeared across the interface; changing it means touching several spots.
• Suggestion: Extract a self._metrics_sink() accessor returning the runtime or None, and use it at each site.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:491-492 — Global singleton + atexit registered as an import side effect

• What: Importing the module constructs _RUNTIME and registers atexit.register(_RUNTIME.shutdown).
• Why: A process-global lifecycle established just by importing the interface is a classic testability smell (the tests already work around it with a fresh-instance fixture). It's cheap today (no OTel imports fire at construction), so low urgency.
• Suggestion: Optionally make get_open_telemetry_runtime() construct lazily on first call and register atexit there once.

packages/data-designer/src/data_designer/integrations/opentelemetry.py:335-345 — Progress gauge blends independent jobs into one process-wide float

• What: _dataset_progress_value is a single number aggregated across all active runs, with the last finished value lingering until the next start.
• Why: For concurrent creates, one gauge that shifts between "this job / blended / stale residual" is semantically muddy on a dashboard. Fine for the common single-job case.
• Suggestion: If overlapping creates become normal, prefer a per-run gauge label so each job reports its own fraction. Defer to a follow-up unless multi-job dashboards are an immediate need (it's a metric-schema change).

What Looks Good

• Failure containment is excellent — _activate_run, _initialize, the log handler, _record_create_duration, and _finish_active_requests all defensively suppress/rollback so telemetry can never break a create. The partial-init cleanup in _initialize (removing the handler and shutting down the provider on failure) is especially careful.
• The P1 deadlock fix is done right — _rebind now returns the old (server, thread) and _stop_server runs outside self._lock in both _activate_run and shutdown, with test_runtime_stops_rebound_listener_outside_runtime_lock locking in the behavior. The RLock choice for the re-entrant progress helpers is appropriate.
• Privacy posture is thorough — _bounded_attribute strips/bounds attribute values, and the metric-only log handler tees severity counts without ever exporting log bodies. The docs section spells out the loopback-only, no-auth, metrics-only contract clearly.
• Test suite is genuinely comprehensive — post-teardown event rejection, orphan/duplicate request reconciliation, idle-listener rebind, conflicting-port reuse, concurrent creates sharing one exporter, and test_disabled_runtime_does_not_import_otel_sdk_in_fresh_process guarding the lazy-import contract. These verify behavior, not plumbing.

Suggested Follow-up Issues

For the larger design items we're intentionally not folding into this PR, let's open tracking issues so they don't get lost (happy to file these — just confirm):

• Unify the three engine observability emission seams — converge direct-sink / global-registry / sink-with-filter into one "producer → fan-out to N sinks" model, including guarding the unguarded emit_request_admission_event call and giving both paths an accepts_* filter. (Strategic; highest-value.)
• Split OpenTelemetryRuntime — extract DatasetProgressTracker (directly unit-testable) and a typed _MetricInstruments container; leave the runtime as orchestration + lifecycle.
• Lazy-construct the runtime singleton — move construction + atexit registration into get_open_telemetry_runtime() to drop the import side effect.
• Per-run progress gauge label — only if overlapping creates become a supported scenario (metric-schema change).

The Warnings above are small enough to fix in this PR rather than defer (the accepts_scheduler_event Protocol declaration, the swallowed-exception logging, and the otel_metrics_port=None docstring), as are the DRY _metrics_sink() accessor and the getattr cleanup.

Verdict

Needs changes — nothing here is a hard blocker on the happy path, but before merge it's worth: (1) getting explicit sign-off on default-on telemetry (or flipping to opt-in), and (2) closing the still-open P2 lease-leak window in model_request_executor. The correlation-propagation question is a "confirm or add a test" item.

On the design side, the low-risk Warnings worth folding into this PR are the hot-path/preview build-then-drop (accepts_request_event gate), the Any typing (TYPE_CHECKING), the phantom-protocol method (declare accepts_scheduler_event), the swallowed-exception logging, and the otel_metrics_port=None docstring — plus the small _metrics_sink()/getattr cleanups. The god-object split, lazy singleton, per-run progress label, and especially the three inconsistent emission seams are the larger items better tracked as follow-up issues than jammed into this PR.

ruff check and ruff format --check pass clean on all changed source files.

---

This review was generated by an AI assistant.

[andreatgretel]

One inline question.

[eric-tramel]

Implemented the requested review follow-ups in 8b63e675 and merged the latest main into the branch.

What changed:

• Moved sync and async model_request_started delivery inside the lease cleanup scope, with regressions proving KeyboardInterrupt cannot leak a lease.
• Added optional request-event interest filtering in both the model executor and admission controller, so inactive/preview OTel paths skip event delivery and attempt snapshot construction.
• Strengthened runtime-correlation coverage across the outer asyncio.to_thread hop and the sync-to-async engine bridge.
• Preserved scheduler completion health diagnostics and report early shutdown as outcome="early_shutdown" rather than success.
• Bounded exported model labels to 128 characters while preserving valid / separators.
• Replaced the defensive scheduler-sink getattr with the declared ResourceProvider field.

Validation:

• make check-all passed.
• make test passed: 3,907 passed, 1 skipped.
• Focused scheduler/request/OTel regression suite: 302 passed.
• Seven-persona delegated review completed; targeted follow-ups report no remaining actionable findings.
• GitHub checks are complete: 60 passed, 2 expected skips, 0 failures.
• Greptile re-reviewed 8b63e675 at 5/5 and reports “Safe to merge” with no new findings.

The PR is conflict-free and mergeable. Please re-review/clear the prior changes-requested state when convenient.