diff --git a/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala b/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
index 96dda080..182efeaf 100644
--- a/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
+++ b/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
@@ -66,6 +66,8 @@ case class Method(
     else if (ReservedMethodNames.contains(name)) s"$name$ReservedMethodNameSuffix"
     else name
 
+  val handlerName: String = s"${name}Handler"
+
 }
 
 object Method {
diff --git a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
index e8c6af53..4af17c4d 100644
--- a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
+++ b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
@@ -107,13 +107,19 @@ object @{serviceName}Handler {
           pekko.grpc.scaladsl.ServerReflection.partial(List(@{service.name})))
 }
 
-    private def methodName(request: model.HttpRequest, prefix: String): String =
+    private def methodName(request: model.HttpRequest, prefix: String): String = {
+      // Use Uri.Path structure matching instead of toString to avoid
+      // renderPath + StringBuilder + String allocations per request.
+      // gRPC path format: /{service}/{method}
       request.uri.path match {
-        case model.Uri.Path.Slash(model.Uri.Path.Segment(`prefix`, model.Uri.Path.Slash(model.Uri.Path.Segment(method, model.Uri.Path.Empty)))) =>
-          method
-        case _ =>
-          null
+        case model.Uri.Path.Slash(
+          model.Uri.Path.Segment(serviceName,
+            model.Uri.Path.Slash(
+              model.Uri.Path.Segment(method, model.Uri.Path.Empty))))
+          if serviceName == prefix => method
+        case _ => null
       }
+    }
 
     private def handler(implementation: @serviceName, prefix: String, eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(implicit system: ClassicActorSystemProvider): model.HttpRequest => scala.concurrent.Future[model.HttpResponse] = {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
@@ -122,6 +128,16 @@ object @{serviceName}Handler {
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e)
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
@@ -129,7 +145,7 @@ object @{serviceName}Handler {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -165,6 +181,16 @@ object @{serviceName}Handler {
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e)
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
@@ -172,7 +198,7 @@ object @{serviceName}Handler {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
index 1d8e2f04..e01df6e3 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
@@ -18,7 +18,7 @@ import pekko.NotUsed
 import pekko.annotation.InternalApi
 import pekko.annotation.InternalStableApi
 import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
-import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText }
+import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText, Gzip, Identity }
 import pekko.http.javadsl.{ model => jmodel }
 import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse, Trailer }
 import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
@@ -141,15 +141,85 @@ object GrpcProtocol {
     case _                                       => None
   }
 
+  // Pre-computed negotiation results for common codec combinations
+  private val NativeIdentityIdentity: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)), GrpcProtocolNative.newWriter(Identity))
+
   /**
    * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
    *
    * @param request the client request to respond to.
    * @return the protocol reader for the request, and a protocol writer for the response.
    */
