refactor: interface-layer OpenTelemetry runtime follow-ups deferred from #802

Priority Level

Low

Task Summary

PR #802 added the default pull-based OpenTelemetry/Prometheus metrics exporter (data_designer.integrations.opentelemetry.OpenTelemetryRuntime). During review a set of non-blocking polish/design items were intentionally deferred rather than expanded that PR's scope. This issue tracks them so they aren't lost.

None of these are correctness blockers — the PR shipped as "Ship it (with nits)." They are hygiene, typing, perf, and design-consolidation items localized to the interface-layer OTel runtime (plus two small engine-seam items).

Line references are against head 9aee1dde of #802.

Technical Details & Implementation Plan

Small / low-risk

• Document the deliberate log-handler swallow. _MetricLogHandler.emit (opentelemetry.py:69-73) is a silent except Exception: pass. This is defensible (logging from inside a failing log handler risks recursion), but it's now the lone silent swallow in the module. Add a short comment explaining the silence is intentional so nobody "fixes" it into re-entrant logging.
• Tighten Any-typed instrument attributes. ~11 runtime attributes (opentelemetry.py:87-99: _registry, _meter_provider, _server, the instruments, etc.) are typed Any because OTel types aren't imported at module scope. Import them under TYPE_CHECKING and annotate precisely (Histogram, Counter, UpDownCounter, ObservableGauge, MeterProvider, CollectorRegistry, HTTPServer).
• DRY the enablement rule. self._open_telemetry if self._run_config.otel_metrics_port is not None else None is duplicated at three call sites in data_designer.py (:753,754,762) plus the port passed into observe_create. Extract a self._metrics_sink() accessor.

Protocol / contract

• Declare the accepts_* filters on the sink Protocols. scheduler_event_sink_accepts / request_event_sink_accepts probe getattr(sink, "accepts_*_event", None) (observability.py:276,324) for methods that aren't declared on SchedulerAdmissionEventSink / RequestAdmissionEventSink. A rename/typo silently degrades a filter to "accept everything." Also, only SchedulerAdmissionEventSink is @runtime_checkable (:217) — RequestAdmissionEventSink (:313) is not. Declare accepts_* on the Protocols (or a Filtering* sub-Protocol) and apply @runtime_checkable consistently.

Perf

• Lock-free interest check. accepts_scheduler_event / accepts_request_event (opentelemetry.py:145-151) acquire the process-wide RLock on every event, on the hottest paths (the scheduler emits many events per task). Since the check only reads bool(self._active_run_ids) plus a static membership test, consider a lock-free "any run active" flag (updated under the lock, read without it) so the cheap gate stays lock- and allocation-free. Measure first.

Metric semantics

• Progress gauge starts at 0 on resume. data_designer.dataset.progress uses the full job total as the denominator (_start_dataset_progress, opentelemetry.py:422-427; row_group_total_rows from async_scheduler.py:326,1091) but the numerator only accrues checkpoints observed in the current run. A resumed (already-partway) job reads near 0 climbing up. Seed initial_completed_records into the starting (scheduled, processed) tuple, or document the limitation in the metric description.

Larger / optional design consolidation

• Split OpenTelemetryRuntime. Extract the dataset-progress math (_start_/_advance_/_finish_active_dataset, _active_dataset_progress) into a small OTel-agnostic, directly unit-testable DatasetProgressTracker, and consider a typed _MetricInstruments container; leave the runtime as orchestration + lifecycle.
• Lazy-construct the singleton. Module import constructs _RUNTIME and registers atexit as a side effect (opentelemetry.py:520-521). Consider constructing lazily on first get_open_telemetry_runtime() call and registering atexit there once.
• Per-run progress-gauge label. _dataset_progress_value is a single process-wide float blended across concurrent creates, with the last finished value lingering. If overlapping creates become a supported dashboard scenario, add a per-run gauge label. (Metric-schema change — only if needed.)

Investigation / Context

• Deferred from PR #802 (feat: add OpenTelemetry metrics for create jobs).
• Related: #447 (introduce AsyncRunController and explicit build/preview sinks) covers the adjacent scheduler-side "converge emission seams into one fan-out" direction — #802 already added _FanoutSchedulerEventSink as a first step. The items here are scoped to the interface-layer OTel exporter, which #447 and the #645 observability epic do not cover.

Dependencies

None. Independent polish; can be picked up piecemeal.

Choose one independently scoped item, then read data_designer/integrations/opentelemetry.py and the referenced sections of data_designer.py, observability.py, or async_scheduler.py. Confirm the existing runtime and protocol behavior first; done means the selected typing, contract, performance, metric-semantics, or lifecycle change is implemented without expanding into the optional design items.