Repository: LangStream/langstream (language: python)

Issue:
Introduce events system backed by streaming topics
### Use case

As application developer, I want to listen to all gateway client new connections. I'd like to consume from a known "events" topic such events and react somehow, for example sending a message to the target topic with a welcome message.

### Proposal
Since events would be useful in other scenarios (e.g. message process failed event) let's introduce a abstract event API which will be enhanced in the future. The first implementation would be to register clients opening/closing connections to the gateway.


gateways.yaml
```
gateways:
  - id: consume-output
    type: consume
    events-topic: events-topic
```

pipeline.yaml
```
topics:
  - name: "events-topic"
    creation-mode: create-if-not-exists
```

The API gateway will check for the app configuration and writes to the events topic if configured to do so.
The schema will be the same for all the events (even non about gateways). It will be in JSON and this format:

```
{
 "type": "ClientConnectedToGateway"|"ClientDisconnectedToGateway"
 "timestamp": <long timestamp>
  "source": {
     # this changes based on the type
     "gatewayId": "xxx",
     "applicationId": "zz",
     "tenant": "tt"
    },
  "data": { 
        # custom data depending on the event kind
       "params": {...}
       "userAgent": ....
   }
}
``` 

 


Discussion / hints:

Fix patch:
diff --git a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/AuthenticationInterceptor.java b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/AuthenticationInterceptor.java
index aa6d25885..9c86d3c9f 100644
--- a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/AuthenticationInterceptor.java
+++ b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/AuthenticationInterceptor.java
@@ -62,7 +62,8 @@ public boolean beforeHandshake(
             final Map<String, String> vars =
                     antPathMatcher.extractUriTemplateVariables(handler.path(), path);
             final GatewayRequestContext gatewayRequestContext =
-                    handler.validateRequest(vars, querystring);
+                    handler.validateRequest(
+                            vars, querystring, request.getHeaders().toSingleValueMap());
 
             final Map<String, String> principalValues;
             try {
@@ -82,7 +83,7 @@ public boolean beforeHandshake(
                     getAuthenticatedGatewayRequestContext(
                             gatewayRequestContext, principalValues, attributes);
             attributes.put("context", authenticatedGatewayRequestContext);
-            handler.onBeforeHandshakeCompleted(authenticatedGatewayRequestContext);
+            handler.onBeforeHandshakeCompleted(authenticatedGatewayRequestContext, attributes);
             return true;
         } catch (Throwable error) {
             log.info("Internal error {}", error.getMessage(), error);
@@ -174,6 +175,11 @@ public Map<String, String> userParameters() {
             public Map<String, String> options() {
                 return gatewayRequestContext.options();
             }
+
+            @Override
+            public Map<String, String> httpHeaders() {
+                return gatewayRequestContext.httpHeaders();
+            }
         };
     }
 
diff --git a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/AbstractHandler.java b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/AbstractHandler.java
index c9746d1e0..ddcc88629 100644
--- a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/AbstractHandler.java
+++ b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/AbstractHandler.java
@@ -15,12 +15,19 @@
  */
 package ai.langstream.apigateway.websocket.handlers;
 
+import ai.langstream.api.events.EventRecord;
+import ai.langstream.api.events.EventSources;
+import ai.langstream.api.events.GatewayEventData;
 import ai.langstream.api.gateway.GatewayRequestContext;
 import ai.langstream.api.model.Application;
 import ai.langstream.api.model.ApplicationSpecs;
 import ai.langstream.api.model.Gateway;
 import ai.langstream.api.model.Gateways;
+import ai.langstream.api.model.StreamingCluster;
+import ai.langstream.api.runner.code.SimpleRecord;
+import ai.langstream.api.runner.topics.TopicConnectionsRuntime;
 import ai.langstream.api.runner.topics.TopicConnectionsRuntimeRegistry;
+import ai.langstream.api.runner.topics.TopicProducer;
 import ai.langstream.api.storage.ApplicationStore;
 import ai.langstream.apigateway.websocket.AuthenticatedGatewayRequestContext;
 import ai.langstream.impl.common.ApplicationPlaceholderResolver;
@@ -34,6 +41,7 @@
 import java.util.Map;
 import java.util.Set;
 import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.util.StringUtils;
 import org.springframework.web.socket.CloseStatus;
@@ -61,7 +69,9 @@ abstract String applicationIdFromPath(
     abstract String gatewayFromPath(
             Map<String, String> parsedPath, Map<String, String> queryString);
 
-    public void onBeforeHandshakeCompleted(AuthenticatedGatewayRequestContext gatewayRequestContext)
+    public void onBeforeHandshakeCompleted(
+            AuthenticatedGatewayRequestContext gatewayRequestContext,
+            Map<String, Object> attributes)
             throws Exception {}
 
     abstract void onOpen(
@@ -131,6 +141,7 @@ public void afterConnectionClosed(WebSocketSession session, CloseStatus status)
             log.error("[{}] error while closing websocket", session.getId(), throwable);
         }
         closeCloseableResources(session);
+        sendClientDisconnectedEvent(getContext(session));
     }
 
     @Override
@@ -179,7 +190,9 @@ private Gateway extractGateway(
     }
 
     public GatewayRequestContext validateRequest(
-            Map<String, String> pathVars, Map<String, String> queryString) {
+            Map<String, String> pathVars,
+            Map<String, String> queryString,
+            Map<String, String> httpHeaders) {
         Map<String, String> options = new HashMap<>();
         Map<String, String> userParameters = new HashMap<>();
 
@@ -261,19 +274,23 @@ public Map<String, String> userParameters() {
             public Map<String, String> options() {
                 return options;
             }
+
+            @Override
+            public Map<String, String> httpHeaders() {
+                return httpHeaders;
+            }
         };
     }
 
     protected void recordCloseableResource(
-            WebSocketSession webSocketSession, AutoCloseable... closeables) {
-        List<AutoCloseable> currentCloseable =
-                (List<AutoCloseable>) webSocketSession.getAttributes().get("closeables");
+            Map<String, Object> attributes, AutoCloseable... closeables) {
+        List<AutoCloseable> currentCloseable = (List<AutoCloseable>) attributes.get("closeables");
 
         if (currentCloseable == null) {
             currentCloseable = new ArrayList<>();
         }
         Collections.addAll(currentCloseable, closeables);
-        webSocketSession.getAttributes().put("closeables", currentCloseable);
+        attributes.put("closeables", currentCloseable);
     }
 
     private void closeCloseableResources(WebSocketSession webSocketSession) {
@@ -290,4 +307,65 @@ private void closeCloseableResources(WebSocketSession webSocketSession) {
             }
         }
     }
+
+    protected void sendClientConnectedEvent(AuthenticatedGatewayRequestContext context) {
+        sendEvent(EventRecord.Types.ClientConnected, context);
+    }
+
+    protected void sendClientDisconnectedEvent(AuthenticatedGatewayRequestContext context) {
+        try {
+            sendEvent(EventRecord.Types.ClientDisconnected, context);
+        } catch (Throwable e) {
+            log.error("error while sending client disconnected event", e);
+        }
+    }
+
+    @SneakyThrows
+    protected void sendEvent(EventRecord.Types type, AuthenticatedGatewayRequestContext context) {
+        final Gateway gateway = context.gateway();
+        if (gateway.eventsTopic() == null) {
+            return;
+        }
+        final StreamingCluster streamingCluster =
+                context.application().getInstance().streamingCluster();
+        final TopicConnectionsRuntime topicConnectionsRuntime =
+                TOPIC_CONNECTIONS_REGISTRY.getTopicConnectionsRuntime(streamingCluster);
+
+        try (final TopicProducer producer =
+                topicConnectionsRuntime.createProducer(
+                        "langstream-events",
+                        streamingCluster,
+                        Map.of("topic", gateway.eventsTopic())); ) {
+            producer.start();
+
+            final EventSources.GatewaySource source =
+                    EventSources.GatewaySource.builder()
+                            .tenant(context.tenant())
+                            .applicationId(context.applicationId())
+                            .gateway(gateway)
+                            .build();
+
+            final GatewayEventData data =
+                    GatewayEventData.builder()
+                            .userParameters(context.userParameters())
+                            .options(context.options())
+                            .httpRequestHeaders(context.httpHeaders())
+                            .build();
+
+            final EventRecord event =
+                    EventRecord.builder()
+                            .category(EventRecord.Categories.Gateway)
+                            .type(type.toString())
+                            .timestamp(System.currentTimeMillis())
+                            .source(mapper.convertValue(source, Map.class))
+                            .data(mapper.convertValue(data, Map.class))
+                            .build();
+
+            final String recordValue = mapper.writeValueAsString(event);
+
+            final SimpleRecord record = SimpleRecord.builder().value(recordValue).build();
+            producer.write(record).get();
+            log.info("sent event {}", recordValue);
+        }
+    }
 }