-  def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] =
-    detect(request).map { variant =>
-      (Codecs.detect(request).map(variant.newReader), variant.newWriter(Codecs.negotiate(request)))
+  def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val mediaType = request.entity.getContentType.mediaType
+    val subType = mediaType.subType
+    val isNative = (subType == "grpc+proto") || (subType == "grpc")
+
+    if (isNative) {
+      // Single-pass header scan for native gRPC
+      var requestEncoding: String = null
+      var acceptEncoding: String = null
+      request match {
+        case sReq: pekko.http.scaladsl.model.HttpMessage =>
+          val headers = sReq.headers
+          var i = 0
+          while (i < headers.size) {
+            val h = headers(i)
+            val name = h.lowercaseName
+            if (name == "grpc-encoding") requestEncoding = h.value
+            else if (name == "grpc-accept-encoding") acceptEncoding = h.value
+            i += 1
+          }
+        case _ =>
+          return slowNegotiateOpt(request, subType)
+      }
+      // Determine reader codec (request encoding)
+      val readerCodec: Codec =
+        if (requestEncoding eq null) Identity
+        else if (requestEncoding == "identity") Identity
+        else if (requestEncoding == "gzip") Gzip
+        else return slowNegotiateOpt(request, subType)
+      // Determine writer codec (response encoding)
+      // Respect client's grpc-accept-encoding preference order (matching Codecs.negotiate).
+      // gRPC format: comma-separated, e.g. "gzip,identity" — client order indicates preference.
+      val writerCodec: Codec =
+        if (acceptEncoding eq null) Identity
+        else if (acceptEncoding == "gzip") Gzip
+        else if (acceptEncoding == "identity") Identity
+        else if (acceptEncoding.contains(",")) {
+          // Multiple encodings: iterate client preference order, pick first supported
+          val parts = acceptEncoding.split(",")
+          var codec: Codec = null
+          var i = 0
+          while (i < parts.length && (codec eq null)) {
+            val trimmed = parts(i).trim
+            if (trimmed == "gzip") codec = Gzip
+            else if (trimmed == "identity") codec = Identity
+            i += 1
+          }
+          if (codec eq null) Identity else codec
+        } else return slowNegotiateOpt(request, subType)
+      // Return pre-computed result for common combinations
+      if ((readerCodec eq Identity) && (writerCodec eq Identity)) return Some(NativeIdentityIdentity)
+      return Some((scala.util.Success(GrpcProtocolNative.newReader(readerCodec)),
+        GrpcProtocolNative.newWriter(writerCodec)))
+    }
+
+    // Non-native protocols - slow path
+    slowNegotiateOpt(request, subType)
+  }
+
+  private def slowNegotiateOpt(request: jmodel.HttpRequest, subType: String)
+      : Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val variant = subType match {
+      case "grpc-web" | "grpc-web+proto"           => Some(GrpcProtocolWeb)
+      case "grpc-web-text" | "grpc-web-text+proto" => Some(GrpcProtocolWebText)
+      case "grpc" | "grpc+proto"                   => Some(GrpcProtocolNative)
+      case _                                       => None
     }
+    variant.map(v => (Codecs.detect(request).map(v.newReader), v.newWriter(Codecs.negotiate(request))))
+  }
 
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
index e5392960..cbc2acc0 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
@@ -27,4 +27,12 @@ trait ProtobufSerializer[T] {
 
 private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] {
   private[grpc] def serializeDataFrame(t: T): ByteString
+
+  /**
+   * Deserialize a protobuf message from a ByteString starting at the given offset.
+   * Avoids ByteString.slice allocation by passing the offset directly to the parser.
+   * Default implementation falls back to slice + deserialize.
+   */
+  private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T =
+    deserialize(data.slice(offset, offset + length))
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
index 8cd4653c..2a11b707 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
@@ -100,10 +100,19 @@ object AbstractGrpcProtocol {
       })
       .toContentType
 
