QFw: attach the provider's own phase times to the job's execute span

backends/qfw_telemetry/__init__.py

@@ -58,6 +58,7 @@
 TELEMETRY_DIR_ENV = "QFW_TELEMETRY_DIR"
 TELEMETRY_SAMPLE_ENV = "QFW_TELEMETRY_SAMPLE"
 TELEMETRY_TRANSPORT_ENV = "QFW_TELEMETRY_TRANSPORT"
+TELEMETRY_VENDOR_ENV = "QFW_TELEMETRY_VENDOR"
 TELEMETRY_ENDPOINT_ENV = "QFW_TELEMETRY_ENDPOINT"
 TELEMETRY_LOGS_ENV = "QFW_TELEMETRY_LOGS"
 
@@ -145,6 +146,7 @@ def __init__(self):
 		self.counters = {}
 		self.streams = []
 		self.transport_spans = False
+		self.vendor_timing = False
 		self.defw_hooks = False
 		self.logger_provider = None
 		self.log_handler = None
@@ -566,6 +568,8 @@ def configure(service_name, service_version=None, role=None, attributes=None):
 
 		_STATE.transport_spans = _env(
 			TELEMETRY_TRANSPORT_ENV, "0").lower() in ("1", "yes", "true", "on")
+		_STATE.vendor_timing = _env(
+			TELEMETRY_VENDOR_ENV, "0").lower() in ("1", "yes", "true", "on")
 
 		profile = _profile()
 		if profile != PROFILE_OFF and not OTEL_AVAILABLE:
@@ -584,6 +588,7 @@ def configure(service_name, service_version=None, role=None, attributes=None):
 			# receive it. Clearing this keeps guarded call sites at the
 			# cost of a boolean test.
 			_STATE.transport_spans = False
+			_STATE.vendor_timing = False
 			return _STATE.profile
 
 		from opentelemetry.sdk.trace import TracerProvider
@@ -650,6 +655,16 @@ def enabled():
 	return _STATE.profile != PROFILE_OFF
 
 
