diff --git a/api/src/main/java/io/grpc/Contexts.java b/api/src/main/java/io/grpc/Contexts.java
index c62ffc80a38..9c3697dd6d4 100644
--- a/api/src/main/java/io/grpc/Contexts.java
+++ b/api/src/main/java/io/grpc/Contexts.java
@@ -118,6 +118,16 @@ public void onReady() {
         context.detach(previous);
       }
     }
+
+    @Override
+    public void onEvent(Object event) {
+      Context previous = context.attach();
+      try {
+        super.onEvent(event);
+      } finally {
+        context.detach(previous);
+      }
+    }
   }
 
   /**
diff --git a/api/src/main/java/io/grpc/PartialForwardingServerCall.java b/api/src/main/java/io/grpc/PartialForwardingServerCall.java
index a313407b23e..8c8f53cf93c 100644
--- a/api/src/main/java/io/grpc/PartialForwardingServerCall.java
+++ b/api/src/main/java/io/grpc/PartialForwardingServerCall.java
@@ -87,6 +87,11 @@ public SecurityLevel getSecurityLevel() {
     return delegate().getSecurityLevel();
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    delegate().triggerEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();
diff --git a/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java b/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
index ca2fd0058c9..23e93bb065e 100644
--- a/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
+++ b/api/src/main/java/io/grpc/PartialForwardingServerCallListener.java
@@ -50,6 +50,11 @@ public void onReady() {
     delegate().onReady();
   }
 
+  @Override
+  public void onEvent(Object event) {
+    delegate().onEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();
diff --git a/api/src/main/java/io/grpc/ServerCall.java b/api/src/main/java/io/grpc/ServerCall.java
index 3db8ac30e83..2e6ee07a23f 100644
--- a/api/src/main/java/io/grpc/ServerCall.java
+++ b/api/src/main/java/io/grpc/ServerCall.java
@@ -100,6 +100,20 @@ public void onComplete() {}
      * <em>another</em> {@code onReady()} callback.
      */
     public void onReady() {}
+
+    /**
+     * A custom event has been triggered by the call.
+     *
+     * <p>This callback is guaranteed to run on the call's executor, serialized with other
+     * callbacks (like {@link #onMessage}, {@link #onHalfClose}). This means the implementation
+     * does not need internal synchronization to access call-specific state.
+     *
+     * @param event the triggered event.
+     */
+    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+    public void onEvent(Object event) {
+      // Default no-op
+    }
   }
 
   /**
@@ -262,6 +276,20 @@ public String getAuthority() {
     return null;
   }
 
+  /**
+   * Triggers a custom event to be processed by the listener.
+   * The event will be delivered to {@link Listener#onEvent(Object)} on the call's executor.
+   *
+   * <p>This method is thread-safe and can be called from any thread. No events will be delivered
+   * after the RPC is cancelled or completed.
+   *
+   * @param event the event to trigger.
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+  public void triggerEvent(Object event) {
+    // Default no-op
+  }
+
   /**
    * The {@link MethodDescriptor} for the call.
    */
diff --git a/api/src/test/java/io/grpc/ContextsTest.java b/api/src/test/java/io/grpc/ContextsTest.java
index ec9dc3929a2..974b1aff0c2 100644
--- a/api/src/test/java/io/grpc/ContextsTest.java
+++ b/api/src/test/java/io/grpc/ContextsTest.java
@@ -82,6 +82,11 @@ public void interceptCall_basic() {
         assertSame(uniqueContext, Context.current());
         methodCalls.add(5);
       }
+
+      @Override public void onEvent(Object event) {
+        assertSame(uniqueContext, Context.current());
+        methodCalls.add(6);
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -101,7 +106,8 @@ public ServerCall.Listener<Object> startCall(
     wrapped.onCancel();
     wrapped.onComplete();
     wrapped.onReady();
-    assertEquals(Arrays.asList(1, 2, 3, 4, 5), methodCalls);
+    wrapped.onEvent(new Object());
+    assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6), methodCalls);
     assertSame(origContext, Context.current());
   }
 