+  private val SmallMessageThreshold = 4096
+
   def encodeFrameData(data: ByteString, isCompressed: Boolean, isTrailer: Boolean): ByteString = {
-    val header = new Array[Byte](FrameHeaderSize)
-    writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
-    ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    if (data.length <= SmallMessageThreshold) {
+      val combined = new Array[Byte](FrameHeaderSize + data.length)
+      writeFrameHeader(combined, 0, data.length, isCompressed, isTrailer)
+      data.copyToArray(combined, FrameHeaderSize)
+      ByteString.fromArrayUnsafe(combined)
+    } else {
+      val header = new Array[Byte](FrameHeaderSize)
+      writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
+      ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    }
   }
 
   def writer(
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
index 6dfbd357..49b7f652 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
@@ -23,8 +23,9 @@ import scala.collection.immutable
 import scala.util.{ Failure, Success, Try }
 
 object Codecs {
-  // TODO should this list be made user-extensible?
-  val supportedCodecs = immutable.Seq(Gzip, Identity)
+  // Codecs supported by this server. The selection order is determined by the
+  // client's grpc-accept-encoding header preference, not by this sequence order.
+  val supportedCodecs = immutable.Seq(Identity, Gzip)
   private val supportedByName: Map[String, Codec] = supportedCodecs.map(c => c.name -> c).toMap
 
   private def extractHeaders(request: jm.HttpMessage): Iterable[jm.HttpHeader] = {
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
index c7f13ff7..00fee07b 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
@@ -44,11 +44,32 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") {
     AbstractGrpcProtocol.writer(this, codec, encodeFrame(codec, _), encodeDataToResponse(codec))
 
   override protected def reader(codec: Codec): GrpcProtocolReader =
-    AbstractGrpcProtocol.reader(codec, decodeFrame)
+    if (codec eq Identity)
+      AbstractGrpcProtocol.reader(codec, decodeFrame).copy(decodeSingleFrame = decodeIdentitySingleFrame)
+    else AbstractGrpcProtocol.reader(codec, decodeFrame)
 
   @inline
   private def decodeFrame(@nowarn("msg=is never used") frameType: Int, data: ByteString) = DataFrame(data)
 
+  private def decodeIdentitySingleFrame(frame: ByteString): ByteString = {
+    if (frame.length < AbstractGrpcProtocol.FrameHeaderSize) throw new MissingParameterException
+
+    val frameType = frame(0)
+    val length = frame.readIntBE(1)
+    val available = frame.length - AbstractGrpcProtocol.FrameHeaderSize
+    if (length > available) throw new MissingParameterException
+    if (length < 0) throw new IllegalStateException(s"Invalid frame length: $length")
+    if (length < available)
+      throw new IllegalStateException("Unexpected trailing data")
+    if ((frameType & 0x80) != 0) throw new IllegalStateException("Cannot read unknown frame")
+
+    if ((frameType & 1) != 0)
+      throw new io.grpc.StatusException(
+        io.grpc.Status.INTERNAL.withDescription(
+          "Compressed-Flag bit is set, but a compression encoding is not specified"))
+    frame.drop(AbstractGrpcProtocol.FrameHeaderSize)
+  }
+
   @inline
   private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
     frame match {
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
index 18c49637..be3b3c84 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
@@ -60,7 +60,25 @@ object GrpcMarshalling {
       reader: GrpcProtocolReader): CompletionStage[T] =
     entity match {
       case strict: pekko.http.scaladsl.model.HttpEntity.Strict =>
-        completedOrFailed(u.deserialize(reader.decodeSingleFrame(strict.data)))
+        completedOrFailed {
+          u match {
+            case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
+              if (strict.data.length < AbstractGrpcProtocol.FrameHeaderSize)
+                throw new MissingParameterException
+              val frameType = strict.data(0)
+              if ((frameType & 1) != 0)
+                throw new io.grpc.StatusException(
+                  io.grpc.Status.INTERNAL.withDescription(
+                    "Compressed-Flag bit is set, but a compression encoding is not specified"))
+              val declaredLength = strict.data.readIntBE(1)
+              val available = strict.data.length - AbstractGrpcProtocol.FrameHeaderSize
+              if (declaredLength > available) throw new MissingParameterException
+              if (declaredLength < 0) throw new IllegalStateException(s"Invalid frame length: $declaredLength")
+              frameSerializer.deserialize(strict.data, AbstractGrpcProtocol.FrameHeaderSize, declaredLength)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(strict.data))
+          }
+        }
       case _ =>
         unmarshal(entity.getDataBytes, u, mat, reader)
     }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
index 42b248fe..eba64a65 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
@@ -55,9 +55,10 @@ object GrpcMarshalling {
   }
 
   def negotiated[T](req: HttpRequest, f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T]): Option[Future[T]] =
-    GrpcProtocol.negotiate(req).map {
-      case (Success(reader), writer) => f(reader, writer)
-      case (Failure(ex), _)          => Future.failed(ex)
+    GrpcProtocol.negotiate(req) match {
+      case Some((Success(reader), writer)) => Some(f(reader, writer))
+      case Some((Failure(ex), _))          => Some(Future.failed(ex))
+      case None                            => None
     }
 
   def unmarshal[T](data: Source[ByteString, Any])(
@@ -125,18 +126,35 @@ object GrpcMarshalling {
     entity match {
       case HttpEntity.Strict(_, data) =>
         try {
-          val in = u.deserialize(reader.decodeSingleFrame(data))
+          val in = u match {
+            case frameSerializer: ProtobufFrameSerializer[In @unchecked] =>
+              if (data.length < AbstractGrpcProtocol.FrameHeaderSize)
+                throw new MissingParameterException
+              val frameType = data(0)
+              if ((frameType & 1) != 0)
+                throw new io.grpc.StatusException(
+                  io.grpc.Status.INTERNAL.withDescription(
+                    "Compressed-Flag bit is set, but a compression encoding is not specified"))
+              val declaredLength = data.readIntBE(1)
+              val available = data.length - AbstractGrpcProtocol.FrameHeaderSize
+              if (declaredLength > available) throw new MissingParameterException
+              if (declaredLength < 0) throw new IllegalStateException(s"Invalid frame length: $declaredLength")
+              frameSerializer.deserialize(data, AbstractGrpcProtocol.FrameHeaderSize, declaredLength)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(data))
+          }
           invokeUnary(in, implementation, eHandler)
         } catch {
-          case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case NonFatal(ex) => handleUnaryException(ex, eHandler)
         }
       case _ =>
         val requestFuture = unmarshal[In](entity)(u, mat, reader)
         requestFuture.value match {
           case Some(Success(in)) => invokeUnary(in, implementation, eHandler)
-          case Some(Failure(ex)) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case Some(Failure(ex)) => handleUnaryException(ex, eHandler)
           case None              =>
-            val exceptionHandler = unaryExceptionHandler(eHandler)
+            // Lazy: only create the PartialFunction if actually needed for recovery
+            lazy val exceptionHandler = unaryExceptionHandler(eHandler)
             requestFuture
               .flatMap(in => invokeUnary(in, implementation, eHandler))
               .recoverWith(exceptionHandler)
@@ -154,7 +172,7 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     try handleUnaryResponse(implementation(in), eHandler)
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
   @inline private def handleUnaryResponse[Out](
@@ -166,9 +184,9 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     responseFuture.value match {
       case Some(Success(out)) => marshalUnaryResponse(out, eHandler)
-      case Some(Failure(ex))  => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case Some(Failure(ex))  => handleUnaryException(ex, eHandler)
       case None               =>
-        val exceptionHandler = unaryExceptionHandler(eHandler)
+        lazy val exceptionHandler = unaryExceptionHandler(eHandler)
         responseFuture.map(out => marshal[Out](out, eHandler)(m, writer, system)).recoverWith(exceptionHandler)
     }
 
@@ -180,9 +198,25 @@ object GrpcMarshalling {
       system: ClassicActorSystemProvider): Future[HttpResponse] =
     try FastFuture.successful(marshal[Out](out, eHandler)(m, writer, system))
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
+  /**
+   * Inline exception handling for synchronous error paths.
+   * Avoids allocating a PartialFunction via GrpcExceptionHandler.from() for each exception.
+   * Instead, directly applies the mapper and converts to response.
+   */
+  @inline private def handleUnaryException(
+      ex: Throwable,
+      eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
+      implicit writer: GrpcProtocolWriter,
+      system: ClassicActorSystemProvider): Future[HttpResponse] = {
+    val mapper = eHandler(system.classicSystem)
+    val trailers = if (mapper.isDefinedAt(ex)) mapper(ex)
+    else GrpcExceptionHandler.defaultMapper(system.classicSystem)(ex)
+    FastFuture.successful(GrpcResponseHelpers.status(trailers))
+  }
+
   @inline private def unaryExceptionHandler(eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
       implicit system: ClassicActorSystemProvider,
       writer: GrpcProtocolWriter): PartialFunction[Throwable, Future[HttpResponse]] =
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
index a9884852..99450eb5 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
@@ -40,8 +40,25 @@ class ScalapbProtobufSerializer[T <: GeneratedMessage](companion: GeneratedMessa
 
     ByteString.fromArrayUnsafe(frame)
   }
-  override def deserialize(bytes: ByteString): T =
-    companion.parseFrom(CodedInputStream.newInstance(bytes.asByteBuffer))
+  override def deserialize(bytes: ByteString): T = {
+    // Fast path: use byte array directly to avoid ByteBuffer wrapper allocation
+    val bb = bytes.asByteBuffer
+    if (bb.hasArray) {
+      companion.parseFrom(CodedInputStream.newInstance(bb.array(), bb.arrayOffset(), bb.remaining()))
+    } else {
+      companion.parseFrom(CodedInputStream.newInstance(bb))
+    }
+  }
+  override private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T = {
+    // Offset-aware fast path: bypass ByteString.slice to avoid ByteString1 + ByteBuffer wrappers
+    val bb = data.asByteBuffer
+    if (bb.hasArray) {
+      companion.parseFrom(CodedInputStream.newInstance(bb.array(), bb.arrayOffset() + offset, length))
+    } else {
+      // Fallback for non-array-backed ByteStrings (rare in practice)
+      super.deserialize(data, offset, length)
+    }
+  }
   override def deserialize(data: InputStream): T =
     companion.parseFrom(data)
 }
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/GrpcProtocolSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcProtocolSpec.scala
new file mode 100644
index 00000000..fa1dcd5f
--- /dev/null
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcProtocolSpec.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+package org.apache.pekko.grpc
+
+import scala.collection.immutable
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.TryValues
+
+import org.apache.pekko
+import pekko.grpc.internal.{ Gzip, Identity }
+import pekko.grpc.scaladsl.headers
+import pekko.http.scaladsl.model.{ ContentTypes, HttpEntity, HttpRequest }
+
+class GrpcProtocolSpec extends AnyWordSpec with Matchers with TryValues {
+
+  private def grpcRequest(encoding: Option[String] = None, acceptEncoding: Option[String] = None): HttpRequest = {
+    val hdrs = immutable.Seq.newBuilder[pekko.http.scaladsl.model.HttpHeader]
+    encoding.foreach(e => hdrs += headers.`Message-Encoding`(e))
+    acceptEncoding.foreach(a => hdrs += headers.`Message-Accept-Encoding`(a))
+    HttpRequest(
+      entity = HttpEntity.Strict(GrpcProtocolNative.contentType, pekko.util.ByteString.empty),
+      headers = hdrs.result())
+  }
+
+  "GrpcProtocol.negotiate" should {
+
+    "return Identity reader and Identity writer for native gRPC with no encoding headers" in {
+      val result = GrpcProtocol.negotiate(grpcRequest())
+      result shouldBe defined
+      val (readerTry, writer) = result.get
+      readerTry.success.value.codec should be(Identity)
+      writer.codec should be(Identity)
+    }
+
+    "return Gzip reader when grpc-encoding is gzip" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(encoding = Some("gzip")))
+      result shouldBe defined
+      val (readerTry, _) = result.get
+      readerTry.success.value.codec should be(Gzip)
+    }
+
+    "return Identity reader when grpc-encoding is identity" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(encoding = Some("identity")))
+      result shouldBe defined
+      val (readerTry, _) = result.get
+      readerTry.success.value.codec should be(Identity)
+    }
+
+    "return Gzip writer when grpc-accept-encoding contains gzip" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(acceptEncoding = Some("gzip")))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.codec should be(Gzip)
+    }
+
+    "return Identity writer when grpc-accept-encoding is identity only" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(acceptEncoding = Some("identity")))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.codec should be(Identity)
+    }
+
+    "return Gzip writer when grpc-accept-encoding lists gzip first" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(acceptEncoding = Some("gzip,identity")))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.codec should be(Gzip)
+    }
+
+    "return Identity writer when grpc-accept-encoding lists identity before gzip" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(acceptEncoding = Some("identity,gzip")))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.codec should be(Identity)
+    }
+
+    "return pre-computed instance for Identity+Identity" in {
+      val result1 = GrpcProtocol.negotiate(grpcRequest())
+      val result2 = GrpcProtocol.negotiate(grpcRequest())
+      result1 shouldBe defined
+      result2 shouldBe defined
+      (result1.get eq result2.get) should be(true)
+    }
+
+    "return None for unsupported content type" in {
+      val request = HttpRequest(
+        entity = HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, pekko.util.ByteString.empty))
+      GrpcProtocol.negotiate(request) should be(None)
+    }
+
+    "return a failed reader for unknown grpc-encoding" in {
+      val result = GrpcProtocol.negotiate(grpcRequest(encoding = Some("zstd")))
+      result shouldBe defined
+      val (readerTry, _) = result.get
+      readerTry.failure.exception shouldBe a[GrpcServiceException]
+    }
+  }
+}
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
index c35c50f4..aa10204c 100644
--- a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
@@ -23,14 +23,16 @@ import scala.concurrent.Await
 import scala.concurrent.duration._
 
 import com.google.protobuf.{ Any => ProtobufAny, ByteString => ProtobufByteString }