+def vendor_timing_enabled():
+	"""
+	True when a provider's own phase times should be attached to the job.
+
+	Flag-guarded like the transport spans: on the QRMI path the times cost
+	one more request to the provider per job.
+	"""
+	return _STATE.vendor_timing
+
+
 def transport_spans_enabled():
 	"""
 	True when qfw.transport.rpc spans should be emitted.
@@ -756,6 +771,8 @@ def use_providers(tracer_provider, meter_provider=None, profile=PROFILE_FILE,
 	with _LOCK:
 		_STATE.transport_spans = _env(
 			TELEMETRY_TRANSPORT_ENV, "0").lower() in ("1", "yes", "true", "on")
+		_STATE.vendor_timing = _env(
+			TELEMETRY_VENDOR_ENV, "0").lower() in ("1", "yes", "true", "on")
 		_STATE.profile = profile
 		_STATE.tracer_provider = tracer_provider
 		_STATE.meter_provider = meter_provider

docs/design/benchmarking.md

@@ -766,6 +766,36 @@ would tax the very quantity this span exists to measure. See
 [Instrumentation Cost](#instrumentation-cost). Never enable it for a Type A
 comparison, where it would perturb the numbers being compared.
 
+### Optional Extension: Vendor Timeline
+
+The provider's own account of a job, attached to the job's execute span and
+to a histogram (October 2026). IQM's server keeps a timeline per job, one
+timestamped event per phase from `created` to `completed`, which is the only
+thing that can say how a device's share of `qfw.backend.execute` divides
+into queueing, compilation and running on the device.
+
+| Signal | Where | What |
+| --- | --- | --- |
+| `qfw.vendor.<phase>_s` | attributes of `qfw.backend.execute` | the seconds the provider spent in each phase: `queue`, `validation`, `compilation`, `execution`, `post_processing`, and `total` from created to completed |
+| `qfw.vendor.duration` | histogram, labelled like the backend metrics plus `qfw.vendor.phase` | the same, aggregated per API path and device for a dashboard's "inside the device" row |
+
+Durations only, never spans. The timestamps come from the provider's clock
+and are compared only with each other; placing spans at them under a span
+from this host's clock would mislead, see [Clocks and
+Precision](#clocks-and-precision).
+
+Where the data comes from, per API path. The native IQM QPM gets the
+timeline structured with the job and already summarizes it. On the QRMI
+path QRMI returns the measurements alone from `task_result`, and prints the
+timeline only as the text of `task_logs`, which the shim reads after a
+completed job (`util/iqm_timeline.py`); that is one more request to the
+provider per job, so the extension is flag-guarded,
+`QFW_TELEMETRY_VENDOR=1`, like the transport spans, and a QRMI change that
+returns the timeline with the result lets the text route go. QDMI's job
+interface exposes counts and a queue position and nothing with a timestamp,
+so the QDMI path carries no vendor timeline, and the histogram's QDMI series
+is absent: a fair statement about the two libraries.
+
 ### Context Attributes
 
 Captured **at run time** as resource attributes (per-process facts) or span
@@ -779,7 +809,7 @@ system looks at report time.
 | Qiskit | `qfw.qiskit.version`; transpiler settings (`qfw.transpile.optimization_level`, seed, basis gates); circuit statistics before/after (`qfw.circuit.num_qubits`, `qfw.circuit.depth`, `qfw.circuit.depth_transpiled`, gate counts, two-qubit gate count); OpenQASM payload size |
 | QRMI | `qfw.qrmi.version`, resource type, `target()` payload digest, acquisition/session identifiers |
 | QDMI / FoMaC | `qfw.device.name`, `qfw.device.version`, qubit count, coupling map, gate set; calibration snapshot (gate fidelities, readout errors, T1/T2 where exposed); `qfw.device.calibration_set_id`, read from the device's `CUSTOM1` property — the key to knowing *which* calibration a result was obtained under |
-| Vendor (e.g. IQM) | Server-side job ID, vendor-reported queue/execution times, vendor calibration identifier |
+| Vendor (e.g. IQM) | Server-side job ID, vendor-reported queue/execution times, vendor calibration identifier. The job id is on the execute span; the phase times are the [vendor timeline extension](#optional-extension-vendor-timeline) |
 | Simulators (NWQ-Sim, TNQVM, …) | Simulator name/version, method/configuration, node and rank count, peak memory where obtainable |
 | SLURM | Job ID, partition, node list, allocated resources (also scrapeable via OpenMetrics in the collector profile) |
 | Environment | Container image tag/digest, Python version, key package versions (qiskit, qrmi, mqt.core, iqm-client, …) |
@@ -833,6 +863,7 @@ the **dimensional** class above:
 | `qfw.backend.duration` | histogram | `qfw.backend.op` = `execute` \| `acquire` \| `submit` \| `collect`, plus api path, device, backend kind and outcome | Back-end cost per stage, and `execute` for the whole interaction so a dashboard needs no sum. Carries the Type A API-path comparison when traces are sampled off |
 | `qfw.transport.duration` | histogram | `qfw.transport.op` = `submit` \| `event` \| `return` | The two legs between the processes and the way back, so a dashboard's per-hop view adds up to the client's end to end. Optional extension, off by default |
 | `qfw.transport.rpc.bytes` | histogram | `qfw.transport.kind` | Transport characterization (libfabric work). Reserved for DEFw's own emission |
+| `qfw.vendor.duration` | histogram | `qfw.vendor.phase` = `queue` \| `validation` \| `compilation` \| `execution` \| `post_processing` \| `total`, plus api path, device, backend kind | The provider's own phase times inside `qfw.backend.execute`. Optional extension, off by default; absent on the QDMI path |
 
 Two conventions hold for this set. Metric names mirror the span they
 aggregate, so moving between a trace view and a dashboard needs no

services/svc_iqm_qpm/util_iqm.py

@@ -164,6 +164,27 @@ def build_timing_summary(record):
 	}
 
 
+# The vendor telemetry's phases, by the names the shim's iqm_timeline uses,
+# from the summary's durations.
+VENDOR_PHASES = (
+	("queue", "queue_wait_received_to_validation_started"),
+	("validation", "validation"),
+	("compilation", "compilation"),
+	("execution", "execution"),
+	("post_processing", "post_processing"),
+	("total", "server_total_created_to_completed"),
+)
+
+
+def vendor_phases(timing_summary):
+	"""The timing summary's durations as the vendor telemetry's phases."""
+	durations = (timing_summary or {}).get("durations_seconds") or {}
+	return {
+		phase: durations[key]
+		for phase, key in VENDOR_PHASES
+		if durations.get(key) is not None}
+
+
 def fetch_iqm_section(name, func):
 	try:
 		return {
@@ -563,6 +584,7 @@ def run_circuit(self, circ):
 			},
 		}
 		timing_summary = build_timing_summary(record)
+		instrumentation.record_vendor_timing(vendor_phases(timing_summary))
 		qhw_result = self.normalize_qhw(
 			"result", raw_payload, include_raw=self._include_raw_results)
 		record["qhw_result"] = qhw_result

services/svc_lib_qpm/drivers/qrmi_driver.py

@@ -1194,6 +1194,21 @@ def _run_ibm_circuit(self, source):
 		from qhw_ibm import normalize_result
 		return normalize_result(raw, device_id=self._device_id())
 
+	def _vendor_timing(self, job_id, credential=None):
+		# The provider's own account of the job for the telemetry, behind
+		# QFW_TELEMETRY_VENDOR because it is one more request per job. IQM
+		# only: QRMI returns the measurements alone from task_result, and the
+		# job's timeline, the server's timestamp per phase, only as the text
+		# of task_logs, which util.iqm_timeline reads. Best effort, like the logs.
+		if not instrumentation.vendor_enabled() or self._provider() != "iqm":
+			return {}
+		from util.iqm_timeline import durations_from_task_logs
+		phases = durations_from_task_logs(
+			self._task_logs(job_id, credential=credential))
+		if phases:
+			instrumentation.record_vendor_timing(phases)
+		return phases
+
 	def _task_logs(self, job_id, credential=None):
 		# Best effort: the caller is already failing, so anything that goes
 		# wrong here is logged and dropped rather than raised. A backend that
@@ -1304,6 +1319,7 @@ def _run_sampler_payload(self, payload, source, delete_finished=True,
 		collect.__exit__(None, None, None)
 		timing["result_fetch_seconds"] = time.monotonic() - result_started
 		timing["total_wall_seconds"] = time.monotonic() - start
+		vendor = self._vendor_timing(job_id, credential=credential)
 
 		# Nothing else to do with the completed task, we can delete it.
 		# The same behavior is implemented in QRMI's task_runner.
@@ -1312,6 +1328,8 @@ def _run_sampler_payload(self, payload, source, delete_finished=True,
 
 		job = {"id": str(job_id), "status": "completed", "cid": cid,
 			   "timing": timing, "shots": shots}
+		if vendor:
+			job["vendor_timing"] = vendor
 		self._jobs.put(job)
 
 		return result_json, job

services/util/instrumentation.py

@@ -66,6 +66,7 @@
 METRIC_APP_JOB_COUNT = "qfw.app.job.count"
 METRIC_QPM_DURATION = "qfw.qpm.duration"
 METRIC_BACKEND_DURATION = "qfw.backend.duration"
+METRIC_VENDOR_DURATION = "qfw.vendor.duration"
 METRIC_TRANSPORT_DURATION = "qfw.transport.duration"
 
 # Dimensional attributes: bounded value sets, allowed as metric labels.
@@ -90,6 +91,7 @@
 ATTR_PAYLOAD_BYTES = "qfw.circuit.payload_bytes"
 ATTR_VENDOR_JOB_ID = "qfw.vendor.job_id"
 ATTR_VENDOR_QUEUE_POSITION = "qfw.vendor.queue_position"
+ATTR_VENDOR_PHASE = "qfw.vendor.phase"
 ATTR_POLL_COUNT = "qfw.backend.poll_count"
 ATTR_POLL_INTERVAL = "qfw.backend.poll_interval_s"
 
@@ -672,6 +674,45 @@ def describe_payload(span, info):
 # QFW_TELEMETRY_TRANSPORT rather than sampled, because a sampled-out span
 # still runs its call site.
 
+def vendor_enabled():
+	"""True when the provider's own phase times should be recorded."""
+	gate = getattr(_telemetry, "vendor_timing_enabled", None)
+	return gate is not None and gate()
+
+
+def record_vendor_timing(phases, labels=None):
+	"""
+	Attach a provider's own account of a job to the telemetry: the seconds
+	it spent in each of the provider's phases (queue, validation,
+	compilation, execution, post_processing, total), as attributes
+	qfw.vendor.<phase>_s on the current span, which is qfw.backend.execute
+	when a driver calls this, and as points of the qfw.vendor.duration
+	histogram labelled by phase like the backend metrics. Durations only:
+	the provider's clock is not this host's, so no span is placed at its
+	timestamps. phases maps a phase name to seconds; a None is skipped.
+	"""
+	if not vendor_enabled() or not phases:
+		return
+	try:
+		values = {
+			f"qfw.vendor.{name}_s": float(seconds)
+			for name, seconds in phases.items() if seconds is not None}
+		if not values:
+			return
+		span = _trace_api().get_current_span()
+		if span is not None:
+			_set(span, values)
+		base = dict(labels or _EXECUTION_LABELS.get() or {})
+		for name, seconds in phases.items():
+			if seconds is None:
+				continue
+			point = dict(base)
+			point[ATTR_VENDOR_PHASE] = name
+			_record(METRIC_VENDOR_DURATION, float(seconds), point)
+	except Exception as exc:
+		_LOG.debug("vendor timing not recorded: %s", exc)
+
+
 def transport_enabled():
 	"""True when the transport call sites should run at all."""
 	return enabled() and _telemetry.transport_spans_enabled()