diff --git a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ConsumeHandler.java b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ConsumeHandler.java
index 7c7e02f03..f51a17742 100644
--- a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ConsumeHandler.java
+++ b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ConsumeHandler.java
@@ -84,7 +84,8 @@ String gatewayFromPath(Map<String, String> parsedPath, Map<String, String> query
     }
 
     @Override
-    public void onBeforeHandshakeCompleted(AuthenticatedGatewayRequestContext context)
+    public void onBeforeHandshakeCompleted(
+            AuthenticatedGatewayRequestContext context, Map<String, Object> attributes)
             throws Exception {
         final Gateway gateway = context.gateway();
         final Application application = context.application();
@@ -115,6 +116,7 @@ public void onBeforeHandshakeCompleted(AuthenticatedGatewayRequestContext contex
                         streamingCluster, Map.of("topic", topicName), position);
         reader.start();
         context.attributes().put("topicReader", reader);
+        sendClientConnectedEvent(context);
     }
 
     @Override
diff --git a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ProduceHandler.java b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ProduceHandler.java
index f1f2f85cd..aa1efd5f7 100644
--- a/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ProduceHandler.java
+++ b/langstream-api-gateway/src/main/java/ai/langstream/apigateway/websocket/handlers/ProduceHandler.java
@@ -20,7 +20,6 @@
 import ai.langstream.api.model.Gateway;
 import ai.langstream.api.model.StreamingCluster;
 import ai.langstream.api.runner.code.Header;
