diff --git a/amber/src/main/scala/org/apache/texera/web/ComputingUnitMaster.scala b/amber/src/main/scala/org/apache/texera/web/ComputingUnitMaster.scala
index b5bdcbf5573..d85f127b740 100644
--- a/amber/src/main/scala/org/apache/texera/web/ComputingUnitMaster.scala
+++ b/amber/src/main/scala/org/apache/texera/web/ComputingUnitMaster.scala
@@ -139,6 +139,7 @@ class ComputingUnitMaster extends io.dropwizard.Application[Configuration] with
 
   override def run(configuration: Configuration, environment: Environment): Unit = {
     org.apache.texera.observability.OtelInit.init("computing-unit-master")
+    org.apache.texera.web.observability.WorkflowMetricsRecorder.init()
     ObjectMapperUtils.warmupObjectMapperForOperatorsSerde()
 
     SqlServer.initConnection(
diff --git a/amber/src/main/scala/org/apache/texera/web/observability/WorkflowMetricsRecorder.scala b/amber/src/main/scala/org/apache/texera/web/observability/WorkflowMetricsRecorder.scala
new file mode 100644
index 00000000000..0f9ab2401aa
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/observability/WorkflowMetricsRecorder.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.observability
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.core.virtualidentity.ExecutionIdentity
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.{
+  COMPLETED,
+  FAILED,
+  KILLED,
+  PAUSED,
+  PAUSING,
+  RESUMING,
+  RUNNING
+}
+import org.apache.texera.observability.WorkflowMetrics
+import org.apache.texera.observability.WorkflowMetrics.WorkflowKind
+import org.apache.texera.web.service.WorkflowService
+
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+  * Drives [[WorkflowMetrics]] from amber's execution lifecycle. The metric
+  * instruments live in common/config; this object is the single place
+  * that records them, so the lifecycle code only needs one-line calls.
+  *
+  *  - start / terminal counters and the duration histogram are recorded
+  *    from [[onStart]] and [[onStateChange]];
+  *  - the always-polled `texera.workflow.active` gauge is sourced from the
+  *    live WorkflowService registry via the supplier registered in [[init]].
+  */
+object WorkflowMetricsRecorder extends LazyLogging {
+
+  // Start time + kind per in-flight run, so a terminal transition can emit
+  // a duration and attribute the outcome. Keyed by the execution identity.
+  private val inFlight = new ConcurrentHashMap[ExecutionIdentity, (Long, WorkflowKind)]()
+
+  private val ActiveStates: Set[WorkflowAggregatedState] = Set(RUNNING, PAUSING, PAUSED, RESUMING)
+  private val TerminalStates: Set[WorkflowAggregatedState] = Set(COMPLETED, FAILED, KILLED)
+
+  /** Register the active-executions gauge supplier and bind the instruments.
+    *  Call once at startup, after OtelInit.init. The supplier is polled on
+    *  every metric collection, so it must never throw.
+    */
+  def init(): Unit = {
+    WorkflowMetrics.setActiveExecutionsSupplier(() =>
+      try {
+        WorkflowService.getAllWorkflowServices.iterator
+          .flatMap(s => Option(s.executionService.getValue))
+          .map(_.executionStateStore.metadataStore.getState.state)
+          .count(ActiveStates.contains)
+          .toLong
+      } catch {
+        case _: Throwable => 0L
+      }
+    )
+    WorkflowMetrics.ensureBound()
+  }
+
+  /** Record that a run started. */
+  def onStart(
+      executionId: ExecutionIdentity,
+      kind: WorkflowKind = WorkflowKind.Interactive
+  ): Unit = {
+    inFlight.put(executionId, (System.currentTimeMillis(), kind))
+    WorkflowMetrics.recordStart(kind)
+  }
+
+  /** Record terminal counters + duration exactly once, on the first
+    *  transition from a non-terminal into a terminal state. Safe to call on
+    *  every state change; non-terminal and repeat-terminal calls are no-ops.
+    */
+  def onStateChange(
+      executionId: ExecutionIdentity,
+      oldState: WorkflowAggregatedState,
+      newState: WorkflowAggregatedState
+  ): Unit = {
+    if (!TerminalStates.contains(newState) || TerminalStates.contains(oldState)) return
+    val entry = Option(inFlight.remove(executionId))
+    val kind = entry.map(_._2).getOrElse(WorkflowKind.Interactive)
+    val durationSec = entry.map(e => (System.currentTimeMillis() - e._1) / 1000.0).getOrElse(0.0)
+    newState match {
+      case COMPLETED => WorkflowMetrics.recordCompletion(kind, durationSec)
+      case FAILED    => WorkflowMetrics.recordFailure(kind, durationSec)
+      case KILLED    => WorkflowMetrics.recordCancellation(kind)
+      case _         => ()
+    }
+  }
+}
diff --git a/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala b/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala
index 4c8eb209b28..1c921fcacbc 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala
@@ -108,7 +108,12 @@ class WorkflowExecutionService(
   var executionRuntimeService: ExecutionRuntimeService = _
   var executionConsoleService: ExecutionConsoleService = _
 
-  def executeWorkflow(): Unit = {
+  // Returns Some(err) when setup fails synchronously (e.g. compilation) after the
+  // failure has already been surfaced to the UI via errorHandler, so the caller can
+  // mark its still-open setup span; None once the run has started, after which
+  // runtime failures arrive asynchronously through errorHandler and belong to later
+  // execution work rather than this setup.
+  def executeWorkflow(): Option[Throwable] = {
     try {
       val compilationResult = new WorkflowCompiler(workflowContext)
         .compile(request.logicalPlan, CompilationErrorHandling.Strict)
@@ -118,7 +123,7 @@ class WorkflowExecutionService(
         // stop here: `workflow` is still null, so falling through would NPE
         // below and mask the reported compilation error
         errorHandler(err)
-        return
+        return Some(err)
     }
 
     client = ComputingUnitMaster.createAmberRuntime(
@@ -172,6 +177,7 @@ class WorkflowExecutionService(
           }
         )
       )
+    None
   }
 
   override def unsubscribeAll(): Unit = {
diff --git a/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala b/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala
index 8ec650ba715..54056c4caea 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala
@@ -21,9 +21,11 @@ package org.apache.texera.web.service
 
 import com.google.protobuf.timestamp.Timestamp
 import com.typesafe.scalalogging.LazyLogging
+import io.opentelemetry.api.trace.StatusCode
 import io.reactivex.rxjava3.disposables.{CompositeDisposable, Disposable}
 import io.reactivex.rxjava3.subjects.BehaviorSubject
 import org.apache.texera.common.config.{ApplicationConfig, StorageConfig}
+import org.apache.texera.observability.{SpanAttrs, TexeraTracer}
 import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE
 import org.apache.texera.amber.core.WorkflowRuntimeException
@@ -229,126 +231,165 @@ class WorkflowService(
       userOpt: Option[User],
       sessionUri: URI
   ): Unit = {
+    val span = TexeraTracer.tracer
+      .spanBuilder("WorkflowService.initExecutionService")
+      .setAttribute(SpanAttrs.WorkflowId, Long.box(workflowId.id))
+      .startSpan()
+    val scope = span.makeCurrent()
+    try {
 
-    val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid, user.getEmail)).unzip
+      val (uidOpt, userEmailOpt) = userOpt.map(user => (user.getUid, user.getEmail)).unzip
 
-    // Validate before touching the execution already in flight: a request that is
-    // going to be refused must not take the running one's subscriptions with it.
-    // uid is NOT NULL in the DB; fail early here rather than letting the insert fail downstream.
-    val uid = uidOpt.getOrElse(
-      throw new IllegalArgumentException(
-        "Cannot start execution: a user id (uid) is required but none was provided."
+      // Validate before touching the execution already in flight: a request that is
+      // going to be refused must not take the running one's subscriptions with it.
+      // uid is NOT NULL in the DB; fail early here rather than letting the insert fail downstream.
+      val uid = uidOpt.getOrElse(
+        throw new IllegalArgumentException(
+          "Cannot start execution: a user id (uid) is required but none was provided."
+        )
       )
-    )
-    val warehouseName = WorkflowService.resolveLakekeeperWarehouseName(req.warehouseId, uid)
+      val warehouseName = WorkflowService.resolveLakekeeperWarehouseName(req.warehouseId, uid)
 
-    if (executionService.hasValue) {
-      executionService.getValue.unsubscribeAll()
-    }
-
-    val workflowContext: WorkflowContext = createWorkflowContext()
-    workflowContext.warehouse = warehouseName
-    var coordinatorConf = CoordinatorConfig.default
-
-    // clean up results from previous run
-    val previousExecutionId =
-      WorkflowExecutionService.getLatestExecutionId(workflowId, req.computingUnitId)
-    previousExecutionId.foreach(eid => {
-      clearExecutionResources(eid)
-    }) // TODO: change this behavior after enabling cache.
-
-    workflowContext.executionId = ExecutionsMetadataPersistService.insertNewExecution(
-      workflowContext.workflowId,
-      uid,
-      req.executionName,
-      convertToJson(req.engineVersion),
-      req.computingUnitId,
-      req.warehouseId
-    )
+      if (executionService.hasValue) {
+        executionService.getValue.unsubscribeAll()
+      }
 
-    if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
-      val writeLocation = ApplicationConfig.faultToleranceLogRootFolder.get.resolve(
-        s"${workflowContext.workflowId}/${workflowContext.executionId}/"
+      val workflowContext: WorkflowContext = createWorkflowContext()
+      workflowContext.warehouse = warehouseName
+      var coordinatorConf = CoordinatorConfig.default
+
+      // clean up results from previous run
+      val previousExecutionId =
+        WorkflowExecutionService.getLatestExecutionId(workflowId, req.computingUnitId)
+      previousExecutionId.foreach(eid => {
+        clearExecutionResources(eid)
+      }) // TODO: change this behavior after enabling cache.
+
+      workflowContext.executionId = ExecutionsMetadataPersistService.insertNewExecution(
+        workflowContext.workflowId,
+        uid,
+        req.executionName,
+        convertToJson(req.engineVersion),
+        req.computingUnitId,
+        req.warehouseId
       )
-      ExecutionsMetadataPersistService.tryUpdateExistingExecution(workflowContext.executionId) {
-        execution => execution.setLogLocation(writeLocation.toString)
+      span.setAttribute(SpanAttrs.ExecutionId, Long.box(workflowContext.executionId.id))
+      // A run has started: record the start counter and stamp its start time.
+      org.apache.texera.web.observability.WorkflowMetricsRecorder
+        .onStart(workflowContext.executionId)
+
+      if (ApplicationConfig.faultToleranceLogRootFolder.isDefined) {
+        val writeLocation = ApplicationConfig.faultToleranceLogRootFolder.get.resolve(
+          s"${workflowContext.workflowId}/${workflowContext.executionId}/"
+        )
+        ExecutionsMetadataPersistService.tryUpdateExistingExecution(workflowContext.executionId) {
+          execution => execution.setLogLocation(writeLocation.toString)
+        }
+        coordinatorConf = coordinatorConf.copy(faultToleranceConfOpt =
+          Some(FaultToleranceConfig(writeTo = writeLocation))
+        )
       }
-      coordinatorConf = coordinatorConf.copy(faultToleranceConfOpt =
-        Some(FaultToleranceConfig(writeTo = writeLocation))
-      )
-    }
-    if (req.replayFromExecution.isDefined) {
-      val replayInfo = req.replayFromExecution.get
-      ExecutionsMetadataPersistService
-        .tryGetExistingExecution(ExecutionIdentity(replayInfo.eid))
-        .foreach { execution =>
-          val readLocation = new URI(execution.getLogLocation)
-          coordinatorConf = coordinatorConf.copy(stateRestoreConfOpt =
-            Some(
-              StateRestoreConfig(
-                readFrom = readLocation,
-                replayDestination = EmbeddedControlMessageIdentity(replayInfo.interaction)
+      if (req.replayFromExecution.isDefined) {
+        val replayInfo = req.replayFromExecution.get
+        ExecutionsMetadataPersistService
+          .tryGetExistingExecution(ExecutionIdentity(replayInfo.eid))
+          .foreach { execution =>
+            val readLocation = new URI(execution.getLogLocation)
+            coordinatorConf = coordinatorConf.copy(stateRestoreConfOpt =
+              Some(
+                StateRestoreConfig(
+                  readFrom = readLocation,
+                  replayDestination = EmbeddedControlMessageIdentity(replayInfo.interaction)
+                )
               )
             )
-          )
-        }
-    }
+          }
+      }
 
-    val executionStateStore = new ExecutionStateStore()
-    // assign execution id to find the execution from DB in case the constructor fails.
-    executionStateStore.metadataStore.updateState(state =>
-      state.withExecutionId(workflowContext.executionId)
-    )
-    val errorHandler: Throwable => Unit = { t =>
-      {
-        val fromActorOpt = t match {
-          case ex: WorkflowRuntimeException =>
-            ex.relatedWorkerId
-          case other =>
-            None
+      val executionStateStore = new ExecutionStateStore()
+      // assign execution id to find the execution from DB in case the constructor fails.
+      executionStateStore.metadataStore.updateState(state =>
+        state.withExecutionId(workflowContext.executionId)
+      )
+      val errorHandler: Throwable => Unit = { t =>
+        {
+          val fromActorOpt = t match {
+            case ex: WorkflowRuntimeException =>
+              ex.relatedWorkerId
+            case other =>
+              None
+          }
+          val (operatorId, workerId) = getOperatorFromActorIdOpt(fromActorOpt)
+          logger.error("error during execution", t)
+          // Do NOT touch `span` here: this handler is passed into
+          // WorkflowExecutionService and invoked asynchronously (runtime,
+          // websocket, startWorkflow callbacks) after initExecutionService has
+          // returned and the setup span has already ended, so recording onto it
+          // would be a silent no-op. The failure is surfaced via the metadata
+          // store below; setup-span errors are recorded in the catch block.
+          executionStateStore.statsStore.updateState(stats =>
+            stats.withEndTimeStamp(System.currentTimeMillis())
+          )
+          executionStateStore.metadataStore.updateState { metadataStore =>
+            updateWorkflowState(FAILED, metadataStore).addFatalErrors(
+              WorkflowFatalError(
+                EXECUTION_FAILURE,
+                Timestamp(Instant.now),
+                t.toString,
+                getStackTraceWithAllCauses(t),
+                operatorId,
+                workerId
+              )
+            )
+          }
         }
-        val (operatorId, workerId) = getOperatorFromActorIdOpt(fromActorOpt)
-        logger.error("error during execution", t)
-        executionStateStore.statsStore.updateState(stats =>
-          stats.withEndTimeStamp(System.currentTimeMillis())
+      }
+      // WorkflowExecutionService construction does no external work and cannot
+      // throw; it registers its error/state diff handler up front. Once published
+      // via `executionService.onNext`, any failure in `executeWorkflow()` is
+      // recorded by `errorHandler` into the metadata store, whose handler emits a
+      // WorkflowErrorEvent that `connectToExecution` forwards.
+      try {
+        val execution = new WorkflowExecutionService(
+          coordinatorConf,
+          workflowContext,
+          resultService,
+          req,
+          executionStateStore,
+          errorHandler,
+          userEmailOpt,
+          sessionUri
         )
-        executionStateStore.metadataStore.updateState { metadataStore =>
-          updateWorkflowState(FAILED, metadataStore).addFatalErrors(
-            WorkflowFatalError(
-              EXECUTION_FAILURE,
-              Timestamp(Instant.now),
-              t.toString,
-              getStackTraceWithAllCauses(t),
-              operatorId,
-              workerId
-            )
-          )
+        lifeCycleManager.registerCleanUpOnStateChange(executionStateStore)
+        executionService.onNext(execution)
+        // A synchronous setup failure (e.g. compilation) is surfaced to the UI by
+        // the run's errorHandler and signalled back here; record it on the still-open
+        // setup span so error-trace searches find it. None means the run started, after
+        // which runtime failures arrive asynchronously through errorHandler and belong
+        // to later execution work, not this setup span.
+        execution.executeWorkflow().foreach { setupErr =>
+          span.recordException(setupErr)
+          span.setStatus(StatusCode.ERROR)
         }
+      } catch {
+        case e: Throwable =>
+          // Setup threw before the run's errorHandler could record it; mark the
+          // open setup span and surface the failure to the UI.
+          span.recordException(e)
+          span.setStatus(StatusCode.ERROR)
+          errorHandler(e)
       }
-    }
-    // WorkflowExecutionService construction does no external work and cannot
-    // throw; it registers its error/state diff handler up front. Once published
-    // via `executionService.onNext`, any failure in `executeWorkflow()` is
-    // recorded by `errorHandler` into the metadata store, whose handler emits a
-    // WorkflowErrorEvent that `connectToExecution` forwards.
-    try {
-      val execution = new WorkflowExecutionService(
-        coordinatorConf,
-        workflowContext,
-        resultService,
-        req,
-        executionStateStore,
-        errorHandler,
-        userEmailOpt,
-        sessionUri
-      )
-      lifeCycleManager.registerCleanUpOnStateChange(executionStateStore)
-      executionService.onNext(execution)
-      execution.executeWorkflow()
+
     } catch {
-      case e: Throwable => errorHandler(e)
+      case t: Throwable =>
+        // Synchronous setup failure (before the run's own errorHandler is wired).
+        span.recordException(t)
+        span.setStatus(StatusCode.ERROR)
+        throw t
+    } finally {
+      scope.close()
+      span.end()
     }
-
   }
 
   def convertToJson(frontendVersion: String): String = {
diff --git a/amber/src/main/scala/org/apache/texera/web/storage/ExecutionStateStore.scala b/amber/src/main/scala/org/apache/texera/web/storage/ExecutionStateStore.scala
index 654acbbefd1..bc97b39d185 100644
--- a/amber/src/main/scala/org/apache/texera/web/storage/ExecutionStateStore.scala
+++ b/amber/src/main/scala/org/apache/texera/web/storage/ExecutionStateStore.scala
@@ -27,6 +27,7 @@ import org.apache.texera.amber.engine.common.executionruntimestate.{
   ExecutionMetadataStore,
   ExecutionStatsStore
 }
+import org.apache.texera.web.observability.WorkflowMetricsRecorder
 import org.apache.texera.web.service.ExecutionsMetadataPersistService
 
 import java.sql.Timestamp
@@ -44,6 +45,9 @@ object ExecutionStateStore {
         execution.setStatus(maptoStatusCode(state))
         execution.setLastUpdateTime(new Timestamp(System.currentTimeMillis()))
     }
+    // Single chokepoint for every state transition: emit lifecycle metrics
+    // once on the first transition into a terminal state.
+    WorkflowMetricsRecorder.onStateChange(metadataStore.executionId, metadataStore.state, state)
     metadataStore.withState(state)
   }
 }
diff --git a/amber/src/test/scala/org/apache/texera/web/observability/WorkflowMetricsRecorderSpec.scala b/amber/src/test/scala/org/apache/texera/web/observability/WorkflowMetricsRecorderSpec.scala
new file mode 100644
index 00000000000..4c45c126c6a
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/observability/WorkflowMetricsRecorderSpec.scala
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.observability
+
+import io.opentelemetry.api.GlobalOpenTelemetry
+import io.opentelemetry.sdk.OpenTelemetrySdk
+import io.opentelemetry.sdk.common.CompletableResultCode
+import io.opentelemetry.sdk.metrics.{InstrumentType, SdkMeterProvider}
+import io.opentelemetry.sdk.metrics.`export`.{MetricExporter, PeriodicMetricReader}
+import io.opentelemetry.sdk.metrics.data.{AggregationTemporality, MetricData}
+import org.apache.texera.amber.core.virtualidentity.ExecutionIdentity
+import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
+import org.apache.texera.observability.WorkflowMetrics.WorkflowKind
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Duration
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicLong
+import scala.jdk.CollectionConverters._
+
+/**
+  * Exercises the execution-state to metric mapping in WorkflowMetricsRecorder
+  * against a real in-memory meter. WorkflowMetrics.bindForTest is package
+  * private to common/observability, so we drive the binding the production way:
+  * install an SDK into GlobalOpenTelemetry and let ensureBound pick it up.
+  */
+class WorkflowMetricsRecorderSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll {
+
+  private val exporter = new CapturingMetricExporter
+  private var provider: SdkMeterProvider = _
+  private val nextId = new AtomicLong(1000)
+  private def newId(): ExecutionIdentity = ExecutionIdentity(nextId.incrementAndGet())
+
+  override def beforeAll(): Unit = {
+    GlobalOpenTelemetry.resetForTest()
+    val reader = PeriodicMetricReader.builder(exporter).setInterval(Duration.ofHours(1)).build()
+    provider = SdkMeterProvider.builder().registerMetricReader(reader).build()
+    GlobalOpenTelemetry.set(OpenTelemetrySdk.builder().setMeterProvider(provider).build())
+  }
+
+  override def afterAll(): Unit = {
+    if (provider != null) provider.close()
+    GlobalOpenTelemetry.resetForTest()
+  }
+
+  // Cumulative total of a counter, summed over all attribute points.
+  private def counter(name: String): Long = {
+    provider.forceFlush().join(5, TimeUnit.SECONDS)
+    exporter.latest.asScala
+      .find(_.getName == name)
+      .map(_.getLongSumData.getPoints.asScala.map(_.getValue.longValue()).sum)
+      .getOrElse(0L)
+  }
+
+  // Cumulative sample count of a histogram, summed over all attribute points.
+  private def histogramCount(name: String): Long = {
+    provider.forceFlush().join(5, TimeUnit.SECONDS)
+    exporter.latest.asScala
+      .find(_.getName == name)
+      .map(_.getHistogramData.getPoints.asScala.map(_.getCount).sum)
+      .getOrElse(0L)
+  }
+
+  "onStart" should "increment texera.workflow.starts" in {
+    val before = counter("texera.workflow.starts")
+    WorkflowMetricsRecorder.onStart(newId(), WorkflowKind.Interactive)
+    counter("texera.workflow.starts") shouldBe before + 1
+  }
+
+  "onStateChange into COMPLETED" should "record a completion and a duration sample" in {
+    val id = newId()
+    WorkflowMetricsRecorder.onStart(id, WorkflowKind.Interactive)
+    val beforeCompletions = counter("texera.workflow.completions")
+    val beforeDurations = histogramCount("texera.workflow.duration")
+    WorkflowMetricsRecorder.onStateChange(
+      id,
+      WorkflowAggregatedState.RUNNING,
+      WorkflowAggregatedState.COMPLETED
+    )
+    counter("texera.workflow.completions") shouldBe beforeCompletions + 1
+    histogramCount("texera.workflow.duration") shouldBe beforeDurations + 1
+  }
+
+  "onStateChange into FAILED" should "record a completion (outcome=failure) and a duration sample" in {
+    val id = newId()
+    WorkflowMetricsRecorder.onStart(id, WorkflowKind.Scheduled)
+    val beforeCompletions = counter("texera.workflow.completions")
+    val beforeDurations = histogramCount("texera.workflow.duration")
+    WorkflowMetricsRecorder.onStateChange(
+      id,
+      WorkflowAggregatedState.RUNNING,
+      WorkflowAggregatedState.FAILED
+    )
+    counter("texera.workflow.completions") shouldBe beforeCompletions + 1
+    histogramCount("texera.workflow.duration") shouldBe beforeDurations + 1
+  }
+
+  "onStateChange into KILLED" should "record a cancellation and no duration sample" in {
+    val id = newId()
+    WorkflowMetricsRecorder.onStart(id, WorkflowKind.Interactive)
+    val beforeCancellations = counter("texera.workflow.cancellations")
+    val beforeDurations = histogramCount("texera.workflow.duration")
+    WorkflowMetricsRecorder.onStateChange(
+      id,
+      WorkflowAggregatedState.RUNNING,
+      WorkflowAggregatedState.KILLED
+    )
+    counter("texera.workflow.cancellations") shouldBe beforeCancellations + 1
+    histogramCount("texera.workflow.duration") shouldBe beforeDurations
+  }
+
+  "onStateChange to a non-terminal state" should "record nothing" in {
+    val id = newId()
+    WorkflowMetricsRecorder.onStart(id, WorkflowKind.Interactive)
+    val before = counter("texera.workflow.completions")
+    WorkflowMetricsRecorder.onStateChange(
+      id,
+      WorkflowAggregatedState.RUNNING,
+      WorkflowAggregatedState.PAUSED
+    )
+    counter("texera.workflow.completions") shouldBe before
+  }
+
+  "onStateChange whose old state is already terminal" should "record nothing (dedup)" in {
+    val id = newId()
+    WorkflowMetricsRecorder.onStart(id, WorkflowKind.Interactive)
+    val before = counter("texera.workflow.completions")
+    WorkflowMetricsRecorder.onStateChange(
+      id,
+      WorkflowAggregatedState.COMPLETED,
+      WorkflowAggregatedState.FAILED
+    )
+    counter("texera.workflow.completions") shouldBe before
+  }
+}
+
+/** Captures the most recent metric export so a test can read cumulative values. */
+private class CapturingMetricExporter extends MetricExporter {
+  @volatile private var captured: java.util.Collection[MetricData] =
+    java.util.Collections.emptyList()
+  def latest: java.util.Collection[MetricData] = captured
+  override def `export`(metrics: java.util.Collection[MetricData]): CompletableResultCode = {
+    captured = new java.util.ArrayList[MetricData](metrics)
+    CompletableResultCode.ofSuccess()
+  }
+  override def flush(): CompletableResultCode = CompletableResultCode.ofSuccess()
+  override def shutdown(): CompletableResultCode = CompletableResultCode.ofSuccess()
+  override def getAggregationTemporality(instrumentType: InstrumentType): AggregationTemporality =
+    AggregationTemporality.CUMULATIVE
+}
diff --git a/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala
index abf2bb039c5..53c63a9f25d 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala
@@ -430,7 +430,12 @@ class WorkflowExecutionServiceSpec
     // null `workflow` on the way in, and throws out of `executeWorkflow` -- which is also what
     // keeps a "unit" test out of the live-runtime half on whatever actor system the shared JVM
     // holds.
-    noException should be thrownBy service.executeWorkflow()
+    var setupFailure: Option[Throwable] = None
+    noException should be thrownBy { setupFailure = service.executeWorkflow() }
+
+    // The compilation failure is returned (not thrown) so the caller can record it on the
+    // still-open setup span, on top of reaching the UI through errorHandler.
+    setupFailure.map(_.getMessage).getOrElse("") should include("No file selected")
 
     errors should have size 1
     errors.head.getMessage should include("No file selected")
diff --git a/common/observability/src/main/scala/org/apache/texera/observability/SpanAttrs.scala b/common/observability/src/main/scala/org/apache/texera/observability/SpanAttrs.scala
new file mode 100644
index 00000000000..c818f69bc16
--- /dev/null
+++ b/common/observability/src/main/scala/org/apache/texera/observability/SpanAttrs.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import io.opentelemetry.api.common.AttributeKey
+
+/**
+  * Standard Texera span-attribute keys plus a free-text sanitizer.
+  *
+  * These are the shared label keys so every callsite tags spans with the
+  * same names. Set them with the standard OTel API at the callsite, e.g.
+  * ``spanBuilder.setAttribute(SpanAttrs.WorkflowId, id)`` or
+  * ``span.setAttribute(SpanAttrs.WorkflowId, id)``. Run any free-text value
+  * through [[sanitizeFreeText]] first to strip CRLF and cap its length.
+  */
+object SpanAttrs {
+
+  /** Maximum length for free-text span attribute values. */
+  val FreeTextMaxLen: Int = 256
+
+  // ---- Standard Texera correlation labels ------------------------------
+
+  val WorkflowId: AttributeKey[java.lang.Long] = AttributeKey.longKey("texera.workflow.id")
+  val ExecutionId: AttributeKey[java.lang.Long] = AttributeKey.longKey("texera.execution.id")
+  val ProjectId: AttributeKey[java.lang.Long] = AttributeKey.longKey("texera.project.id")
+  val UserId: AttributeKey[java.lang.Long] = AttributeKey.longKey("texera.user.id")
+  val OperatorId: AttributeKey[String] = AttributeKey.stringKey("texera.operator.id")
+  val OperatorName: AttributeKey[String] = AttributeKey.stringKey("texera.operator.name")
+  val Outcome: AttributeKey[String] = AttributeKey.stringKey("texera.outcome")
+
+  // ---- Pure helpers (exposed for testing) ------------------------------
+
+  /**
+    * Strip CR/LF and other C0 control characters from a free-text
+    * value, then cap at [[FreeTextMaxLen]]. Returns null for
+    * null/empty input (caller skips the setAttribute call).
+    */
+  def sanitizeFreeText(value: String): String = {
+    if (value == null || value.isEmpty) return null
+    val stripped = value.filter(c => c >= 0x20 && c != 0x7f)
+    if (stripped.isEmpty) null
+    else if (stripped.length <= FreeTextMaxLen) stripped
+    else stripped.substring(0, FreeTextMaxLen)
+  }
+}
diff --git a/common/observability/src/main/scala/org/apache/texera/observability/TexeraTracer.scala b/common/observability/src/main/scala/org/apache/texera/observability/TexeraTracer.scala
new file mode 100644
index 00000000000..38e1c36b620
--- /dev/null
+++ b/common/observability/src/main/scala/org/apache/texera/observability/TexeraTracer.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import io.opentelemetry.api.GlobalOpenTelemetry
+import io.opentelemetry.api.trace.Tracer
+import io.opentelemetry.context.Context
+
+/**
+  * Accessor for the Texera OTel tracer.
+  *
+  * The only thing this adds over calling ``GlobalOpenTelemetry.getTracer``
+  * directly is a single instrumentation scope name (``org.apache.texera``),
+  * so every Texera-produced span shows up under one logical scope in the
+  * backend, separable from anything emitted by transitive libraries.
+  *
+  * Start and end spans with the standard OTel API at the callsite (see the
+  * OpenTelemetry Java demo for the recommended pattern):
+  *
+  * {{{
+  *   val span = TexeraTracer.tracer.spanBuilder("MyClass.myMethod").startSpan()
+  *   val scope = span.makeCurrent()
+  *   try { ... } catch {
+  *     case t: Throwable => span.recordException(t); span.setStatus(ERROR); throw t
+  *   } finally { scope.close(); span.end() }
+  * }}}
+  *
+  * When the SDK is disabled, ``GlobalOpenTelemetry.getTracer`` returns a
+  * no-op tracer, so calling this is safe at any time.
+  */
+object TexeraTracer {
+
+  private val InstrumentationScope = "org.apache.texera"
+
+  def tracer: Tracer = GlobalOpenTelemetry.getTracer(InstrumentationScope)
+
+  /**
+    * Snapshot the current OTel ``Context`` so async callbacks can
+    * re-attach it via ``Context.makeCurrent`` later. Useful at the
+    * Scala to Python boundary where the calling thread is not the
+    * receiving thread.
+    */
+  def currentContext: Context = Context.current()
+}
diff --git a/common/observability/src/main/scala/org/apache/texera/observability/TraceparentValidator.scala b/common/observability/src/main/scala/org/apache/texera/observability/TraceparentValidator.scala
new file mode 100644
index 00000000000..f5ea4d6cfa4
--- /dev/null
+++ b/common/observability/src/main/scala/org/apache/texera/observability/TraceparentValidator.scala
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+/**
+  * Pure validators for W3C Trace Context headers crossing the
+  * Scala↔Python boundary. The single rule: if the inbound bytes do
+  * not match the strict regex, we discard the value and start a fresh
+  * trace. We never echo a rejected value back into a span, log, or
+  * error message.
+  *
+  * Spec reference: https://www.w3.org/TR/trace-context/
+  *
+  * The regexes here intentionally do NOT use any context-sensitive
+  * grouping or backreferences, keeps the validators safe against
+  * pathological inputs (ReDoS) and trivially fast.
+  */
+object TraceparentValidator {
+
+  /** W3C traceparent format: `<version>-<trace-id>-<parent-id>-<flags>`.
+    *  We accept only version `00` (the only published version) with the
+    *  canonical 32-hex / 16-hex / 2-hex layout. All hex lowercase per
+    *  spec, uppercase is invalid.
+    */
+  private val TraceparentPattern =
+    "^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$".r.pattern
+
+  /** Bounded tracestate length. Spec recommends ≤512 chars. We are
+    *  stricter to remove a small DoS surface, an attacker can't send
+    *  a 1 MiB tracestate to balloon downstream context allocations.
+    */
+  val MaxTracestateLength: Int = 512
+
+  /** Validate a traceparent header. Returns the input unchanged on
+    *  success, None on any failure (rejected, caller starts a fresh
+    *  trace). Null and empty are silent failures.
+    */
+  def validateTraceparent(header: String): Option[String] = {
+    if (header == null || header.isEmpty) return None
+    // Trace-id and parent-id must not be all-zero per spec, an
+    // all-zero ID is a sentinel for "no value" and MUST be rejected.
+    if (!TraceparentPattern.matcher(header).matches()) return None
+    val parts = header.split('-')
+    val traceId = parts(1)
+    val spanId = parts(2)
+    if (isAllZero(traceId) || isAllZero(spanId)) return None
+    Some(header)
+  }
+
+  /** Validate a tracestate header. Spec: comma-separated list of
+    *  key=value pairs, ASCII-printable only, total length capped.
+    *  Returns the input unchanged on success, None on rejection.
+    */
+  def validateTracestate(header: String): Option[String] = {
+    if (header == null || header.isEmpty) return None
+    if (header.length > MaxTracestateLength) return None
+    if (!isAsciiPrintable(header)) return None
+    Some(header)
+  }
+
+  private def isAllZero(s: String): Boolean = {
+    var i = 0
+    while (i < s.length) {
+      if (s.charAt(i) != '0') return false
+      i += 1
+    }
+    true
+  }
+
+  private def isAsciiPrintable(s: String): Boolean = {
+    var i = 0
+    while (i < s.length) {
+      val c = s.charAt(i)
+      // Allow printable ASCII range (0x20-0x7E). Tab and CR/LF are
+      // rejected, a tracestate must not span lines.
+      if (c < 0x20 || c > 0x7e) return false
+      i += 1
+    }
+    true
+  }
+}
diff --git a/common/observability/src/main/scala/org/apache/texera/observability/WorkflowMetrics.scala b/common/observability/src/main/scala/org/apache/texera/observability/WorkflowMetrics.scala
new file mode 100644
index 00000000000..84cd2d0e8f5
--- /dev/null
+++ b/common/observability/src/main/scala/org/apache/texera/observability/WorkflowMetrics.scala
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import com.typesafe.scalalogging.LazyLogging
+import io.opentelemetry.api.GlobalOpenTelemetry
+import io.opentelemetry.api.common.{AttributeKey, Attributes}
+import io.opentelemetry.api.metrics.Meter
+
+/**
+  * Strongly-typed facade for the workflow-execution metrics cluster.
+  *
+  * This facade pattern is deliberate here because these instruments are a
+  * small, fixed, correlated set with strict cardinality rules, and it is
+  * worth centralizing that control in one place. It is NOT the default
+  * pattern for metrics in general: most call sites should just call the OTel
+  * meter API directly next to the business logic, e.g.
+  *
+  * {{{
+  *   private val meter = GlobalOpenTelemetry.getMeter("org.apache.texera")
+  *   private val requests = meter.counterBuilder("myfeature.requests").build()
+  *   // in the handler:
+  *   requests.add(1, Attributes.of(AttributeKey.stringKey("route"), route))
+  * }}}
+  *
+  * Only reach for a facade like this one when the metrics are complex or
+  * need centralized, standardized control. Do not copy it by default.
+  *
+  * Cardinality safety here is enforced by the API surface, not by
+  * documentation: there is no public method that accepts an arbitrary
+  * string as a label key or value. The only labels that ever land on
+  * an instrument are the two enums [[Outcome]] and [[WorkflowKind]],
+  * each restricted to a fixed set. ``workflow.id`` / ``execution.id``
+  * are deliberately NOT metric labels: per-execution detail belongs
+  * in traces and logs, joined on ``trace_id`` at query time.
+  *
+  * Histogram bucket bounds are hard-coded constants so they can't be
+  * coerced by request input. The OTel SDK applies its own default
+  * attribute-value-length cap to anything that does slip through.
+  */
+object WorkflowMetrics extends LazyLogging {
+
+  /** Outcome enum, the only mutable label on lifecycle counters. */
+  sealed abstract class Outcome(val name: String)
+  object Outcome {
+    case object Success extends Outcome("success")
+    case object Failure extends Outcome("failure")
+    case object Cancelled extends Outcome("cancelled")
+  }
+
+  /** Workflow kind enum. Distinguishes interactive vs. scheduled
+    *  workflows for the dashboard's basic split, extend deliberately.
+    */
+  sealed abstract class WorkflowKind(val name: String)
+  object WorkflowKind {
+    case object Interactive extends WorkflowKind("interactive")
+    case object Scheduled extends WorkflowKind("scheduled")
+    case object Unknown extends WorkflowKind("unknown")
+  }
+
+  private val OutcomeKey: AttributeKey[String] = AttributeKey.stringKey("texera.outcome")
+  private val WorkflowKindKey: AttributeKey[String] = AttributeKey.stringKey("texera.workflow.kind")
+
+  /** Histogram bucket bounds in seconds. Hard-coded, constants so
+    *  request input can't reshape the histogram. Range covers
+    *  fast (<1s) to long (>1h) workflows.
+    */
+  private val DurationBuckets: java.util.List[java.lang.Double] = {
+    val builder = new java.util.ArrayList[java.lang.Double]()
+    Seq(0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 300.0, 600.0, 1800.0, 3600.0)
+      .foreach(b => builder.add(java.lang.Double.valueOf(b)))
+    builder
+  }
+
+  private val InstrumentationScope = "org.apache.texera"
+
+  // Instruments are lazy + memoised. The first call after SDK init
+  // builds them against the active GlobalOpenTelemetry meter; once
+  // built they hold the meter instance, so a later GlobalOpenTelemetry
+  // reset (in tests) wouldn't be visible here, see [[resetForTest]].
+  @volatile private var _starts: io.opentelemetry.api.metrics.LongCounter = _
+  @volatile private var _completions: io.opentelemetry.api.metrics.LongCounter = _
+  @volatile private var _cancellations: io.opentelemetry.api.metrics.LongCounter = _
+  // `texera.workflow.active` is an OBSERVABLE gauge, not a manual up/down
+  // counter. A manual +1/-1 counter leaks whenever a run starts but its
+  // terminal event never fires (the controller is killed, the process
+  // restarts mid-run, or the +1 and -1 are split across the engine and web
+  // tiers and one side is missed), the gauge then drifts upward forever and
+  // the dashboard shows phantom "active" executions. An observable gauge
+  // instead reports the TRUE in-progress count from the live execution
+  // registry on every collection, so it cannot leak. The count is supplied
+  // by the host process via [[setActiveExecutionsSupplier]].
+  @volatile private var _active: io.opentelemetry.api.metrics.ObservableLongGauge = _
+  @volatile private var _duration: io.opentelemetry.api.metrics.DoubleHistogram = _
+
+  // Supplier of the current in-progress execution count. Defaults to 0 until
+  // the host process registers the real source (so a process that never
+  // registers reports a flat 0 rather than a wrong number). Read by the gauge
+  // callback on every metric collection.
+  @volatile private var activeExecutionsSupplier: () => Long = () => 0L
+
+  /** Register the authoritative source of "currently active executions".
+    *  The supplier is polled on every metric collection, so the gauge always
+    *  reflects ground truth and can never leak. Called once at process
+    *  startup (e.g. by ComputingUnitMaster).
+    */
+  def setActiveExecutionsSupplier(supplier: () => Long): Unit =
+    synchronized {
+      activeExecutionsSupplier = supplier
+      ensureBound()
+    }
+
+  /** Bind instruments to the current global meter. Idempotent, the
+    *  first call wins; later calls are no-ops. Tests can call
+    *  [[bindForTest]] with an explicit Meter, then [[resetForTest]] to
+    *  rebind.
+    */
+  def ensureBound(): Unit =
+    synchronized {
+      if (_starts == null) bind(GlobalOpenTelemetry.getMeter(InstrumentationScope))
+    }
+
+  private[observability] def bindForTest(meter: Meter): Unit =
+    synchronized {
+      bind(meter)
+    }
+
+  private[observability] def resetForTest(): Unit =
+    synchronized {
+      _starts = null
+      _completions = null
+      _cancellations = null
+      // The observable gauge registered a collection callback, close it so the
+      // previous test's meter provider stops being polled after it's discarded.
+      if (_active != null) _active.close()
+      _active = null
+      _duration = null
+    }
+
+  private def bind(meter: Meter): Unit = {
+    _starts = meter
+      .counterBuilder("texera.workflow.starts")
+      .setDescription("Number of workflow executions started.")
+      .build()
+    // Completions carry texera.outcome={success|failure}. Both success
+    // and self-terminating failures land here so the success/failure-rate
+    // queries (non-success ÷ all completions) have a denominator that
+    // means "runs that finished on their own". User-initiated kills are
+    // NOT completions, see _cancellations.
+    _completions = meter
+      .counterBuilder("texera.workflow.completions")
+      .setDescription("Number of workflow executions that ran to completion (success or failure).")
+      .build()
+    // Cancellations (user kills) are tracked separately so they decrement
+    // the active gauge without polluting the success/failure-rate
+    // denominator. Deliberately label-free apart from workflow.kind.
+    _cancellations = meter
+      .counterBuilder("texera.workflow.cancellations")
+      .setDescription("Number of workflow executions cancelled/killed before finishing.")
+      .build()
+    // Observable gauge: the callback runs on each collection and reports the
+    // live in-progress count, so the value cannot leak. No per-execution
+    // labels (cardinality-safe); the dashboard queries sum(texera_workflow_active).
+    _active = meter
+      .gaugeBuilder("texera.workflow.active")
+      .ofLongs()
+      .setDescription(
+        "Number of workflow executions currently in progress (observed from the live registry)."
+      )
+      .buildWithCallback(obs => obs.record(activeExecutionsSupplier()))
+    _duration = meter
+      .histogramBuilder("texera.workflow.duration")
+      .setDescription("End-to-end duration of a workflow execution.")
+      .setUnit("s")
+      .setExplicitBucketBoundariesAdvice(DurationBuckets)
+      .build()
+    logger.info(s"Texera metric instruments bound to meter scope '$InstrumentationScope'")
+  }
+
+  // ---- Public emitters, typed, no untyped escape hatch --------------
+
+  def recordStart(kind: WorkflowKind): Unit = {
+    ensureBound()
+    logger.debug(s"metric: workflow started (kind=${kind.name}), starts +1")
+    _starts.add(1L, Attributes.of(WorkflowKindKey, kind.name))
+    // `active` is no longer mutated here, it is an observable gauge sourced
+    // from the live execution registry (see setActiveExecutionsSupplier).
+  }
+
+  def recordCompletion(kind: WorkflowKind, durationSec: Double): Unit = {
+    ensureBound()
+    logger.debug(
+      f"metric: workflow completed successfully (kind=${kind.name}, ${durationSec}%.1fs), completions +1"
+    )
+    val attrs = Attributes.of(OutcomeKey, Outcome.Success.name, WorkflowKindKey, kind.name)
+    _completions.add(1L, attrs)
+    _duration.record(durationSec, attrs)
+  }
+
+  def recordFailure(kind: WorkflowKind, durationSec: Double): Unit = {
+    ensureBound()
+    logger.debug(
+      f"metric: workflow failed (kind=${kind.name}, ${durationSec}%.1fs), completions +1 (outcome=failure)"
+    )
+    // A failure is a completion with outcome=failure, it shares the
+    // completions counter (and duration histogram) with successes so the
+    // failure-rate query has both numerator and denominator.
+    val attrs = Attributes.of(OutcomeKey, Outcome.Failure.name, WorkflowKindKey, kind.name)
+    _completions.add(1L, attrs)
+    _duration.record(durationSec, attrs)
+  }
+
+  /** A user-initiated kill/cancel. Bumps a dedicated counter. Deliberately
+    *  NOT recorded as a completion: a cancelled run never finished on its own,
+    *  so it must not drag down the success rate. No duration is recorded for
+    *  the same reason, a killed run's wall-clock time is not a real runtime
+    *  and would skew the duration percentiles. (The active gauge is observed
+    *  from the live registry and needs no decrement here.)
+    */
+  def recordCancellation(kind: WorkflowKind): Unit = {
+    ensureBound()
+    logger.debug(s"metric: workflow cancelled (kind=${kind.name}), cancellations +1")
+    val attrs = Attributes.of(WorkflowKindKey, kind.name)
+    _cancellations.add(1L, attrs)
+  }
+}
diff --git a/common/observability/src/test/scala/org/apache/texera/observability/SpanAttrsSpec.scala b/common/observability/src/test/scala/org/apache/texera/observability/SpanAttrsSpec.scala
new file mode 100644
index 00000000000..8fabc602a8e
--- /dev/null
+++ b/common/observability/src/test/scala/org/apache/texera/observability/SpanAttrsSpec.scala
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import io.opentelemetry.sdk.OpenTelemetrySdk
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter
+import io.opentelemetry.sdk.trace.SdkTracerProvider
+import io.opentelemetry.sdk.trace.`export`.SimpleSpanProcessor
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.jdk.CollectionConverters._
+
+class SpanAttrsSpec extends AnyFlatSpec with Matchers {
+
+  // ----- sanitizeFreeText: pure ----------------------------------------
+
+  "sanitizeFreeText" should "return null for null / empty input" in {
+    SpanAttrs.sanitizeFreeText(null) shouldBe null
+    SpanAttrs.sanitizeFreeText("") shouldBe null
+  }
+
+  it should "strip CR/LF and other control characters" in {
+    SpanAttrs.sanitizeFreeText("hello\r\nworld") shouldBe "helloworld"
+    SpanAttrs.sanitizeFreeText("a\u0000b\u0001c\u001fd\u007fe") shouldBe "abcde"
+  }
+
+  it should "preserve printable text including spaces" in {
+    SpanAttrs.sanitizeFreeText("my-op v1 (alpha)") shouldBe "my-op v1 (alpha)"
+  }
+
+  it should "cap at FreeTextMaxLen" in {
+    val long = "x" * (SpanAttrs.FreeTextMaxLen * 4)
+    val out = SpanAttrs.sanitizeFreeText(long)
+    out.length shouldBe SpanAttrs.FreeTextMaxLen
+  }
+
+  it should "return null when stripping reduces the input to empty" in {
+    SpanAttrs.sanitizeFreeText("\r\n\u0000\u0001") shouldBe null
+  }
+
+  // ----- standard label keys land via the plain OTel API ---------------
+
+  "the standard label keys" should "set correctly through the OTel builder API" in {
+    val (exporter, tracer) = newTracer()
+    val sb = tracer.spanBuilder("workflow.execute")
+    sb.setAttribute(SpanAttrs.WorkflowId, java.lang.Long.valueOf(42L))
+    sb.setAttribute(SpanAttrs.ExecutionId, java.lang.Long.valueOf(7L))
+    sb.setAttribute(SpanAttrs.OperatorName, SpanAttrs.sanitizeFreeText("good\r\nFAKE LINE"))
+    sb.startSpan().end()
+
+    val attrs = exporter.getFinishedSpanItems.asScala.head.getAttributes
+    attrs.get(SpanAttrs.WorkflowId) shouldBe 42L
+    attrs.get(SpanAttrs.ExecutionId) shouldBe 7L
+    attrs.get(SpanAttrs.OperatorName) shouldBe "goodFAKE LINE"
+  }
+
+  it should "expose stable key names so callsites share one label vocabulary" in {
+    SpanAttrs.WorkflowId.getKey shouldBe "texera.workflow.id"
+    SpanAttrs.ExecutionId.getKey shouldBe "texera.execution.id"
+  }
+
+  // ----- attribute-count cap is the OTel SDK default (128) -------------
+
+  it should "respect the SDK's per-span attribute cap of 128" in {
+    val (exporter, tracer) = newTracer()
+    val sb = tracer.spanBuilder("many")
+    // Set 200 distinct keys; SDK should drop everything past the
+    // default cap of 128 without crashing the test.
+    (0 until 200).foreach { i =>
+      sb.setAttribute(s"k$i", s"v$i")
+    }
+    sb.startSpan().end()
+
+    val attrs = exporter.getFinishedSpanItems.asScala.head.getAttributes
+    attrs.size should be <= 128
+  }
+
+  private def newTracer(): (InMemorySpanExporter, io.opentelemetry.api.trace.Tracer) = {
+    val exporter = InMemorySpanExporter.create()
+    val tp = SdkTracerProvider
+      .builder()
+      .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+      .build()
+    val sdk = OpenTelemetrySdk.builder().setTracerProvider(tp).build()
+    (exporter, sdk.getTracer("test"))
+  }
+}
diff --git a/common/observability/src/test/scala/org/apache/texera/observability/TexeraTracerSpec.scala b/common/observability/src/test/scala/org/apache/texera/observability/TexeraTracerSpec.scala
new file mode 100644
index 00000000000..77d5b3ab0f9
--- /dev/null
+++ b/common/observability/src/test/scala/org/apache/texera/observability/TexeraTracerSpec.scala
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import io.opentelemetry.context.Context
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class TexeraTracerSpec extends AnyFlatSpec with Matchers {
+
+  "TexeraTracer.tracer" should "return a tracer that can build and start a span" in {
+    val tracer = TexeraTracer.tracer
+    tracer should not be null
+    // Without an installed SDK this is a no-op tracer, but it must still
+    // hand back a usable span rather than null so callsites never NPE.
+    val span = tracer.spanBuilder("texera.tracer.spec").startSpan()
+    span should not be null
+    span.end()
+  }
+
+  "TexeraTracer.tracer" should "be stable across calls (same instrumentation scope)" in {
+    TexeraTracer.tracer should not be null
+    TexeraTracer.tracer should not be null
+  }
+
+  "TexeraTracer.currentContext" should "return the active context" in {
+    TexeraTracer.currentContext shouldBe Context.current()
+  }
+
+  it should "observe a value made current via the OpenTelemetry context" in {
+    val key = io.opentelemetry.context.ContextKey.named[String]("texera.tracer.spec.key")
+    val scope = Context.current().`with`(key, "v").makeCurrent()
+    try {
+      TexeraTracer.currentContext.get(key) shouldBe "v"
+    } finally {
+      scope.close()
+    }
+  }
+}
diff --git a/common/observability/src/test/scala/org/apache/texera/observability/TraceparentValidatorSpec.scala b/common/observability/src/test/scala/org/apache/texera/observability/TraceparentValidatorSpec.scala
new file mode 100644
index 00000000000..6bf74d99b38
--- /dev/null
+++ b/common/observability/src/test/scala/org/apache/texera/observability/TraceparentValidatorSpec.scala
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class TraceparentValidatorSpec extends AnyFlatSpec with Matchers {
+
+  // ----- traceparent: positive paths -----------------------------------
+
+  "validateTraceparent" should "accept a canonical W3C traceparent" in {
+    val good = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+    TraceparentValidator.validateTraceparent(good) shouldBe Some(good)
+  }
+
+  it should "accept sampled=0 flag" in {
+    val good = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-00"
+    TraceparentValidator.validateTraceparent(good) shouldBe Some(good)
+  }
+
+  // ----- traceparent: negative paths -----------------------------------
+
+  it should "reject null and empty input" in {
+    TraceparentValidator.validateTraceparent(null) shouldBe None
+    TraceparentValidator.validateTraceparent("") shouldBe None
+  }
+
+  it should "reject path-traversal-style spoofing attempts" in {
+    TraceparentValidator.validateTraceparent("../../etc/passwd") shouldBe None
+    TraceparentValidator.validateTraceparent(
+      "00-../../../etc/passwd-b7ad6b7169203331-01"
+    ) shouldBe None
+  }
+
+  it should "reject the wrong version byte" in {
+    // Only version 00 is published. 01, ff, etc. are invalid.
+    val bad01 = "01-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+    val badff = "ff-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+    TraceparentValidator.validateTraceparent(bad01) shouldBe None
+    TraceparentValidator.validateTraceparent(badff) shouldBe None
+  }
+
+  it should "reject all-zero trace-id (sentinel for no-value)" in {
+    val zero = "00-00000000000000000000000000000000-b7ad6b7169203331-01"
+    TraceparentValidator.validateTraceparent(zero) shouldBe None
+  }
+
+  it should "reject all-zero span-id (sentinel for no-value)" in {
+    val zero = "00-0af7651916cd43dd8448eb211c80319c-0000000000000000-01"
+    TraceparentValidator.validateTraceparent(zero) shouldBe None
+  }
+
+  it should "reject UPPERCASE hex (spec requires lowercase)" in {
+    val upper = "00-0AF7651916CD43DD8448EB211C80319C-b7ad6b7169203331-01"
+    TraceparentValidator.validateTraceparent(upper) shouldBe None
+  }
+
+  it should "reject non-hex characters in trace-id" in {
+    val bad = "00-XYZ7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+    TraceparentValidator.validateTraceparent(bad) shouldBe None
+  }
+
+  it should "reject wrong-length segments" in {
+    // trace-id one char short
+    val short = "00-af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+    // span-id one char long
+    val long = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b71692033311-01"
+    TraceparentValidator.validateTraceparent(short) shouldBe None
+    TraceparentValidator.validateTraceparent(long) shouldBe None
+  }
+
+  it should "reject embedded CRLF (header injection)" in {
+    val crlf =
+      "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01\r\nX-Forge: yes"
+    TraceparentValidator.validateTraceparent(crlf) shouldBe None
+  }
+
+  // ----- tracestate ----------------------------------------------------
+
+  "validateTracestate" should "accept a canonical tracestate" in {
+    val ts = "vendor1=hello,vendor2=world"
+    TraceparentValidator.validateTracestate(ts) shouldBe Some(ts)
+  }
+
+  it should "reject null and empty" in {
+    TraceparentValidator.validateTracestate(null) shouldBe None
+    TraceparentValidator.validateTracestate("") shouldBe None
+  }
+
+  it should "reject tracestate longer than the cap" in {
+    val oversize = "a=" + ("b" * (TraceparentValidator.MaxTracestateLength + 1))
+    TraceparentValidator.validateTracestate(oversize) shouldBe None
+  }
+
+  it should "reject non-ASCII / control characters" in {
+    TraceparentValidator.validateTracestate("vendor=hello\u00ffworld") shouldBe None
+    TraceparentValidator.validateTracestate("vendor=hello\r\nworld") shouldBe None
+    TraceparentValidator.validateTracestate("vendor=hello\u0000world") shouldBe None
+  }
+
+  it should "accept tracestate at exactly the cap" in {
+    val atCap = "k=" + ("v" * (TraceparentValidator.MaxTracestateLength - 2))
+    atCap.length shouldBe TraceparentValidator.MaxTracestateLength
+    TraceparentValidator.validateTracestate(atCap) shouldBe Some(atCap)
+  }
+}
diff --git a/common/observability/src/test/scala/org/apache/texera/observability/WorkflowMetricsSpec.scala b/common/observability/src/test/scala/org/apache/texera/observability/WorkflowMetricsSpec.scala
new file mode 100644
index 00000000000..17c9c4c74dd
--- /dev/null
+++ b/common/observability/src/test/scala/org/apache/texera/observability/WorkflowMetricsSpec.scala
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.observability
+
+import io.opentelemetry.sdk.metrics.SdkMeterProvider
+import io.opentelemetry.sdk.metrics.data.MetricData
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.jdk.CollectionConverters._
+
+class WorkflowMetricsSpec extends AnyFlatSpec with Matchers with BeforeAndAfterEach {
+
+  private var reader: InMemoryMetricReader = _
+  private var provider: SdkMeterProvider = _
+
+  override def beforeEach(): Unit = {
+    reader = InMemoryMetricReader.create()
+    provider = SdkMeterProvider.builder().registerMetricReader(reader).build()
+    WorkflowMetrics.resetForTest()
+    WorkflowMetrics.bindForTest(provider.get("org.apache.texera"))
+  }
+
+  override def afterEach(): Unit = {
+    WorkflowMetrics.resetForTest()
+    provider.close()
+  }
+
+  private def collectAll(): Map[String, MetricData] = {
+    reader.collectAllMetrics().asScala.map(m => m.getName -> m).toMap
+  }
+
+  // ----- positive: lifecycle emissions ----------------------------------
+
+  "WorkflowMetrics" should "increment workflow.starts on recordStart" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+
+    val metrics = collectAll()
+    metrics.keySet should contain("texera.workflow.starts")
+
+    val starts = metrics("texera.workflow.starts").getLongSumData.getPoints.asScala.head
+    starts.getValue shouldBe 1L
+    // recordStart no longer mutates `active` — that gauge is observed from the
+    // live registry (see "report the active-execution count…" below).
+  }
+
+  it should "report the active-execution count from the registered supplier, never a manual counter" in {
+    // The whole point of the fix: `active` is an observable gauge sourced from
+    // ground truth, so it cannot leak. Drive it with a stub supplier and
+    // confirm the gauge reports exactly what the supplier returns — regardless
+    // of how many starts/completions were recorded.
+    @volatile var live = 3L
+    WorkflowMetrics.setActiveExecutionsSupplier(() => live)
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+
+    val first = collectAll()("texera.workflow.active").getLongGaugeData.getPoints.asScala.head
+    first.getValue shouldBe 3L
+
+    // It tracks the live source on the next collection — a manual +1/-1
+    // counter could never drop like this without an explicit decrement.
+    live = 0L
+    val second = collectAll()("texera.workflow.active").getLongGaugeData.getPoints.asScala.head
+    second.getValue shouldBe 0L
+  }
+
+  it should "record a completion and duration sample on recordCompletion" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+    WorkflowMetrics.recordCompletion(WorkflowMetrics.WorkflowKind.Interactive, durationSec = 2.5)
+
+    val metrics = collectAll()
+    metrics.keySet should contain allOf (
+      "texera.workflow.completions",
+      "texera.workflow.duration"
+    )
+
+    metrics(
+      "texera.workflow.completions"
+    ).getLongSumData.getPoints.asScala.head.getValue shouldBe 1L
+
+    val histogram = metrics("texera.workflow.duration").getHistogramData.getPoints.asScala.head
+    histogram.getCount shouldBe 1L
+    histogram.getSum shouldBe 2.5
+  }
+
+  it should "record a failure as a non-success completion (so failure-rate queries work)" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Scheduled)
+    WorkflowMetrics.recordFailure(WorkflowMetrics.WorkflowKind.Scheduled, durationSec = 12.0)
+
+    val metrics = collectAll()
+    // A failure shares the completions counter with successes — the
+    // failure-rate query divides non-success completions by all
+    // completions, so failures must live here (not in a separate
+    // series the query never reads).
+    val completion = metrics("texera.workflow.completions").getLongSumData.getPoints.asScala.head
+    completion.getValue shouldBe 1L
+    completion.getAttributes.asMap.asScala.map {
+      case (k, v) => k.getKey -> v.toString
+    } should contain(
+      "texera.outcome" -> "failure"
+    )
+    metrics("texera.workflow.duration").getHistogramData.getPoints.asScala.head.getSum shouldBe 12.0
+    // The orphan counter the old wiring used must be gone.
+    metrics.keySet should not contain "texera.workflow.failures"
+  }
+
+  it should "record a cancellation that is not a completion" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+    WorkflowMetrics.recordCancellation(WorkflowMetrics.WorkflowKind.Interactive)
+
+    val metrics = collectAll()
+    metrics(
+      "texera.workflow.cancellations"
+    ).getLongSumData.getPoints.asScala.head.getValue shouldBe 1L
+    // A kill is not a completion and records no duration: it must not
+    // drag down the success rate nor skew the duration percentiles.
+    metrics.keySet should not contain "texera.workflow.completions"
+    metrics.keySet should not contain "texera.workflow.duration"
+  }
+
+  // ----- security: cardinality safety -----------------------------------
+
+  it should "only emit the texera.outcome and texera.workflow.kind labels" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+    WorkflowMetrics.recordCompletion(WorkflowMetrics.WorkflowKind.Interactive, durationSec = 1.0)
+
+    val attrKeys = collectAll().values.flatMap { md =>
+      val pointSet = md.getType.name() match {
+        case "HISTOGRAM"  => md.getHistogramData.getPoints.asScala
+        case "LONG_GAUGE" => md.getLongGaugeData.getPoints.asScala
+        case _            => md.getLongSumData.getPoints.asScala
+      }
+      pointSet.flatMap(_.getAttributes.asMap.keySet.asScala.map(_.getKey))
+    }.toSet
+
+    attrKeys.foreach { key =>
+      Set("texera.outcome", "texera.workflow.kind") should contain(key)
+    }
+    attrKeys should not contain "texera.workflow.id"
+    attrKeys should not contain "texera.execution.id"
+  }
+
+  it should "expose no public API to attach an arbitrary string label" in {
+    // The class has only typed emitters whose attribute set is
+    // hard-coded. This test is intentionally a compile-time check
+    // disguised as a runtime one — if a future contributor adds an
+    // untyped public method like recordStart(attrs: Attributes),
+    // this assertion still passes but the design intent is broken.
+    // Make the intent explicit:
+    val methodNames = classOf[WorkflowMetrics.type].getDeclaredMethods
+      .map(_.getName)
+      .toSet
+    methodNames should contain allOf ("recordStart", "recordCompletion", "recordFailure")
+    methodNames should not contain "recordWithAttributes"
+  }
+
+  // ----- histogram buckets are constants --------------------------------
+
+  it should "use the hard-coded explicit bucket boundaries for duration" in {
+    WorkflowMetrics.recordStart(WorkflowMetrics.WorkflowKind.Interactive)
+    // Hit a few bucket bounds.
+    Seq(0.05, 0.6, 7.0, 65.0, 400.0).foreach { d =>
+      WorkflowMetrics.recordCompletion(WorkflowMetrics.WorkflowKind.Interactive, durationSec = d)
+    }
+
+    val histogram = collectAll()("texera.workflow.duration").getHistogramData
+    val point = histogram.getPoints.asScala.head
+    point.getCount shouldBe 5L
+    // The point exposes the SDK-configured boundaries; we don't
+    // assert exact values here (would couple the test to the impl)
+    // but we do assert there ARE explicit boundaries — anything
+    // empty would mean the .setExplicitBucketBoundariesAdvice call
+    // was lost during a refactor.
+    point.getBoundaries.size should be > 0
+  }
+}