@@ -145,6 +151,10 @@ public void interceptCall_restoresIfListenerThrows() {
       @Override public void onReady() {
         throw new RuntimeException();
       }
+
+      @Override public void onEvent(Object event) {
+        throw new RuntimeException();
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -180,6 +190,11 @@ public ServerCall.Listener<Object> startCall(
       fail("Exception expected");
     } catch (RuntimeException expected) {
     }
+    try {
+      wrapped.onEvent(new Object());
+      fail("Exception expected");
+    } catch (RuntimeException expected) {
+    }
     assertSame(origContext, Context.current());
   }
 
diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java
index 83fc8273d6f..5671d808114 100644
--- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java
+++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java
@@ -668,6 +668,16 @@ protected void deliverCloseAbnormal(Status status) {
       listener.closed(status);
     }
 
+    void triggerEvent(Object event) {
+      ServerStreamListener localListener;
+      synchronized (this) {
+        localListener = listener;
+      }
+      if (localListener != null) {
+        localListener.triggerEvent(event);
+      }
+    }
+
     @GuardedBy("this")
     void onCloseSent(Status status) {
       if (!isClosed()) {
diff --git a/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java b/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
index f54769caefa..7a57138ce22 100644
--- a/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
+++ b/binder/src/main/java/io/grpc/binder/internal/MultiMessageServerStream.java
@@ -175,6 +175,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.
diff --git a/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java b/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
index 383bd7a2593..5f1dd511f73 100644
--- a/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
+++ b/binder/src/main/java/io/grpc/binder/internal/SingleMessageServerStream.java
@@ -167,6 +167,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.
diff --git a/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java b/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
index b0d84f1be74..e556954fd40 100644
--- a/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
+++ b/binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java
@@ -289,6 +289,7 @@ public ListenableFuture<Status> checkAuthorizationAsync(int uid) {
     ListenableFuture<Status> authFuture = asyncPolicy.checkAuthorizationAsync(SOME_UID);
     assertThat(awaitResult(settableUid)).isEqualTo(SOME_UID);
     authFuture.cancel(false);
+    executor.submit(() -> {}).get(10, TimeUnit.SECONDS);
 
     assertThat(delegateAuthFuture.isCancelled()).isTrue();
   }
diff --git a/core/src/main/java/io/grpc/internal/AbstractServerStream.java b/core/src/main/java/io/grpc/internal/AbstractServerStream.java
index c468cba978a..67dfdc93d42 100644
--- a/core/src/main/java/io/grpc/internal/AbstractServerStream.java
+++ b/core/src/main/java/io/grpc/internal/AbstractServerStream.java
@@ -173,6 +173,16 @@ public final void setListener(ServerStreamListener serverStreamListener) {
     transportState().setListener(serverStreamListener);
   }
 
+  @Override
+  public final void triggerEvent(final Object event) {
+    transportState().runOnTransportThread(new Runnable() {
+      @Override
+      public void run() {
+        transportState().triggerEvent(event);
+      }
+    });
+  }
+
   @Override
   public StatsTraceContext statsTraceContext() {
     return statsTraceCtx;
@@ -259,6 +269,13 @@ public void deframerClosed(boolean hasPartialMessage) {
 
 
 
+    public final void triggerEvent(Object event) {
+      if (listenerClosed) {
+        return;
+      }
+      listener().triggerEvent(event);
+    }
+
     @Override
     protected ServerStreamListener listener() {
       return listener;
diff --git a/core/src/main/java/io/grpc/internal/ServerCallImpl.java b/core/src/main/java/io/grpc/internal/ServerCallImpl.java
index e224384ce8f..6e894371be1 100644
--- a/core/src/main/java/io/grpc/internal/ServerCallImpl.java
+++ b/core/src/main/java/io/grpc/internal/ServerCallImpl.java
@@ -254,6 +254,11 @@ public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
     return method;
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    stream.triggerEvent(event);
+  }
+
   @Override
   public SecurityLevel getSecurityLevel() {
     final Attributes attributes = getAttributes();
@@ -395,5 +400,13 @@ public void onReady() {
         listener.onReady();
       }
     }
+
+    @Override
+    public void triggerEvent(Object event) {
+      if (call.cancelled) {
+        return;
+      }
+      listener.onEvent(event);
+    }
   }
 }
diff --git a/core/src/main/java/io/grpc/internal/ServerImpl.java b/core/src/main/java/io/grpc/internal/ServerImpl.java
index d9f64c2d473..767d85f443b 100644
--- a/core/src/main/java/io/grpc/internal/ServerImpl.java
+++ b/core/src/main/java/io/grpc/internal/ServerImpl.java
@@ -781,6 +781,9 @@ public void closed(Status status) {}
 
     @Override
     public void onReady() {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   /**
@@ -960,6 +963,34 @@ public void runInContext() {
         callExecutor.execute(new OnReady());
       }
     }
+
+    @Override
+    public void triggerEvent(final Object event) {
+      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
+        PerfMark.attachTag(tag);
+        final Link link = PerfMark.linkOut();
+
+        final class TriggerEvent extends ContextRunnable {
+          TriggerEvent() {
+            super(context);
+          }
+
+          @Override
+          public void runInContext() {
+            try (TaskCloseable ignore = PerfMark.traceTask("ServerCallListener(app).onEvent")) {
+              PerfMark.attachTag(tag);
+              PerfMark.linkIn(link);
+              getListener().triggerEvent(event);
+            } catch (Throwable t) {
+              internalClose(t);
+              throw t;
+            }
+          }
+        }
+
+        callExecutor.execute(new TriggerEvent());
+      }
+    }
   }
 
   @VisibleForTesting
diff --git a/core/src/main/java/io/grpc/internal/ServerStream.java b/core/src/main/java/io/grpc/internal/ServerStream.java
index aa5ba10329c..4c88e94e4d7 100644
--- a/core/src/main/java/io/grpc/internal/ServerStream.java
+++ b/core/src/main/java/io/grpc/internal/ServerStream.java
@@ -87,6 +87,12 @@ public interface ServerStream extends Stream {
    */
   void setListener(ServerStreamListener serverStreamListener);
 
+  /**
+   * Triggers a custom event. Implementations must ensure this is propagated to the
+   * listener on the transport thread.
+   */
+  void triggerEvent(Object event);
+
   /**
    * The context for recording stats and traces for this stream.
    */
diff --git a/core/src/main/java/io/grpc/internal/ServerStreamListener.java b/core/src/main/java/io/grpc/internal/ServerStreamListener.java
index e55217ab422..74de0f2079e 100644
--- a/core/src/main/java/io/grpc/internal/ServerStreamListener.java
+++ b/core/src/main/java/io/grpc/internal/ServerStreamListener.java
@@ -42,4 +42,9 @@ public interface ServerStreamListener extends StreamListener {
    * @param status details about the remote closure
    */
   void closed(Status status);
+
+  /**
+   * Propagates a custom event to the listener. Must be called on the transport thread.
+   */
+  void triggerEvent(Object event);
 }
diff --git a/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java b/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
index 137ba19bfea..5defd17fdd0 100644
--- a/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
+++ b/core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java
@@ -361,6 +361,31 @@ public void close_sendTrailersClearsReservedFields() {
     assertEquals("bad", metadataCaptor.getValue().get(InternalStatus.MESSAGE_KEY));
   }
 
+  @Test
+  public void triggerEvent_propagatesToListener() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener).triggerEvent(event);
+  }
+
+  @Test
+  public void triggerEvent_ignoredAfterClose() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    stream.close(Status.OK, new Metadata());
+    stream.transportState().complete();
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener, never()).triggerEvent(any());
+  }
+
   @Test
   public void changeOnReadyThreshold() {
     stream.setListener(new ServerStreamListenerBase());
@@ -391,6 +416,9 @@ public void halfClosed() {}
 
     @Override
     public void closed(Status status) {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   private static class AbstractServerStreamBase extends AbstractServerStream {
diff --git a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
index 7394c83eab2..4a2de9f3936 100644
--- a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
+++ b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java
@@ -493,6 +493,32 @@ public void streamListener_unexpectedRuntimeException() {
     assertThat(e).hasMessageThat().isEqualTo("unexpected exception");
   }
 
+  @Test
+  public void triggerEvent_propagatesToStream() {
+    Object event = new Object();
+    call.triggerEvent(event);
+    verify(stream).triggerEvent(event);
+  }
+
+  @Test
+  public void streamListener_triggerEvent() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+    Object event = new Object();
+    streamListener.triggerEvent(event);
+    verify(callListener).onEvent(event);
+  }
+
+  @Test
+  public void streamListener_triggerEvent_cancelled() {
+    ServerStreamListenerImpl<Long> streamListener =
+        new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
+    Object event = new Object();
+    streamListener.closed(Status.CANCELLED);
+    streamListener.triggerEvent(event);
+    verify(callListener, never()).onEvent(event);
+  }
+
   private static class LongMarshaller implements Marshaller<Long> {
     @Override
     public InputStream stream(Long value) {
diff --git a/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java b/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
index 5d07de32df9..5127c7c2b0f 100644
--- a/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
+++ b/core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java
@@ -2088,6 +2088,35 @@ public void clientChecksInboundMetadataSize_trailer() throws Exception {
     assertNull(metadata.get(tellTaleKey));
   }
 
+  @Test
+  public void serverStream_triggerEvent() throws Exception {
+    server.start(serverListener);
+    client = newClientTransport(server);
+    startTransport(client, mockClientTransportListener);
+    MockServerTransportListener serverTransportListener
+        = serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    serverTransport = serverTransportListener.transport;
+
+    ClientStream clientStream = client.newStream(
+        methodDescriptor, new Metadata(), callOptions, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    clientStream.start(clientStreamListener);
+
+    StreamCreation serverStreamCreation
+        = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    ServerStream serverStream = serverStreamCreation.stream;
+    ServerStreamListenerBase serverStreamListener = serverStreamCreation.listener;
+
+    Object event = new Object();
+    serverStream.triggerEvent(event);
+
+    Object receivedEvent = serverStreamListener.eventQueue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    assertEquals(event, receivedEvent);
+
+    // Cleanup
+    clientStream.cancel(Status.CANCELLED);
+  }
+
   /**
    * Helper that simply does an RPC. It can be used similar to a sleep for negative testing: to give
    * time for actions _not_ to happen. Since it is based on doing an actual RPC with actual
diff --git a/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java b/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
index aaa70600542..e4ac01912e4 100644
--- a/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
+++ b/core/src/testFixtures/java/io/grpc/internal/ServerStreamListenerBase.java
@@ -89,6 +89,8 @@ public void halfClosed() {
     halfClosedLatch.countDown();
   }
 
+  public final BlockingQueue<Object> eventQueue = new LinkedBlockingQueue<>();
+
   @Override
   public void closed(Status status) {
     if (this.status.isDone()) {
@@ -96,4 +98,12 @@ public void closed(Status status) {
     }
     this.status.set(status);
   }
+
+  @Override
+  public void triggerEvent(Object event) {
+    if (this.status.isDone()) {
+      fail("triggerEvent invoked after closed");
+    }
+    eventQueue.add(event);
+  }
 }
diff --git a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
index a92f10fd5c5..57820b396ad 100644
--- a/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
+++ b/inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java
@@ -429,6 +429,11 @@ public void setListener(ServerStreamListener serverStreamListener) {
         clientStream.setListener(serverStreamListener);
       }
 
+      @Override
+      public void triggerEvent(Object event) {
+        clientStream.triggerServerEvent(event);
+      }
+
       @Override
       public void request(int numMessages) {
         boolean onReady = clientStream.serverRequested(numMessages);
@@ -732,6 +737,20 @@ private synchronized void setListener(ServerStreamListener listener) {
         this.serverStreamListener = listener;
       }
 
+      void triggerServerEvent(final Object event) {
+        synchronized (this) {
+          if (!closed) {
+            syncContext.executeLater(new Runnable() {
+              @Override
+              public void run() {
+                serverStreamListener.triggerEvent(event);
+              }
+            });
+          }
+        }
+        syncContext.drain();
+      }
+
       @Override
       public void request(int numMessages) {
         boolean onReady = serverStream.clientRequested(numMessages);
diff --git a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
index ef8d2e5efda..b22be2460f1 100644
--- a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
+++ b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java
@@ -1343,6 +1343,10 @@ public void halfClosed() {
     @Override
     public void closed(Status status) {
     }
+
+    @Override
+    public void triggerEvent(Object event) {
+    }
   }
 
   private final class EchoServerListener implements ServerListener {
diff --git a/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java b/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
index 00db6e1d339..d3e50cf3821 100644
--- a/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
+++ b/okhttp/src/test/java/io/grpc/okhttp/OkHttpServerTransportTest.java
@@ -1519,6 +1519,10 @@ public void closed(Status status) {
     public void onReady() {
     }
 
+    @Override
+    public void triggerEvent(Object event) {
+    }
+
     static String getContent(InputStream message) throws IOException {
       try {
         return new String(ByteStreams.toByteArray(message), UTF_8);