-import ai.langstream.api.runner.code.Record;
 import ai.langstream.api.runner.code.SimpleRecord;
 import ai.langstream.api.runner.topics.TopicConnectionsRuntime;
 import ai.langstream.api.runner.topics.TopicProducer;
@@ -37,8 +36,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
-import lombok.AllArgsConstructor;
-import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.TextMessage;
@@ -77,39 +74,41 @@ String gatewayFromPath(Map<String, String> parsedPath, Map<String, String> query
     }
 
     @Override
-    public void onOpen(
-            WebSocketSession webSocketSession, AuthenticatedGatewayRequestContext context) {
-        Gateway selectedGateway = context.gateway();
+    public void onBeforeHandshakeCompleted(
+            AuthenticatedGatewayRequestContext context, Map<String, Object> attributes)
+            throws Exception {
+        Gateway gateway = context.gateway();
 
         final List<Header> headers =
-                getCommonHeaders(
-                        selectedGateway, context.userParameters(), context.principalValues());
+                getCommonHeaders(gateway, context.userParameters(), context.principalValues());
         final StreamingCluster streamingCluster =
                 context.application().getInstance().streamingCluster();
 
         final TopicConnectionsRuntime topicConnectionsRuntime =
                 TOPIC_CONNECTIONS_REGISTRY.getTopicConnectionsRuntime(streamingCluster);
 
-        final String topicName = selectedGateway.topic();
+        final String topicName = gateway.topic();
         final TopicProducer producer =
                 topicConnectionsRuntime.createProducer(
-                        "ag-" + webSocketSession.getId(),
-                        streamingCluster,
-                        Map.of("topic", topicName));
-        recordCloseableResource(webSocketSession, producer);
+                        null, streamingCluster, Map.of("topic", topicName));
+        recordCloseableResource(attributes, producer);
         producer.start();
 
-        webSocketSession.getAttributes().put("producer", producer);
-        webSocketSession.getAttributes().put("headers", Collections.unmodifiableList(headers));
-
+        attributes.put("producer", producer);
+        attributes.put("headers", Collections.unmodifiableList(headers));
         log.info(
                 "Started produced for gateway {}/{}/{} on topic {}",
                 context.tenant(),
                 context.applicationId(),
                 context.gateway().id(),
                 topicName);
+        sendClientConnectedEvent(context);
     }
 
+    @Override
+    public void onOpen(
+            WebSocketSession webSocketSession, AuthenticatedGatewayRequestContext context) {}
+
     @Override
     public void onMessage(
             WebSocketSession webSocketSession,
@@ -155,8 +154,12 @@ public void onMessage(
             }
         }
         try {
-            final ProduceHandlerRecord record =
-                    new ProduceHandlerRecord(produceRequest.key(), produceRequest.value(), headers);
+            final SimpleRecord record =
+                    SimpleRecord.builder()
+                            .key(produceRequest.key())
+                            .value(produceRequest.value())
+                            .headers(headers)
+                            .build();
             topicProducer.write(record).get();
             log.info("[{}] Produced record {}", webSocketSession.getId(), record);
         } catch (Throwable tt) {
@@ -233,37 +236,4 @@ private List<Header> getCommonHeaders(
         }
         return headers;
     }
-
-    @AllArgsConstructor
-    @ToString
-    private static class ProduceHandlerRecord implements Record {
-        private final Object key;
-        private final Object value;
-        private final Collection<Header> headers;
-
-        @Override
-        public Object key() {
-            return key;
-        }
-
-        @Override
-        public Object value() {
-            return value;
-        }
-
-        @Override
-        public String origin() {
-            return null;
-        }
-
-        @Override
-        public Long timestamp() {
-            return null;
-        }
-
-        @Override
-        public Collection<Header> headers() {
-            return headers;
-        }
-    }
 }
diff --git a/langstream-api-gateway/src/main/resources/application.properties b/langstream-api-gateway/src/main/resources/application.properties
index e02ea1359..4dcdad86a 100644
--- a/langstream-api-gateway/src/main/resources/application.properties
+++ b/langstream-api-gateway/src/main/resources/application.properties
@@ -27,5 +27,4 @@ spring.jackson.serialization.indent-output=true
 spring.jackson.serialization.order-map-entries-by-keys=true
 
 application.storage.apps.type=kubernetes