services/util/iqm_timeline.py

@@ -0,0 +1,70 @@
+"""
+IQM's job timeline, as QRMI's task_logs prints it, into phase durations.
+
+QRMI's IQM resource returns a job's measurements only. What the server
+recorded about the job, one timestamped event per phase (created,
+received, validation, compilation, execution, post-processing, ready,
+completed), reaches us as text from task_logs, one event per line:
+
+    Timeline   :
+      2026-10-12 18:39:22.123 [station                 ] execution_started
+      2026-10-12 18:39:24.971 [station                 ] execution_ended
+    Messages   : (none)
+
+The native IQM QPM gets the same events structured and summarizes them in
+util_iqm.build_timing_summary; the durations here are the same ones, by the
+same names, so the vendor histogram reads alike on both paths. Timestamps
+are the server's own and are compared only with each other. This lives in
+util, beside the instrumentation, so a test can import it without booting
+a service package.
+"""
+import re
+from datetime import datetime
+
+_EVENT = re.compile(
+	r"^\s*(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}) \[(.*?)\s*\] (\S+)\s*$")
+
+# (phase, from this status, to that status); each status is its first
+# occurrence in the timeline.
+PHASES = (
+	("queue", "received", "validation_started"),
+	("validation", "validation_started", "validation_ended"),
+	("compilation", "compilation_started", "compilation_ended"),
+	("execution", "execution_started", "execution_ended"),
+	("post_processing", "post_processing_started", "post_processing_ended"),
+	("total", "created", "completed"),
+)
+
+
+def parse_task_logs(text):
+	"""The timeline's (time, source, status) events, in order, from the text."""
+	events = []
+	for line in (text or "").splitlines():
+		match = _EVENT.match(line)
+		if not match:
+			continue
+		try:
+			when = datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S.%f")
+		except ValueError:
+			continue
+		events.append((when, match.group(2).strip(), match.group(3)))
+	return events
+
+
+def phase_durations(events):
+	"""Seconds per phase from the events; a phase whose bounds are missing is left out."""
+	first = {}
+	for when, _source, status in events:
+		first.setdefault(status, when)
+	durations = {}
+	for phase, start, end in PHASES:
+		if start in first and end in first:
+			seconds = (first[end] - first[start]).total_seconds()
+			if seconds >= 0:
+				durations[phase] = seconds
+	return durations
+
+
+def durations_from_task_logs(text):
+	"""Phase durations straight from task_logs' text; {} when it holds no timeline."""
+	return phase_durations(parse_task_logs(text))