+import io.grpc.StatusException
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
 import org.apache.pekko
 import pekko.actor.ActorSystem
-import pekko.grpc.internal.{ AbstractGrpcProtocol, GrpcProtocolNative, Identity }
+import pekko.grpc.internal.{ AbstractGrpcProtocol, GrpcProtocolNative, Identity, MissingParameterException }
 import pekko.http.scaladsl.model.HttpEntity
 import pekko.stream.SystemMaterializer
+import pekko.util.ByteString
 
 class GrpcMarshallingSpec extends AnyWordSpec with Matchers {
   "The javadsl GrpcMarshalling" should {
@@ -59,6 +61,45 @@ class GrpcMarshallingSpec extends AnyWordSpec with Matchers {
       }
     }
 
+    "decode a strict native identity frame directly" in {
+      val serializer = new GoogleProtobufSerializer(ProtobufAny.parser())
+      val message =
+        ProtobufAny.newBuilder().setTypeUrl("benchmark").setValue(ProtobufByteString.copyFromUtf8("payload")).build()
+      val payload = serializer.serialize(message)
+      val frame = AbstractGrpcProtocol.encodeFrameData(payload, isCompressed = false, isTrailer = false)
+
+      GrpcProtocolNative.newReader(Identity).decodeSingleFrame(frame) should be(payload)
+    }
+
+    "reject compressed strict native identity frames" in {
+      val frame = AbstractGrpcProtocol.encodeFrameData(ByteString(1, 2, 3), isCompressed = true, isTrailer = false)
+
+      a[StatusException] should be thrownBy GrpcProtocolNative.newReader(Identity).decodeSingleFrame(frame)
+    }
+
+    "reject truncated strict native identity frames" in {
+      a[MissingParameterException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(ByteString(0, 0, 0, 0))
+    }
+
+    "reject strict native identity frames with missing payload bytes" in {
+      a[MissingParameterException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(ByteString(0, 0, 0, 0, 3, 1, 2))
+    }
+
+    "reject strict native identity frames with trailing bytes" in {
+      val frame = AbstractGrpcProtocol.encodeFrameData(ByteString(1, 2, 3), isCompressed = false, isTrailer = false)
+
+      an[IllegalStateException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(frame ++ ByteString(4))
+    }
+
     "recover a failed unary response stage" in {
       val system = ActorSystem("GrpcMarshallingSpec")
       try {