fix(client): preserve interoperable trace context

.config/mise/tasks/semconv/check

@@ -0,0 +1,22 @@
+#!/bin/sh
+#MISE description="Validate the client semantic convention registry and generated constants"
+
+set -eu
+
+root=$(CDPATH='' cd -- "$(dirname -- "$0")/../../../.." && pwd)
+expected="$root/src/KurrentDB.Client/Core/Common/Diagnostics/Generated"
+generated=$(mktemp -d)
+trap 'rm -rf "$generated"' EXIT HUP INT TERM
+
+"$root/.config/mise/tasks/semconv/generate" "$generated"
+diff -ru "$expected" "$generated"
+
+for generated_file in "$expected"/*.g.cs
+do
+	last_byte=$(tail -c 1 "$generated_file" | od -An -t u1 | tr -d '[:space:]')
+	newline_count=$(tail -c 2 "$generated_file" | wc -l | tr -d '[:space:]')
+	if [ "$last_byte" != 10 ] || [ "$newline_count" != 1 ]; then
+		echo "$generated_file must end with exactly one newline." >&2
+		exit 1
+	fi
+done

.config/mise/tasks/semconv/generate

@@ -0,0 +1,46 @@
+#!/bin/sh
+#MISE description="Generate client semantic convention constants"
+
+set -eu
+
+root=$(CDPATH='' cd -- "$(dirname -- "$0")/../../../.." && pwd)
+output=${1:-"$root/src/KurrentDB.Client/Core/Common/Diagnostics/Generated"}
+registry_version=$(sed -n '1p' "$root/otel/semconv/registry-version")
+registry="$root/otel/semconv/registry"
+official_registry="https://github.com/open-telemetry/semantic-conventions@${registry_version}[model]"
+staging=$(mktemp -d)
+trap 'rm -rf "$staging"' EXIT HUP INT TERM
+
+grep -Fqx "    registry_path: $official_registry" "$registry/manifest.yaml"
+
+weaver registry check \
+	--future \
+	--registry "$registry"
+
+weaver registry generate csharp "$staging" \
+	--future \
+	--registry "$registry" \
+	--templates "$root/otel/semconv/templates" \
+	-D custom_attributes=true \
+	-D official_attributes=false
+
+weaver registry generate csharp "$staging" \
+	--future \
+	--registry "$official_registry" \
+	--templates "$root/otel/semconv/templates" \
+	-D custom_attributes=false \
+	-D official_attributes=true
+
+set -- "$staging"/*.g.cs
+[ -e "$1" ]
+mkdir -p "$output"
+for generated_file in "$output"/*.g.cs
+do
+	[ -e "$generated_file" ] || break
+	rm "$generated_file"
+done
+
+for generated_file
+do
+	mv "$generated_file" "$output/"
+done

.github/workflows/dotnet.yml

@@ -9,6 +9,23 @@ permissions:
   contents: read
 
 jobs:
+  semantic-conventions:
+    name: semantic-conventions
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v5
+        with:
+          persist-credentials: false
+      - name: Install semantic convention tooling
+        uses: jdx/mise-action@9e7f7633ff6f6d6048a9418a68d48f288f50eb14 # v4.2.3
+        with:
+          version: 2026.8.2
+          install_args: github:open-telemetry/weaver
+          cache: false
+      - name: Verify semantic conventions
+        run: mise run --skip-tools semconv:check
+
   vulnerability-scan:
     name: scan-vulnerabilities
     runs-on: ubuntu-latest

mise.toml

@@ -0,0 +1,2 @@
+[tools]
+"github:open-telemetry/weaver" = "0.24.2"

otel/semconv/registry-version

@@ -0,0 +1 @@
+v1.43.0

otel/semconv/registry/manifest.yaml

@@ -0,0 +1,6 @@
+name: trogon_eventstore_client
+description: Semantic conventions for TrogonEventStore client telemetry.
+schema_url: https://trogondb.com/schemas/client/0.1.0
+dependencies:
+  - schema_url: https://opentelemetry.io/schemas/1.43.0
+    registry_path: https://github.com/open-telemetry/semantic-conventions@v1.43.0[model]

otel/semconv/registry/trogon/eventstore/client/spans.yaml

@@ -0,0 +1,60 @@
+groups:
+  - id: registry.trogon.eventstore.client.attributes
+    type: attribute_group
+    stability: development
+    brief: Attributes used by TrogonEventStore client spans.
+    attributes:
+      - id: trogon.eventstore.event.type
+        type: string
+        stability: development
+        brief: Event type processed by the client.
+        examples: [order-created]
+  - id: span.trogon.eventstore.client.database.operation
+    type: span
+    stability: development
+    span_kind: client
+    brief: Describes a database operation performed by the client.
+    attributes:
+      - ref: db.system.name
+        requirement_level: required
+      - ref: db.operation.name
+        requirement_level: required
+      - ref: db.collection.name
+        requirement_level: recommended
+      - ref: db.operation.batch.size
+        requirement_level: recommended
+      - ref: error.type
+        requirement_level:
+          conditionally_required: If the operation failed.
+      - ref: server.address
+        requirement_level: recommended
+      - ref: server.port
+        requirement_level:
+          conditionally_required: If the server port is available.
+
+  - id: span.trogon.eventstore.client.process
+    type: span
+    stability: development
+    span_kind: consumer
+    brief: Describes processing an event delivered by a subscription.
+    attributes:
+      - ref: messaging.system
+        requirement_level: required
+      - ref: messaging.operation.name
+        requirement_level: required
+      - ref: messaging.operation.type
+        requirement_level: required
+      - ref: messaging.destination.name
+        requirement_level: required
+      - ref: messaging.message.id
+        requirement_level: recommended
+      - ref: messaging.consumer.group.name
+        requirement_level:
+          conditionally_required: If the event was delivered by a persistent subscription.
+      - ref: trogon.eventstore.event.type
+        requirement_level: recommended
+      - ref: server.address
+        requirement_level: recommended
+      - ref: server.port
+        requirement_level:
+          conditionally_required: If the server port is available.

otel/semconv/templates/registry/csharp/telemetry-attributes.cs.j2

@@ -0,0 +1,11 @@
+// <auto-generated />
+
+namespace KurrentDB.Diagnostics.Telemetry;
+
+static class TelemetryAttributes {
+{% for group in ctx %}
+{% for attribute in group.attributes | sort(attribute="name") %}
+	public const string {{ attribute.name | pascal_case }} = "{{ attribute.name }}";
+{% endfor %}
+{% endfor %}
+}{{- "\n" -}}

otel/semconv/templates/registry/csharp/trogon-telemetry-attributes.cs.j2

@@ -0,0 +1,11 @@
+// <auto-generated />
+
+namespace KurrentDB.Diagnostics.Telemetry;
+
+static class TrogonTelemetryAttributes {
+{% for group in ctx %}
+{% for attribute in group.attributes | sort(attribute="name") %}
+	public const string {{ attribute.name | pascal_case | regex_replace("^TrogonEventstore", "") }} = "{{ attribute.name }}";
+{% endfor %}
+{% endfor %}
+}{{- "\n" -}}

otel/semconv/templates/registry/csharp/weaver.yaml

@@ -0,0 +1,54 @@
+whitespace_control:
+  trim_blocks: true
+  lstrip_blocks: true
+
+params:
+  custom_attributes: false
+  official_attributes: false
+
+templates:
+  - template: telemetry-attributes.cs.j2
+    filter: >
+      if $official_attributes then
+        semconv_grouped_attributes
+        | map({
+            root_namespace: .root_namespace,
+            attributes: [.attributes[] | select(
+              .name == "db.collection.name" or
+              .name == "db.operation.batch.size" or
+              .name == "db.operation.name" or
+              .name == "db.system.name" or
+              .name == "error.type" or
+              .name == "exception.message" or
+              .name == "exception.stacktrace" or
+              .name == "exception.type" or
+              .name == "messaging.consumer.group.name" or
+              .name == "messaging.destination.name" or
+              .name == "messaging.message.id" or
+              .name == "messaging.operation.name" or
+              .name == "messaging.operation.type" or
+              .name == "messaging.system" or
+              .name == "server.address" or
+              .name == "server.port"
+            )]
+          })
+        | map(select(.attributes | length > 0))
+      else
+        empty
+      end
+    application_mode: single
+    file_name: TelemetryAttributes.g.cs
+  - template: trogon-telemetry-attributes.cs.j2
+    filter: >
+      if $custom_attributes then
+        semconv_grouped_attributes
+        | map({
+            root_namespace: .root_namespace,
+            attributes: [.attributes[] | select(.name | startswith("trogon.eventstore."))]
+          })
+        | map(select(.attributes | length > 0))
+      else
+        empty
+      end
+    application_mode: single
+    file_name: TrogonTelemetryAttributes.g.cs

samples/diagnostics/Program.cs

@@ -11,7 +11,7 @@
 /**
 # region import-required-packages
 // required
-dotnet add package EventStore.Client.Extensions.OpenTelemetry
+dotnet add package TrogonEventStore.Client
 
 // recommended
 dotnet add package OpenTelemetry.Exporter.OpenTelemetryProtocol

samples/diagnostics/diagnostics.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <RootNamespace>connecting_to_a_cluster</RootNamespace>
+        <RootNamespace>diagnostics</RootNamespace>
     </PropertyGroup>
 
 	<ItemGroup>

src/KurrentDB.Client/Core/Common/Diagnostics/ActivitySourceExtensions.cs

@@ -4,21 +4,38 @@
 using System.Diagnostics;
 using KurrentDB.Diagnostics;
 using KurrentDB.Diagnostics.Telemetry;
+using OpenTelemetry;
 using static KurrentDB.Diagnostics.Tracing.TracingConstants;
 
 namespace KurrentDB.Client.Diagnostics;
 
 static class ActivitySourceExtensions {
-	public static async ValueTask<T> TraceClientOperation<T>(
+	public static ValueTask<T> TraceClientOperation<T>(
 		this ActivitySource source,
 		Func<ValueTask<T>> tracedOperation,
 		string operationName,
 		ActivityTagsCollection? tags = null
+	) => source.TraceClientOperation(_ => tracedOperation(), operationName, tags);
+
+	public static async ValueTask<T> TraceClientOperation<T>(
+		this ActivitySource source,
+		Func<Activity?, ValueTask<T>> tracedOperation,
+		string operationName,
+		ActivityTagsCollection? tags = null
 	) {
-		using var activity = StartActivity(source, operationName, ActivityKind.Client, tags, Activity.Current?.Context);
+		if (source.HasNoActiveListeners())
+			return await tracedOperation(null).ConfigureAwait(false);
+
+		(tags ??= new ActivityTagsCollection())
+			.WithRequiredTag(TelemetryAttributes.DbSystemName, SystemName)
+			.WithRequiredTag(TelemetryAttributes.DbOperationName, operationName);
+
+		var target = tags.FirstOrDefault(tag => tag.Key == TelemetryAttributes.DbCollectionName).Value as string;
+		var spanName = target is null ? operationName : $"{operationName} {target}";
+		using var activity = StartActivity(source, spanName, ActivityKind.Client, tags, Activity.Current?.Context);
 
 		try {
-			var res = await tracedOperation().ConfigureAwait(false);
+			var res = await tracedOperation(activity).ConfigureAwait(false);
 			activity?.StatusOk();
 			return res;
 		} catch (Exception ex) {
@@ -29,34 +46,44 @@ public static async ValueTask<T> TraceClientOperation<T>(
 
 	public static void TraceSubscriptionEvent(
 		this ActivitySource source,
-		string? subscriptionId,
+		string? consumerGroupName,
 		ResolvedEvent resolvedEvent,
 		ChannelInfo channelInfo,
-		KurrentDBClientSettings settings,
-		UserCredentials? userCredentials
+		KurrentDBClientSettings settings
 	) {
 		if (source.HasNoActiveListeners() || resolvedEvent.Event is null)
 			return;
 
-		var parentContext = resolvedEvent.Event.Metadata.ExtractPropagationContext();
+		var propagationContext = resolvedEvent.Event.Metadata.ExtractPropagationContext();
 
-		if (parentContext == default(ActivityContext)) return;
+		if (propagationContext.ActivityContext == default)
+			return;
 
+		var destination = resolvedEvent.OriginalEvent.EventStreamId;
 		var tags = new ActivityTagsCollection()
-			.WithRequiredTag(TelemetryTags.KurrentDB.Stream, resolvedEvent.OriginalEvent.EventStreamId)
-			.WithOptionalTag(TelemetryTags.KurrentDB.SubscriptionId, subscriptionId)
-			.WithRequiredTag(TelemetryTags.KurrentDB.EventId, resolvedEvent.OriginalEvent.EventId.ToString())
-			.WithRequiredTag(TelemetryTags.KurrentDB.EventType, resolvedEvent.OriginalEvent.EventType)
-			// Ensure consistent server.address attribute when connecting to cluster via dns discovery
+			.WithRequiredTag(TelemetryAttributes.MessagingSystem, SystemName)
+			.WithRequiredTag(TelemetryAttributes.MessagingOperationName, Operations.Process)
+			.WithRequiredTag(TelemetryAttributes.MessagingOperationType, Operations.Process)
+			.WithRequiredTag(TelemetryAttributes.MessagingDestinationName, destination)
+			.WithOptionalTag(TelemetryAttributes.MessagingConsumerGroupName, consumerGroupName)
+			.WithRequiredTag(TelemetryAttributes.MessagingMessageId, resolvedEvent.OriginalEvent.EventId.ToString())
+			.WithRequiredTag(TrogonTelemetryAttributes.EventType, resolvedEvent.OriginalEvent.EventType)
 			.WithGrpcChannelServerTags(channelInfo)
-			.WithClientSettingsServerTags(settings)
-			.WithOptionalTag(
-				TelemetryTags.Database.User,
-				userCredentials?.Username ?? settings.DefaultCredentials?.Username
-			);
-
-		StartActivity(source, Operations.Subscribe, ActivityKind.Consumer, tags, parentContext)
-			?.Dispose();
+			.WithClientSettingsServerTags(settings);
+
+		using var activity = StartActivity(
+			source,
+			$"{Operations.Process} {destination}",
+			ActivityKind.Consumer,
+			tags,
+			propagationContext.ActivityContext
+		);
+
+		if (activity is null)
+			return;
+
+		foreach (var (name, value) in propagationContext.Baggage.GetBaggage())
+			activity.AddBaggage(name, value);
 	}
 
 	static Activity? StartActivity(
@@ -67,10 +94,6 @@ public static void TraceSubscriptionEvent(
 		if (source.HasNoActiveListeners())
 			return null;
 
-		(tags ??= new ActivityTagsCollection())
-			.WithRequiredTag(TelemetryTags.Database.System, KurrentDBClientDiagnostics.InstrumentationName)
-			.WithRequiredTag(TelemetryTags.Database.Operation, operationName);
-
 		return source
 			.CreateActivity(
 				operationName,