@@ -35,7 +35,10 @@
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanBuilder;
 import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
@@ -51,19 +54,28 @@ class OpenTelemetryTracingTracer implements ApiTracer {
   private final Tracer tracer;
   private final Map<String, Object> attemptAttributes;
   private final String attemptSpanName;
+  private final String operationSpanName;
   private final ApiTracerContext apiTracerContext;
   // Captures the active trace context from the calling thread at RPC initiation.
-  // This allows attempt spans—including retries dispatched on background threads—to
-  // link back to the original parent trace.
-  private final io.opentelemetry.context.Context parentContext;
+  // This allows the operation span and attempt spans to link back to the caller's trace.
+  private final Context parentContext;
+  // Trace context containing the operationSpan, serving as the parent for attempt spans.
+  private final Context operationContext;
+  private @Nullable Span operationSpan;
   private @Nullable Span attemptSpan;
 
   @Override
-  public void injectTraceContext(java.util.Map<String, String> carrier) {
-    if (attemptSpan != null) {
-      io.opentelemetry.context.Context context =
-          io.opentelemetry.context.Context.current().with(attemptSpan);
-      io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator.getInstance()
+  public void injectTraceContext(Map<String, String> carrier) {
+    if (carrier == null) {
+      return;
+    }
+    // Prefer the active attempt span (T4) so outgoing RPC wire context reflects the specific
+    // attempt;
+    // fall back to the overall operation span (T3) if no attempt is currently in-flight.
+    Span currentSpan = attemptSpan != null ? attemptSpan : operationSpan;
+    if (currentSpan != null) {
+      Context context = Context.current().with(currentSpan);
+      W3CTraceContextPropagator.getInstance()
           .inject(
               context,
               carrier,
@@ -75,19 +87,27 @@ public void injectTraceContext(java.util.Map<String, String> carrier) {
     }
   }
 
+  @Override
+  @SuppressWarnings("MustBeClosedChecker")
+  public Scope inScope() {
+    // Attach the active attempt span to the current execution thread context;
+    // fall back to the overall operation span when between attempts.
+    Span currentSpan = attemptSpan != null ? attemptSpan : operationSpan;
+    if (currentSpan == null) {
+      return () -> {};
+    }
+    io.opentelemetry.context.Scope otelScope = currentSpan.makeCurrent();
+    return otelScope::close;
+  }
+
   /**
    * Creates a new instance of {@code OpenTelemetryTracingTracer}.
    *
    * @param tracer the {@link Tracer} to use for recording spans
    * @param apiTracerContext the {@link ApiTracerContext} to use for recording spans
    */
   OpenTelemetryTracingTracer(Tracer tracer, ApiTracerContext apiTracerContext) {
-    this.tracer = tracer;
-    this.apiTracerContext = apiTracerContext;
-    this.attemptSpanName = resolveAttemptSpanName(apiTracerContext);
-    this.attemptAttributes = new HashMap<>();
-    this.parentContext = io.opentelemetry.context.Context.current();
-    buildAttributes();
+    this(tracer, apiTracerContext, resolveAttemptSpanName(apiTracerContext));
   }
 
   /**
@@ -101,14 +121,72 @@ public void injectTraceContext(java.util.Map<String, String> carrier) {
   @InternalApi
   OpenTelemetryTracingTracer(
       Tracer tracer, ApiTracerContext apiTracerContext, String attemptSpanName) {
+    this(tracer, apiTracerContext, attemptSpanName, resolveOperationSpanName(attemptSpanName));
+  }
+
+  /**
+   * Creates a new instance of {@code OpenTelemetryTracingTracer} with explicitly provided attempt
+   * and operation span names.
+   *
+   * @param tracer the {@link Tracer} to use for recording spans
+   * @param apiTracerContext the {@link ApiTracerContext} to use for recording spans
+   * @param attemptSpanName the name of the individual attempt spans
+   * @param operationSpanName the name of the overall client request operation span
+   */
+  @InternalApi
+  OpenTelemetryTracingTracer(
+      Tracer tracer,
+      ApiTracerContext apiTracerContext,
+      String attemptSpanName,
+      String operationSpanName) {
     this.tracer = tracer;
-    this.attemptSpanName = attemptSpanName;
     this.apiTracerContext = apiTracerContext;
+    this.attemptSpanName = attemptSpanName;
+    this.operationSpanName = operationSpanName;
     this.attemptAttributes = new HashMap<>();
-    this.parentContext = io.opentelemetry.context.Context.current();
+    this.parentContext = Context.current();
     buildAttributes();
+    this.operationSpan = startOperationSpan();
+    this.operationContext = parentContext.with(this.operationSpan);
   }
 
+  /**
+   * Starts and initializes the operation-level client request span (T3).
+   *
+   * @return the newly started {@link Span} for the overall operation
+   */
+  private Span startOperationSpan() {
+    SpanBuilder operationSpanBuilder = tracer.spanBuilder(operationSpanName);
+    operationSpanBuilder.setSpanKind(SpanKind.INTERNAL);
+    operationSpanBuilder.setParent(parentContext);
+    operationSpanBuilder.setAllAttributes(
+        ObservabilityUtils.toOtelAttributes(this.attemptAttributes));
+    return operationSpanBuilder.startSpan();
+  }
+
+  /**
+   * Derives the operation-level span name from the attempt span name.
+   *
+   * @param attemptSpanName the attempt span name
+   * @return the operation span name
+   */
+  private static String resolveOperationSpanName(String attemptSpanName) {
+    if (!Strings.isNullOrEmpty(attemptSpanName)) {
+      if (attemptSpanName.endsWith("/attempt")) {
+        String name = attemptSpanName.substring(0, attemptSpanName.length() - "/attempt".length());
+        return name.isEmpty() ? "operation" : name;
+      }
+      return "attempt".equals(attemptSpanName) ? "operation" : attemptSpanName;
+    }
+    return "operation";
+  }
+
+  /**
+   * Resolves the canonical attempt-level span name based on transport and context.
+   *
+   * @param apiTracerContext the tracer context containing transport and method metadata
+   * @return the attempt span name
+   */
   private static String resolveAttemptSpanName(ApiTracerContext apiTracerContext) {
     if (apiTracerContext.transport() == ApiTracerContext.Transport.GRPC) {
       // gRPC Uses the full method name as span name.
@@ -124,12 +202,24 @@ private static String resolveAttemptSpanName(ApiTracerContext apiTracerContext)
     }
   }
 
+  /** Copies attempt-level attributes from the tracer context into the local attribute cache. */
   private void buildAttributes() {
     this.attemptAttributes.putAll(this.apiTracerContext.getAttemptAttributes());
   }
 
   @Override
   public void attemptStarted(Object request, int attemptNumber) {
+    // Prevent creating new attempt spans if the overall operation has already concluded.
+    if (operationSpan == null) {
+      return;
+    }
+    // If a previous attempt was not explicitly closed before a retry started,
+    // end it cleanly so it does not linger.
+    if (attemptSpan != null) {
+      endSpan(attemptSpan, null);
+      attemptSpan = null;
+    }
+
     Map<String, Object> currentAttemptAttributes = new HashMap<>(this.attemptAttributes);
 
     if (attemptNumber > 0) {
@@ -148,8 +238,8 @@ public void attemptStarted(Object request, int attemptNumber) {
     // Attempt spans are of the CLIENT kind
     spanBuilder.setSpanKind(SpanKind.CLIENT);
 
-    // Link attempt span to parent context
-    spanBuilder.setParent(parentContext);
+    // Link attempt span to operation context (parent T3 span)
+    spanBuilder.setParent(operationContext);
 
     // Pass the combined attributes to the new SpanBuilder method
     spanBuilder.setAllAttributes(ObservabilityUtils.toOtelAttributes(currentAttemptAttributes));
@@ -160,33 +250,51 @@ public void attemptStarted(Object request, int attemptNumber) {
   /**
    * Signals that the overall logical operation succeeded.
    *
-   * <p>Closes any remaining in-flight attempt span.
+   * <p>Closes any remaining in-flight attempt span and ends the operation span.
    */
   @Override
   public void operationSucceeded() {
-    recordErrorAndEndAttempt(null);
+    recordErrorAndEndOperation(null);
   }
 
   /**
    * Signals that the overall logical operation was cancelled.
    *
-   * <p>Closes any remaining in-flight attempt span with a {@link CancellationException}.
+   * <p>Closes any remaining in-flight attempt span with a {@link CancellationException} and ends
+   * the operation span with an ERROR status.
    */
   @Override
   public void operationCancelled() {
-    recordErrorAndEndAttempt(new CancellationException());
+    recordErrorAndEndOperation(new CancellationException());
   }
 
   /**
    * Signals that the overall logical operation failed permanently.
    *
-   * <p>Closes any remaining in-flight attempt span with the provided error details.
+   * <p>Closes any remaining in-flight attempt span with the provided error details and ends the
+   * operation span with an ERROR status.
    *
    * @param error the cause of the operation failure
    */
   @Override
   public void operationFailed(Throwable error) {
-    recordErrorAndEndAttempt(error);
+    recordErrorAndEndOperation(error);
+  }
+
+  /**
+   * Records error details and ends both the active attempt span and the operation span.
+   *
+   * @param error the exception associated with the operation failure, or {@code null} if successful
+   */
+  private void recordErrorAndEndOperation(@Nullable Throwable error) {
+    if (attemptSpan != null) {
+      endSpan(attemptSpan, error);
+      attemptSpan = null;
+    }
+    if (operationSpan != null) {
+      endSpan(operationSpan, error);
+      operationSpan = null;
+    }
   }
 
   @Override
@@ -195,7 +303,7 @@ public void attemptSucceeded() {
   }
 
   @Override
-  public void responseHeadersReceived(java.util.Map<String, Object> headers) {
+  public void responseHeadersReceived(Map<String, Object> headers) {
     if (attemptSpan == null) {
       return;
     }
@@ -215,7 +323,7 @@ public void responseHeadersReceived(java.util.Map<String, Object> headers) {
    * @param headers the map of response headers.
    * @return the content length in bytes, or -1 if the header is missing or malformed.
    */
-  private long extractContentLength(java.util.Map<String, Object> headers) {
+  private long extractContentLength(Map<String, Object> headers) {
     try {
       if (headers == null || headers.isEmpty()) return -1;
       // google-http-client HttpHeaders uses a case-insensitive map but we copy it for safety
@@ -265,20 +373,31 @@ private void recordErrorAndEndAttempt(@Nullable Throwable error) {
     if (attemptSpan == null) {
       return;
     }
+    endSpan(attemptSpan, error);
+    attemptSpan = null;
+  }
 
+  /**
+   * Attaches response status attributes and error messages to the span and ends it.
+   *
+   * @param span the span to finish
+   * @param error the exception that caused the span to end, or {@code null} if successful
+   */
+  private void endSpan(Span span, @Nullable Throwable error) {
     Map<String, Object> responseAttributes =
         ObservabilityUtils.getResponseAttributes(error, this.apiTracerContext.transport());
     if (!responseAttributes.isEmpty()) {
-      attemptSpan.setAllAttributes(ObservabilityUtils.toOtelAttributes(responseAttributes));
+      span.setAllAttributes(ObservabilityUtils.toOtelAttributes(responseAttributes));
     }
 
-    if (error != null && !Strings.isNullOrEmpty(error.getMessage())) {
-      attemptSpan.setAttribute(
-          ObservabilityAttributes.STATUS_MESSAGE_ATTRIBUTE, error.getMessage());
+    if (error != null) {
+      span.setStatus(StatusCode.ERROR);
+      if (!Strings.isNullOrEmpty(error.getMessage())) {
+        span.setAttribute(ObservabilityAttributes.STATUS_MESSAGE_ATTRIBUTE, error.getMessage());
+      }
     }
 
-    attemptSpan.end();
-    attemptSpan = null;
+    span.end();
   }
 
   @Override
 
@@ -229,7 +229,7 @@ void testNewTracer_withContext_grpc_usesFullMethodName() {
 
     tracerInstance.attemptStarted(null, 1);
 
-    verify(tracer).spanBuilder("google.cloud.v1.Service/Method");
+    verify(tracer, atLeastOnce()).spanBuilder("google.cloud.v1.Service/Method");
   }
 
   @ParameterizedTest
@@ -256,7 +256,7 @@ void testNewTracer_withContext_http_usesHttpMethodAndPathTemplate(
 
     tracerInstance.attemptStarted(null, 1);
 
-    verify(tracer).spanBuilder(expectedSpanName);
+    verify(tracer, atLeastOnce()).spanBuilder(expectedSpanName);
   }
 
   @Test
@@ -274,7 +274,7 @@ void testNewTracer_withContext_http_noHttpMethodOrPathTemplate_usesFullMethodNam
 
     tracerInstance.attemptStarted(null, 1);
 
-    verify(tracer).spanBuilder("google.cloud.v1.Service.Method");
+    verify(tracer, atLeastOnce()).spanBuilder("google.cloud.v1.Service.Method");
   }
 
   @Test
@@ -310,7 +310,7 @@ void testNewTracer_mergesFactoryContext() {
     tracerInstance.attemptStarted(null, 1);
 
     ArgumentCaptor<Attributes> attributesCaptor = ArgumentCaptor.forClass(Attributes.class);
-    verify(spanBuilder).setAllAttributes(attributesCaptor.capture());
+    verify(spanBuilder, atLeastOnce()).setAllAttributes(attributesCaptor.capture());
 
     Attributes attributes = attributesCaptor.getValue();
     assertThat(attributes.asMap())