-application.storage.apps.configuration.namespaceprefix=langstream-
-application.storage.apps.configuration.deployer-runtime.image=langstream/langstream-runtime:latest-dev
\ No newline at end of file
+application.storage.apps.configuration.namespaceprefix=langstream-
\ No newline at end of file
diff --git a/langstream-api/src/main/java/ai/langstream/api/events/EventRecord.java b/langstream-api/src/main/java/ai/langstream/api/events/EventRecord.java
new file mode 100644
index 000000000..105c324bd
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/events/EventRecord.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.events;
+
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class EventRecord {
+
+    public enum Categories {
+        Gateway
+    }
+
+    public enum Types {
+        // gateway
+        ClientConnected,
+        ClientDisconnected;
+    }
+
+    private Categories category;
+    private String type;
+    private long timestamp;
+    private Map<String, Object> source;
+    private Map<String, Object> data;
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/events/EventSources.java b/langstream-api/src/main/java/ai/langstream/api/events/EventSources.java
new file mode 100644
index 000000000..e7b41f8a7
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/events/EventSources.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.events;
+
+import ai.langstream.api.model.Gateway;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+public class EventSources {
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class ApplicationSource {
+        private String tenant;
+        private String applicationId;
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class GatewaySource extends ApplicationSource {
+        private Gateway gateway;
+
+        @Builder
+        public GatewaySource(String tenant, String applicationId, Gateway gateway) {
+            super(tenant, applicationId);
+            this.gateway = gateway;
+        }
+    }
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/events/GatewayEventData.java b/langstream-api/src/main/java/ai/langstream/api/events/GatewayEventData.java
new file mode 100644
index 000000000..7aeb96724
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/events/GatewayEventData.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.events;
+
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class GatewayEventData {
+
+    private Map<String, String> userParameters;
+    private Map<String, String> options;
+    private Map<String, String> httpRequestHeaders;
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/gateway/GatewayRequestContext.java b/langstream-api/src/main/java/ai/langstream/api/gateway/GatewayRequestContext.java
index 434256ffc..e25a8451e 100644
--- a/langstream-api/src/main/java/ai/langstream/api/gateway/GatewayRequestContext.java
+++ b/langstream-api/src/main/java/ai/langstream/api/gateway/GatewayRequestContext.java
@@ -34,4 +34,6 @@ public interface GatewayRequestContext {
     Map<String, String> userParameters();
 
     Map<String, String> options();
+
+    Map<String, String> httpHeaders();
 }
diff --git a/langstream-api/src/main/java/ai/langstream/api/model/Gateway.java b/langstream-api/src/main/java/ai/langstream/api/model/Gateway.java
index 2c63dcadd..5a9ec9b17 100644
--- a/langstream-api/src/main/java/ai/langstream/api/model/Gateway.java
+++ b/langstream-api/src/main/java/ai/langstream/api/model/Gateway.java
@@ -15,6 +15,8 @@
  */
 package ai.langstream.api.model;
 
+import com.fasterxml.jackson.annotation.JsonAlias;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import java.util.List;
 import java.util.Map;
 
@@ -24,8 +26,9 @@ public record Gateway(
         String topic,
         Authentication authentication,
         List<String> parameters,
-        ProduceOptions produceOptions,
-        ConsumeOptions consumeOptions) {
+        @JsonAlias({"produce-options"}) ProduceOptions produceOptions,
+        @JsonAlias({"consume-options"}) ConsumeOptions consumeOptions,
+        @JsonProperty("events-topic") String eventsTopic) {
     public enum GatewayType {
         produce,
         consume
@@ -38,7 +41,18 @@ public Gateway(
             List<String> parameters,
             ProduceOptions produceOptions,
             ConsumeOptions consumeOptions) {
-        this(id, type, topic, null, parameters, produceOptions, consumeOptions);
+        this(id, type, topic, null, parameters, produceOptions, consumeOptions, null);
+    }
+
+    public Gateway(
+            String id,
+            GatewayType type,
+            String topic,
+            Authentication authentication,
+            List<String> parameters,
+            ProduceOptions produceOptions,
+            ConsumeOptions consumeOptions) {
+        this(id, type, topic, authentication, parameters, produceOptions, consumeOptions, null);
     }
 
     public record Authentication(String provider, Map<String, Object> configuration) {}
diff --git a/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java b/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
index 63ae4abe2..847688c65 100644
--- a/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
+++ b/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
@@ -183,15 +183,22 @@ private static Gateways resolveGateways(Application instance, Map<String, Object
                                 authentication.provider(),
                                 resolveMap(context, gateway.authentication().configuration()));
             }
+
+            final String topic = resolveValue(context, gateway.topic());
+            final String eventsTopic =
+                    gateway.eventsTopic() == null
+                            ? null
+                            : resolveValue(context, gateway.eventsTopic());
             newGateways.add(
                     new Gateway(
                             gateway.id(),
                             gateway.type(),
-                            gateway.topic(),
+                            topic,
                             authentication,
                             gateway.parameters(),
                             gateway.produceOptions(),
-                            gateway.consumeOptions()));
+                            gateway.consumeOptions(),
+                            eventsTopic));
         }
         return new Gateways(newGateways);
     }
diff --git a/langstream-kafka/src/main/java/ai/langstream/kafka/runner/KafkaTopicConnectionsRuntime.java b/langstream-kafka/src/main/java/ai/langstream/kafka/runner/KafkaTopicConnectionsRuntime.java
index 68a4bb9a1..677ad6ba1 100644
--- a/langstream-kafka/src/main/java/ai/langstream/kafka/runner/KafkaTopicConnectionsRuntime.java
+++ b/langstream-kafka/src/main/java/ai/langstream/kafka/runner/KafkaTopicConnectionsRuntime.java
@@ -56,19 +56,14 @@ public TopicConsumer createConsumer(
             String agentId, StreamingCluster streamingCluster, Map<String, Object> configuration) {
 
         Map<String, Object> copy = new HashMap<>(configuration);
-        applyDefaultConfiguration(agentId, streamingCluster, copy);
+        applyDefaultConfiguration(streamingCluster, copy);
+        applyConsumerConfiguration(agentId, copy);
         String topicName = (String) copy.remove("topic");
 
         return new KafkaConsumerWrapper(copy, topicName);
     }
 
-    private static void applyDefaultConfiguration(
-            String agentId, StreamingCluster streamingCluster, Map<String, Object> copy) {
-        KafkaClusterRuntimeConfiguration configuration =
-                KafkaStreamingClusterRuntime.getKafkaClusterRuntimeConfiguration(streamingCluster);
-        copy.putAll(configuration.getAdmin());
-
-        // consumer
+    private void applyConsumerConfiguration(String agentId, Map<String, Object> copy) {
         copy.putIfAbsent(
                 "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         copy.putIfAbsent(
@@ -77,8 +72,16 @@ private static void applyDefaultConfiguration(
         copy.putIfAbsent("enable.auto.commit", "false");
         copy.putIfAbsent("group.id", "langstream-" + agentId);
         copy.putIfAbsent("auto.offset.reset", "earliest");
+    }
+
+    private static void applyDefaultConfiguration(
+            StreamingCluster streamingCluster, Map<String, Object> copy) {
+        KafkaClusterRuntimeConfiguration configuration =
+                KafkaStreamingClusterRuntime.getKafkaClusterRuntimeConfiguration(streamingCluster);
+        copy.putAll(configuration.getAdmin());
+    }
 
-        // producer
+    private static void applyProducerConfiguration(Map<String, Object> copy) {
         copy.putIfAbsent(
                 "key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
         copy.putIfAbsent(
@@ -89,7 +92,8 @@ private static void applyDefaultConfiguration(
     public TopicProducer createProducer(
             String agentId, StreamingCluster streamingCluster, Map<String, Object> configuration) {
         Map<String, Object> copy = new HashMap<>(configuration);
-        applyDefaultConfiguration(agentId, streamingCluster, copy);
+        applyDefaultConfiguration(streamingCluster, copy);
+        applyProducerConfiguration(copy);
         String topicName = (String) copy.remove("topic");
 
         return new KafkaProducerWrapper(copy, topicName);
@@ -114,7 +118,7 @@ public TopicProducer createDeadletterTopicProducer(
     public TopicAdmin createTopicAdmin(
             String agentId, StreamingCluster streamingCluster, Map<String, Object> configuration) {
         Map<String, Object> copy = new HashMap<>(configuration);
-        applyDefaultConfiguration(agentId, streamingCluster, copy);
+        applyDefaultConfiguration(streamingCluster, copy);
         return new TopicAdmin() {
 
             org.apache.kafka.connect.util.TopicAdmin topicAdmin;


Test patch:
diff --git a/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/ProduceConsumeHandlerTest.java b/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/ProduceConsumeHandlerTest.java
index 6e2ee426e..47f3f7c5b 100644
--- a/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/ProduceConsumeHandlerTest.java
+++ b/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/ProduceConsumeHandlerTest.java
@@ -16,12 +16,17 @@
 package ai.langstream.apigateway.websocket.handlers;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doAnswer;
 
+import ai.langstream.api.events.EventRecord;
+import ai.langstream.api.events.EventSources;
+import ai.langstream.api.events.GatewayEventData;
 import ai.langstream.api.model.Application;
 import ai.langstream.api.model.ApplicationSpecs;
 import ai.langstream.api.model.Gateway;
@@ -1132,4 +1137,157 @@ public void close() {
             }
         };
     }
+
+    @Test
+    void testSendEvents() throws Exception {
+        final String topic = genTopic();
+        final String eventsTopic = genTopic();
+        prepareTopicsForTest(topic, eventsTopic);
+        testGateways =
+                new Gateways(
+                        List.of(
+                                new Gateway(
+                                        "produce",
+                                        Gateway.GatewayType.produce,
+                                        topic,
+                                        null,
+                                        List.of("p"),
+                                        null,
+                                        null,
+                                        eventsTopic),
+                                new Gateway(
+                                        "consume",
+                                        Gateway.GatewayType.consume,
+                                        eventsTopic,
+                                        null,
+                                        List.of("p"),
+                                        null,
+                                        null,
+                                        eventsTopic)));
+
+        CountDownLatch consumerReady = new CountDownLatch(1);
+        CountDownLatch countDownLatch = new CountDownLatch(5);
+        List<String> messages = new ArrayList<>();
+        try (final TestWebSocketClient consumerClient =
+                new TestWebSocketClient(
+                                new TestWebSocketClient.Handler() {
+                                    @Override
+                                    public void onMessage(String msg) {
+                                        messages.add(msg);
+                                        countDownLatch.countDown();
+                                    }
+
+                                    @Override
+                                    public void onOpen(Session session) {
+                                        consumerReady.countDown();
+                                    }
+
+                                    @Override
+                                    public void onClose(CloseReason closeReason) {
+                                        countDownLatch.countDown();
+                                    }
+
+                                    @Override
+                                    public void onError(Throwable throwable) {
+                                        countDownLatch.countDown();
+                                    }
+                                })
+                        .connect(
+                                URI.create(
+                                        "ws://localhost:%d/v1/consume/tenant1/application1/consume?param:p=consumer"
+                                                .formatted(port)))) {
+            consumerReady.await();
+
+            try (final TestWebSocketClient producer =
+                    new TestWebSocketClient(TestWebSocketClient.NOOP)
+                            .connect(
+                                    URI.create(
+                                            "ws://localhost:%d/v1/produce/tenant1/application1/produce?param:p=producer"
+                                                    .formatted(port)))) {
+                final ProduceRequest produceRequest =
+                        new ProduceRequest(null, "this is a message", null);
+                produce(produceRequest, producer);
+            }
+            new TestWebSocketClient(new TestWebSocketClient.Handler() {})
+                    .connect(
+                            URI.create(
+                                    "ws://localhost:%d/v1/consume/tenant1/application1/consume?param:p=consumer1"
+                                            .formatted(port)))
+                    .close();
+
+            countDownLatch.await();
+            System.out.println("got messages: " + messages);
+
+            ConsumePushMessage msg = MAPPER.readValue(messages.get(0), ConsumePushMessage.class);
+            EventRecord event = MAPPER.readValue(msg.record().value() + "", EventRecord.class);
+            assertEquals(EventRecord.Categories.Gateway, event.getCategory());
+            assertEquals(EventRecord.Types.ClientConnected + "", event.getType());
+            EventSources.GatewaySource source =
+                    MAPPER.convertValue(event.getSource(), EventSources.GatewaySource.class);
+            assertEquals("tenant1", source.getTenant());
+            assertEquals("application1", source.getApplicationId());
+            assertEquals("consume", source.getGateway().id());
+            GatewayEventData data = MAPPER.convertValue(event.getData(), GatewayEventData.class);
+            assertEquals("consumer", data.getUserParameters().get("p"));
+            assertEquals(0, data.getOptions().size());
+            assertNotNull(data.getHttpRequestHeaders().get("host"));
+            assertTrue(event.getTimestamp() > 0);
+
+            msg = MAPPER.readValue(messages.get(1), ConsumePushMessage.class);
+            event = MAPPER.readValue(msg.record().value() + "", EventRecord.class);
+            assertEquals(EventRecord.Categories.Gateway, event.getCategory());
+            assertEquals(EventRecord.Types.ClientConnected + "", event.getType());
+            source = MAPPER.convertValue(event.getSource(), EventSources.GatewaySource.class);
+            assertEquals("tenant1", source.getTenant());
+            assertEquals("application1", source.getApplicationId());
+            assertEquals("produce", source.getGateway().id());
+            data = MAPPER.convertValue(event.getData(), GatewayEventData.class);
+            assertEquals("producer", data.getUserParameters().get("p"));
+            assertNotNull(data.getHttpRequestHeaders().get("host"));
+            assertEquals(0, data.getOptions().size());
+            assertTrue(event.getTimestamp() > 0);
+
+            msg = MAPPER.readValue(messages.get(2), ConsumePushMessage.class);
+            event = MAPPER.readValue(msg.record().value() + "", EventRecord.class);
+            assertEquals(EventRecord.Categories.Gateway, event.getCategory());
+            assertEquals(EventRecord.Types.ClientDisconnected + "", event.getType());
+            source = MAPPER.convertValue(event.getSource(), EventSources.GatewaySource.class);
+            assertEquals("tenant1", source.getTenant());
+            assertEquals("application1", source.getApplicationId());
+            assertEquals("produce", source.getGateway().id());
+            data = MAPPER.convertValue(event.getData(), GatewayEventData.class);
+            assertEquals("producer", data.getUserParameters().get("p"));
+            assertNotNull(data.getHttpRequestHeaders().get("host"));
+            assertEquals(0, data.getOptions().size());
+            assertTrue(event.getTimestamp() > 0);
+
+            msg = MAPPER.readValue(messages.get(3), ConsumePushMessage.class);
+            event = MAPPER.readValue(msg.record().value() + "", EventRecord.class);
+            assertEquals(EventRecord.Categories.Gateway, event.getCategory());
+            assertEquals(EventRecord.Types.ClientConnected + "", event.getType());
+            source = MAPPER.convertValue(event.getSource(), EventSources.GatewaySource.class);
+            assertEquals("tenant1", source.getTenant());
+            assertEquals("application1", source.getApplicationId());
+            assertEquals("consume", source.getGateway().id());
+            data = MAPPER.convertValue(event.getData(), GatewayEventData.class);
+            assertEquals("consumer1", data.getUserParameters().get("p"));
+            assertNotNull(data.getHttpRequestHeaders().get("host"));
+            assertEquals(0, data.getOptions().size());
+            assertTrue(event.getTimestamp() > 0);
+
+            msg = MAPPER.readValue(messages.get(4), ConsumePushMessage.class);
+            event = MAPPER.readValue(msg.record().value() + "", EventRecord.class);
+            assertEquals(EventRecord.Categories.Gateway, event.getCategory());
+            assertEquals(EventRecord.Types.ClientDisconnected + "", event.getType());
+            source = MAPPER.convertValue(event.getSource(), EventSources.GatewaySource.class);
+            assertEquals("tenant1", source.getTenant());
+            assertEquals("application1", source.getApplicationId());
+            assertEquals("consume", source.getGateway().id());
+            data = MAPPER.convertValue(event.getData(), GatewayEventData.class);
+            assertEquals("consumer1", data.getUserParameters().get("p"));
+            assertNotNull(data.getHttpRequestHeaders().get("host"));
+            assertEquals(0, data.getOptions().size());
+            assertTrue(event.getTimestamp() > 0);
+        }
+    }
 }
diff --git a/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/TestWebSocketClient.java b/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/TestWebSocketClient.java
index dd2226ddb..18b36b91b 100644
--- a/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/TestWebSocketClient.java
+++ b/langstream-api-gateway/src/test/java/ai/langstream/apigateway/websocket/handlers/TestWebSocketClient.java
@@ -50,9 +50,11 @@ default void onOpen(Session session) {
             log.info("onOpen client: {}", session.getId());
         }
 
-        void onMessage(String msg);
+        default void onMessage(String msg) {}
+        ;
 
-        void onClose(CloseReason closeReason);
+        default void onClose(CloseReason closeReason) {}
+        ;
 
         default void onError(Throwable throwable) {
             throw new RuntimeException(throwable);
diff --git a/langstream-core/src/test/java/ai/langstream/impl/common/ApplicationPlaceholderResolverTest.java b/langstream-core/src/test/java/ai/langstream/impl/common/ApplicationPlaceholderResolverTest.java
index dd26fce30..4ea5405e2 100644
--- a/langstream-core/src/test/java/ai/langstream/impl/common/ApplicationPlaceholderResolverTest.java
+++ b/langstream-core/src/test/java/ai/langstream/impl/common/ApplicationPlaceholderResolverTest.java
@@ -15,6 +15,8 @@
  */
 package ai.langstream.impl.common;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 import ai.langstream.api.model.Application;
 import ai.langstream.api.model.Resource;
 import ai.langstream.impl.parser.ModelBuilder;
@@ -54,15 +56,15 @@ void testAvailablePlaceholders() throws Exception {
 
         final Map<String, Object> context =
                 ApplicationPlaceholderResolver.createContext(applicationInstance);
-        Assertions.assertEquals(
+        assertEquals(
                 "my-access-key",
                 ApplicationPlaceholderResolver.resolveValue(
                         context, "{{secrets.openai-credentials.accessKey}}"));
-        Assertions.assertEquals(
+        assertEquals(
                 "http://mypulsar.localhost:8080",
                 ApplicationPlaceholderResolver.resolveValue(
                         context, "{{cluster.configuration.admin.serviceUrl}}"));
-        Assertions.assertEquals(
+        assertEquals(
                 "http://myurl.localhost:8080/endpoint",
                 ApplicationPlaceholderResolver.resolveValue(context, "{{globals.open-api-url}}"));
     }
@@ -102,9 +104,8 @@ void testResolveSecretsInConfiguration() throws Exception {
         final Application resolved =
                 ApplicationPlaceholderResolver.resolvePlaceholders(applicationInstance);
         final Resource resource = resolved.getResources().get("openai-azure");
-        Assertions.assertEquals("my-access-key", resource.configuration().get("credentials"));
-        Assertions.assertEquals(
-                "http://myurl.localhost:8080/endpoint", resource.configuration().get("url"));
+        assertEquals("my-access-key", resource.configuration().get("credentials"));
+        assertEquals("http://myurl.localhost:8080/endpoint", resource.configuration().get("url"));
     }
 
     @Test
@@ -149,7 +150,7 @@ void testResolveInAgentConfiguration() throws Exception {
 
         final Application resolved =
                 ApplicationPlaceholderResolver.resolvePlaceholders(applicationInstance);
-        Assertions.assertEquals(
+        assertEquals(
                 "my-access-key",
                 resolved
                         .getModule("module-1")
@@ -164,7 +165,7 @@ void testResolveInAgentConfiguration() throws Exception {
                         .orElseThrow()
                         .getConfiguration()
                         .get("access-key"));
-        Assertions.assertEquals(
+        assertEquals(
                 "my-input-topic",
                 resolved
                         .getModule("module-1")
@@ -179,7 +180,7 @@ void testResolveInAgentConfiguration() throws Exception {
                         .orElseThrow()
                         .getInput()
                         .definition());
-        Assertions.assertEquals(
+        assertEquals(
                 "my-output-topic",
                 resolved
                         .getModule("module-1")
@@ -194,7 +195,7 @@ void testResolveInAgentConfiguration() throws Exception {
                         .orElseThrow()
                         .getOutput()
                         .definition());
-        Assertions.assertEquals(
+        assertEquals(
                 "my-stream-topic",
                 resolved
                         .getModule("module-1")
@@ -209,13 +210,13 @@ void testResolveInAgentConfiguration() throws Exception {
                         .orElseThrow()
                         .getConfiguration()
                         .get("stream-to-topic"));
-        Assertions.assertEquals(
+        assertEquals(
                 "my-stream-topic",
                 resolved.getModule("module-1").getTopics().get("my-stream-topic").getName());
-        Assertions.assertEquals(
+        assertEquals(
                 "my-input-topic",
                 resolved.getModule("module-1").getTopics().get("my-input-topic").getName());
-        Assertions.assertEquals(
+        assertEquals(
                 "my-output-topic",
                 resolved.getModule("module-1").getTopics().get("my-output-topic").getName());
     }
@@ -275,7 +276,7 @@ void testKeepStruct() throws Exception {
 
     @Test
     void testEscapeMustache() {
-        Assertions.assertEquals(
+        assertEquals(
                 """
                 {{ do not resolve }} resolved
                 {{# value.related_documents}}
@@ -289,4 +290,56 @@ void testEscapeMustache() {
                         {{% text}}
                         {{%/ value.related_documents}}"""));
     }
+
+    @Test
+    void testResolveTopicsInGateway() throws Exception {
+        Application applicationInstance =
+                ModelBuilder.buildApplicationInstance(
+                                Map.of(
+                                        "module1.yaml",
+                                        """
+                                        module: "module-1"
+                                        id: "pipeline-1"
+                                        topics:
+                                            - name: "{{{globals.input-topic}}}"
+                                            - name: "{{{globals.output-topic}}}"
+                                            - name: "{{{globals.stream-response-topic}}}"
+                                        pipeline:
+                                          - name: "agent1"
+                                            id: "agent1"
+                                            type: "ai-chat-completions"
+                                            input: "{{{globals.input-topic}}}"
+                                            output: "{{{globals.output-topic}}}"
+                                        """,
+                                        "gateways.yaml",
+                                        """
+                                        gateways:
+                                          - id: produce
+                                            type: produce
+                                            topic: "{{{globals.input-topic}}}"
+                                            events-topic: "{{{globals.stream-response-topic}}}"
+                                            produce-options: {}
+                                          - id: consume
+                                            type: consume
+                                            topic: "{{{globals.input-topic}}}"
+                                            events-topic: "{{{globals.stream-response-topic}}}"
+                                            consume-options: {}
+                                        """),
+                                """
+                                instance:
+                                    globals:
+                                        input-topic: my-input-topic
+                                        output-topic: my-output-topic
+                                        stream-response-topic: my-stream-topic
+                                """,
+                                null)
+                        .getApplication();
+
+        final Application resolved =
+                ApplicationPlaceholderResolver.resolvePlaceholders(applicationInstance);
+        assertEquals("my-input-topic", resolved.getGateways().gateways().get(0).topic());
+        assertEquals("my-stream-topic", resolved.getGateways().gateways().get(0).eventsTopic());
+        assertEquals("my-input-topic", resolved.getGateways().gateways().get(1).topic());
+        assertEquals("my-stream-topic", resolved.getGateways().gateways().get(1).eventsTopic());
+    }
 }