tests/mock/test_instrumentation.py

@@ -424,6 +424,53 @@ def test_transport_spans_cover_the_way_in_and_the_way_back(monkeypatch, recordin
 		assert "qfw.qpm.cid" not in attributes
 
 
+def _vendor_on(monkeypatch):
+	# Like the transport flag: read when providers are adopted, so flip the
+	# state directly.
+	monkeypatch.setattr(qfw_telemetry._STATE, "vendor_timing", True)
+
+
+def test_vendor_timing_is_off_unless_asked_for(recording):
+	circuit = types.SimpleNamespace(
+		info={"qtask_id": 1, "num_qubits": 2, "num_shots": 10},
+		get_cid=lambda: "cid-v", creation_time=1.0, resources_consumed_time=2.0)
+	with instrumentation.backend_execution(circuit, "qrmi", device="q20"):
+		instrumentation.record_vendor_timing({"execution": 2.5, "total": 5.0})
+	(execute,) = recording.spans_by_name()["qfw.backend.execute"]
+	assert "qfw.vendor.execution_s" not in execute.attributes
+	assert recording.points("qfw.vendor.duration") == []
+
+
+def test_vendor_timing_lands_on_the_execute_span_and_its_histogram(
+		monkeypatch, recording):
+	_vendor_on(monkeypatch)
+	circuit = types.SimpleNamespace(
+		info={"qtask_id": 1, "num_qubits": 2, "num_shots": 10},
+		get_cid=lambda: "cid-v", creation_time=1.0, resources_consumed_time=2.0)
+	with instrumentation.backend_execution(
+			circuit, "qrmi", device="q20", backend_kind="iqm"):
+		with instrumentation.backend_phase("collect"):
+			pass
+		# After the phases, as a driver does it: the execute span is current.
+		instrumentation.record_vendor_timing(
+			{"queue": 0.05, "execution": 2.5, "total": 5.0, "validation": None})
+	(execute,) = recording.spans_by_name()["qfw.backend.execute"]
+	assert execute.attributes["qfw.vendor.execution_s"] == 2.5
+	assert execute.attributes["qfw.vendor.total_s"] == 5.0
+	assert "qfw.vendor.validation_s" not in execute.attributes
+
+	points = recording.points("qfw.vendor.duration")
+	by_phase = {a["qfw.vendor.phase"]: (a, p) for a, p in points}
+	assert set(by_phase) == {"queue", "execution", "total"}
+	attributes, point = by_phase["execution"]
+	assert point.count == 1 and point.sum == 2.5
+	# Labelled like the backend metrics, so a dashboard reads it per api path
+	# and device.
+	assert attributes["qfw.stack.api_path"] == "qrmi"
+	assert attributes["qfw.device.name"] == "q20"
+	assert attributes["qfw.backend.kind"] == "iqm"
+
+
 class _RecordingEndpoint:
 	"""Stands in for the client's remote event API, the thing the QPM pushes to."""
 	instances = []