Repository: LangStream/langstream (language: python)

Issue:
Using syntax value.xx, key.xxx on a primitive schema has no effect
Example pipeline: 
```
pipeline:
# here is raw string
  - name: "Sample"
    type: "query"
    input: "in-topic"
    configuration:
      datasource: "xx"
      query: "SELECT id from XX"
      output-field: "value.myid"
```

If I send a message to `in-topic` with a raw string, the result of the query is ignored.

This might be expected and the solution is to prepend to `document-to-json` agent. 
But it's very confusing that there's no error/log anywhere and the result is just ignored. 

We should either throw an error (incompatible change) or at least add some logs




Discussion / hints:
The problem is only when using the syntax for output fields. If you use that syntax while reading you get a meaningful error message (I changed it in the Solr patch support) 
We can do the same for this case.

The case happens when the current object is not a Map or a JsonNode or a Avro generic record.


Fix patch:
diff --git a/langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgent.java b/langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgent.java
index 43b638d7e..8a40b130b 100644
--- a/langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgent.java
+++ b/langstream-agents/langstream-agent-http-request/src/main/java/ai/langstream/agents/azureblobstorage/source/HttpRequestAgent.java
@@ -16,7 +16,7 @@
 package ai.langstream.agents.azureblobstorage.source;
 
 import ai.langstream.ai.agents.commons.JsonRecord;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.api.runner.code.AbstractAgentCode;
 import ai.langstream.api.runner.code.AgentContext;
 import ai.langstream.api.runner.code.AgentProcessor;
@@ -134,7 +134,7 @@ public ComponentType componentType() {
 
     public void processRecord(Record record, RecordSink recordSink) {
         try {
-            TransformContext context = TransformContext.recordToTransformContext(record, true);
+            MutableRecord context = MutableRecord.recordToMutableRecord(record, true);
             final JsonRecord jsonRecord = context.toJsonRecord();
 
             final URI uri = URI.create(url + computeQueryString(jsonRecord));
@@ -174,7 +174,7 @@ public void processRecord(Record record, RecordSink recordSink) {
                                             avroKeySchemaCache,
                                             avroValueSchemaCache);
                                     Optional<Record> recordResult =
-                                            TransformContext.transformContextToRecord(context);
+                                            MutableRecord.mutableRecordToRecord(context);
                                     if (log.isDebugEnabled()) {
                                         log.debug("recordResult {}", recordResult);
                                     }
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/TransformContext.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/MutableRecord.java
similarity index 93%
rename from langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/TransformContext.java
rename to langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/MutableRecord.java
index 25a1ceb6c..a004ac886 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/TransformContext.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/MutableRecord.java
@@ -55,7 +55,7 @@
 
 @Slf4j
 @Data
-public class TransformContext {
+public class MutableRecord {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
     private TransformSchemaType keySchemaType;
     private Object keyNativeSchema;
@@ -73,8 +73,8 @@ public class TransformContext {
     // only for fn:filter
     private Object recordObject;
 
-    public TransformContext copy() {
-        TransformContext copy = new TransformContext();
+    public MutableRecord copy() {
+        MutableRecord copy = new MutableRecord();
 
         copy.keyObject = safeClone(keyObject);
         copy.valueObject = safeClone(valueObject);
@@ -204,6 +204,10 @@ public void addOrReplaceValueFields(
             addOrReplaceAvroValueFields(newFields, schemaCache);
         } else if (valueSchemaType == TransformSchemaType.JSON) {
             addOrReplaceJsonValueFields(newFields, schemaCache);
+        } else {
+            throw new IllegalArgumentException(
+                    "It is not possible to set fields on the value because the current value schema is "
+                            + valueSchemaType);
         }
     }
 
@@ -235,6 +239,10 @@ public void addOrReplaceKeyFields(
             addOrReplaceAvroKeyFields(newFields, schemaCache);
         } else if (keySchemaType == TransformSchemaType.JSON) {
             addOrReplaceJsonKeyFields(newFields, schemaCache);
+        } else {
+            throw new IllegalArgumentException(
+                    "It is not possible to set fields on the key because the current key schema is "
+                            + keySchemaType);
         }
     }
 
@@ -322,6 +330,10 @@ public void setResultField(
             if (valueObject instanceof Map) {
                 ((Map<String, Object>) valueObject).put(valueFieldName, content);
             } else {
+                if (fieldSchema == null) {
+                    throw new IllegalStateException(
+                            "Cannot set a value field without a schema on a non-map value");
+                }
                 Schema.Field fieldSchemaField =
                         new Schema.Field(valueFieldName, fieldSchema, null, null);
                 addOrReplaceValueFields(Map.of(fieldSchemaField, content), avroValueSchemaCache);
@@ -331,10 +343,19 @@ public void setResultField(
             if (keyObject instanceof Map) {
                 ((Map<String, Object>) keyObject).put(keyFieldName, content);
             } else {
+                if (fieldSchema == null) {
+                    throw new IllegalStateException(
+                            "Cannot set a key field without a schema on a non-map key");
+                }
                 Schema.Field fieldSchemaField =
                         new Schema.Field(keyFieldName, fieldSchema, null, null);
                 addOrReplaceKeyFields(Map.of(fieldSchemaField, content), avroKeySchemaCache);
             }
+        } else {
+            throw new IllegalArgumentException(
+                    "Cannot set field "
+                            + fieldName
+                            + ": it does not refer to any part of the message");
         }
     }
 
@@ -372,9 +393,9 @@ public static Object safeClone(Object object) {
         throw new UnsupportedOperationException("Cannot copy a value of " + object.getClass());
     }
 
-    public static TransformContext recordToTransformContext(
+    public static MutableRecord recordToMutableRecord(
             Record record, boolean attemptJsonConversion) {
-        TransformContext context = new TransformContext();
+        MutableRecord context = new MutableRecord();
         context.setKeyObject(record.key());
         context.setKeySchemaType(
                 record.key() == null ? null : getSchemaType(record.key().getClass()));
@@ -417,7 +438,7 @@ public static TransformContext recordToTransformContext(
         return context;
     }
 
-    public static Optional<Record> transformContextToRecord(TransformContext context) {
+    public static Optional<Record> mutableRecordToRecord(MutableRecord context) {
         if (context.isDropCurrentRecord()) {
             return Optional.empty();
         }
@@ -432,9 +453,9 @@ public static Optional<Record> transformContextToRecord(TransformContext context
         return Optional.of(new TransformRecord(context, headers));
     }
 
-    private record TransformRecord(TransformContext context, Collection<Header> headers)
+    private record TransformRecord(MutableRecord context, Collection<Header> headers)
             implements Record {
-        private TransformRecord(TransformContext context, Collection<Header> headers) {
+        private TransformRecord(MutableRecord context, Collection<Header> headers) {
             this.context = context;
             this.headers = new ArrayList<>(headers);
         }
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlEvaluator.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlEvaluator.java
index 9a2c1d587..779a60f26 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlEvaluator.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlEvaluator.java
@@ -15,7 +15,7 @@
  */
 package ai.langstream.ai.agents.commons.jstl;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.el.ELContext;
 import jakarta.el.ExpressionFactory;
@@ -175,6 +175,12 @@ private void registerFunctions() {
         this.expressionContext
                 .getFunctionMapper()
                 .mapFunction("fn", "now", JstlFunctions.class.getMethod("now"));
+        this.expressionContext
+                .getFunctionMapper()
+                .mapFunction("fn", "uuid", JstlFunctions.class.getMethod("uuid"));
+        this.expressionContext
+                .getFunctionMapper()
+                .mapFunction("fn", "random", JstlFunctions.class.getMethod("random", Object.class));
         this.expressionContext
                 .getFunctionMapper()
                 .mapFunction(
@@ -206,8 +212,8 @@ private void registerFunctions() {
                                 "dateadd", Object.class, Object.class, Object.class));
     }
 
-    public T evaluate(TransformContext transformContext) {
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+    public T evaluate(MutableRecord mutableRecord) {
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
         FACTORY.createValueExpression(expressionContext, "${key}", Object.class)
                 .setValue(expressionContext, adapter.getKey());
         FACTORY.createValueExpression(expressionContext, "${value}", Object.class)
@@ -234,7 +240,7 @@ public T evaluate(TransformContext transformContext) {
 
             // this is a very common error, so we provide a better error message
             if (expression.startsWith("value.")) {
-                Object valueObject = transformContext.getValueObject();
+                Object valueObject = mutableRecord.getValueObject();
                 if (valueObject instanceof String s) {
                     try {
                         new ObjectMapper().readValue(s, Object.class);
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlFunctions.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlFunctions.java
index 2f1f7d39c..287543774 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlFunctions.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlFunctions.java
@@ -15,7 +15,7 @@
  */
 package ai.langstream.ai.agents.commons.jstl;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.predicate.JstlPredicate;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.el.ELException;
@@ -30,6 +30,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.Setter;
@@ -255,7 +257,7 @@ public static List<Object> filter(Object input, String expression) {
             }
             // nulls are always filtered out
             if (o != null) {
-                TransformContext context = new TransformContext();
+                MutableRecord context = new MutableRecord();
                 context.setRecordObject(o);
                 boolean evaluate = predicate.test(context);
                 if (log.isDebugEnabled()) {
@@ -321,6 +323,18 @@ public static Instant now() {
         return Instant.now(clock);
     }
 
+    public static String uuid() {
+        return UUID.randomUUID().toString();
+    }
+
+    public static int random(Object max) {
+        if (max == null) {
+            throw new IllegalArgumentException("max cannot be null for fn:random(max)");
+        }
+        Integer maxValue = (Integer) toInt(max);
+        return ThreadLocalRandom.current().nextInt(maxValue);
+    }
+
     public static Instant timestampAdd(Object input, Object delta, Object unit) {
         if (input == null || unit == null) {
             throw new ELException(MessageFactory.get("error.method.notypes"));
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlTransformContextAdapter.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlTransformContextAdapter.java
index 5abd4b2c4..6b5cbeaa2 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlTransformContextAdapter.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlTransformContextAdapter.java
@@ -16,7 +16,7 @@
 package ai.langstream.ai.agents.commons.jstl;
 
 import ai.langstream.ai.agents.commons.AvroUtil;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -36,11 +36,11 @@
 import org.apache.commons.collections4.map.LazyMap;
 
 /**
- * A java bean that adapts the underlying {@link TransformContext} to be ready for jstl expression
+ * A java bean that adapts the underlying {@link MutableRecord} to be ready for jstl expression
  * language binding.
  */
 public class JstlTransformContextAdapter {
-    private TransformContext transformContext;
+    private MutableRecord mutableRecord;
 
     /**
      * A key transformer backing the lazy key map. It transforms top level key fields to either a
@@ -48,7 +48,7 @@ public class JstlTransformContextAdapter {
      */
     private final Transformer<String, Object> keyTransformer =
             (fieldName) -> {
-                Object keyObject = this.transformContext.getKeyObject();
+                Object keyObject = this.mutableRecord.getKeyObject();
                 if (keyObject instanceof GenericRecord) {
                     GenericRecord genericRecord = (GenericRecord) keyObject;
                     GenericRecordTransformer transformer =
@@ -57,7 +57,7 @@ public class JstlTransformContextAdapter {
                 }
                 if (keyObject instanceof JsonNode) {
                     JsonNode jsonNode = (JsonNode) keyObject;
-                    Schema schema = (Schema) this.transformContext.getKeyNativeSchema();
+                    Schema schema = (Schema) this.mutableRecord.getKeyNativeSchema();
                     JsonNodeTransformer transformer = new JsonNodeTransformer(jsonNode, schema);
                     return transformer.transform(fieldName);
                 }
@@ -72,7 +72,7 @@ public class JstlTransformContextAdapter {
      */
     private final Transformer<String, Object> valueTransformer =
             (fieldName) -> {
-                Object valueObject = this.transformContext.getValueObject();
+                Object valueObject = this.mutableRecord.getValueObject();
                 if (valueObject instanceof GenericRecord) {
                     GenericRecord genericRecord = (GenericRecord) valueObject;
                     GenericRecordTransformer transformer =
@@ -81,7 +81,7 @@ public class JstlTransformContextAdapter {
                 }
                 if (valueObject instanceof JsonNode) {
                     JsonNode jsonNode = (JsonNode) valueObject;
-                    Schema schema = (Schema) this.transformContext.getValueNativeSchema();
+                    Schema schema = (Schema) this.mutableRecord.getValueNativeSchema();
                     JsonNodeTransformer transformer = new JsonNodeTransformer(jsonNode, schema);
                     return transformer.transform(fieldName);
                 }
@@ -97,15 +97,15 @@ public class JstlTransformContextAdapter {
                 // Allow list message headers in the expression
                 switch (fieldName) {
                     case "messageKey":
-                        return transformContext.getKey();
+                        return mutableRecord.getKey();
                     case "topicName":
-                        return transformContext.getInputTopic();
+                        return mutableRecord.getInputTopic();
                     case "destinationTopic":
-                        return transformContext.getOutputTopic();
+                        return mutableRecord.getOutputTopic();
                     case "eventTime":
-                        return transformContext.getEventTime();
+                        return mutableRecord.getEventTime();
                     case "properties":
-                        return transformContext.getProperties();
+                        return mutableRecord.getProperties();
                     default:
                         return null;
                 }
@@ -114,8 +114,8 @@ public class JstlTransformContextAdapter {
     private final Map<String, Object> lazyHeader =
             LazyMap.lazyMap(new HashMap<>(), headerTransformer);
 
-    public JstlTransformContextAdapter(TransformContext transformContext) {
-        this.transformContext = transformContext;
+    public JstlTransformContextAdapter(MutableRecord mutableRecord) {
+        this.mutableRecord = mutableRecord;
     }
 
     /**
@@ -123,9 +123,9 @@ public JstlTransformContextAdapter(TransformContext transformContext) {
      *     object, or the primitive type itself.
      */
     public Object getKey() {
-        Object keyObject = this.transformContext.getKeyObject();
+        Object keyObject = this.mutableRecord.getKeyObject();
         if (keyObject == null) {
-            return transformContext.getKey();
+            return mutableRecord.getKey();
         }
         return keyObject instanceof GenericRecord || keyObject instanceof JsonNode
                 ? lazyKey
@@ -137,14 +137,14 @@ public Object getKey() {
      *     object, or the primitive type itself.
      */
     public Object adaptValue() {
-        Object valueObject = this.transformContext.getValueObject();
+        Object valueObject = this.mutableRecord.getValueObject();
         return valueObject instanceof GenericRecord || valueObject instanceof JsonNode
                 ? lazyValue
                 : valueObject;
     }
 
     public Object adaptRecord() {
-        return transformContext.getRecordObject();
+        return mutableRecord.getRecordObject();
     }
 
     public Map<String, Object> getHeader() {
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/JstlPredicate.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/JstlPredicate.java
index bfc3da421..e1756dbf5 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/JstlPredicate.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/JstlPredicate.java
@@ -15,7 +15,7 @@
  */
 package ai.langstream.ai.agents.commons.jstl.predicate;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import jakarta.el.ELException;
 import jakarta.el.PropertyNotFoundException;
@@ -36,9 +36,9 @@ public JstlPredicate(String when) {
     }
 
     @Override
-    public boolean test(TransformContext transformContext) {
+    public boolean test(MutableRecord mutableRecord) {
         try {
-            return this.evaluator.evaluate(transformContext);
+            return this.evaluator.evaluate(mutableRecord);
         } catch (PropertyNotFoundException ex) {
             log.warn("a property in the when expression was not found in the message", ex);
             return false;
diff --git a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/TransformPredicate.java b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/TransformPredicate.java
index bcdd4f424..57ae34c54 100644
--- a/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/TransformPredicate.java
+++ b/langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/predicate/TransformPredicate.java
@@ -15,13 +15,13 @@
  */
 package ai.langstream.ai.agents.commons.jstl.predicate;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import java.util.function.Predicate;
 
 /**
- * A predicate functional interface that applies to {@link TransformContext}. Implementations of
- * this interface should respect the current record encapsulated in the {@link TransformContext}
- * when evaluating the predicate.
+ * A predicate functional interface that applies to {@link MutableRecord}. Implementations of this
+ * interface should respect the current record encapsulated in the {@link MutableRecord} when
+ * evaluating the predicate.
  */
 @FunctionalInterface
-public interface TransformPredicate extends Predicate<TransformContext> {}
+public interface TransformPredicate extends Predicate<MutableRecord> {}
diff --git a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/DispatchAgent.java b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/DispatchAgent.java
index 217982644..f7cd925db 100644
--- a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/DispatchAgent.java
+++ b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/DispatchAgent.java
@@ -15,7 +15,7 @@
  */
 package ai.langstream.agents.flow;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.predicate.JstlPredicate;
 import ai.langstream.api.runner.code.AbstractAgentCode;
 import ai.langstream.api.runner.code.AgentContext;
@@ -113,7 +113,7 @@ public ComponentType componentType() {
 
     public void processRecord(Record record, RecordSink recordSink) {
         try {
-            TransformContext context = TransformContext.recordToTransformContext(record, true);
+            MutableRecord context = MutableRecord.recordToMutableRecord(record, true);
 
             for (Route r : routes) {
                 boolean test = r.predicate.test(context);
diff --git a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/FlowControlAgentsCodeProvider.java b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/FlowControlAgentsCodeProvider.java
index 47cb232ec..31af49ed6 100644
--- a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/FlowControlAgentsCodeProvider.java
+++ b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/FlowControlAgentsCodeProvider.java
@@ -15,15 +15,21 @@
  */
 package ai.langstream.agents.flow;
 
+import ai.langstream.api.runner.code.AgentCode;
 import ai.langstream.api.runner.code.AgentCodeProvider;
-import ai.langstream.api.runner.code.AgentProcessor;
 import java.util.Map;
 import java.util.function.Supplier;
 
 public class FlowControlAgentsCodeProvider implements AgentCodeProvider {
 
-    private static final Map<String, Supplier<AgentProcessor>> FACTORIES =
-            Map.of("dispatch", DispatchAgent::new);
+    private static final Map<String, Supplier<AgentCode>> FACTORIES =
+            Map.of(
+                    "dispatch",
+                    DispatchAgent::new,
+                    "trigger-event",
+                    TriggerEventProcessor::new,
+                    "timer-source",
+                    TimerSource::new);
 
     @Override
     public boolean supports(String agentType) {
@@ -31,7 +37,7 @@ public boolean supports(String agentType) {
     }
 
     @Override
-    public AgentProcessor createInstance(String agentType) {
+    public AgentCode createInstance(String agentType) {
         return FACTORIES.get(agentType).get();
     }
 }
diff --git a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TimerSource.java b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TimerSource.java
new file mode 100644
index 000000000..ec2111540
--- /dev/null
+++ b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TimerSource.java
@@ -0,0 +1,128 @@
+/*
+ * 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.agents.flow;
+
+import ai.langstream.ai.agents.commons.MutableRecord;
+import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
+import ai.langstream.api.runner.code.AbstractAgentCode;
+import ai.langstream.api.runner.code.AgentSource;
+import ai.langstream.api.runner.code.Record;
+import ai.langstream.api.runner.code.SimpleRecord;
+import ai.langstream.api.util.ConfigurationUtils;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TimerSource extends AbstractAgentCode implements AgentSource {
+
+    record FieldDefinition(String name, JstlEvaluator<Object> expressionEvaluator) {}
+
+    private static final int MAX_QUEUE_SIZE = 2;
+    private final List<FieldDefinition> fields = new ArrayList<>();
+    private final BlockingQueue<Record> queue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
+    private int periodSeconds;
+    private ScheduledExecutorService executorService;
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public void init(Map<String, Object> configuration) {
+        periodSeconds = ConfigurationUtils.getInt("period-seconds", 60, configuration);
+        List<Map<String, Object>> fields =
+                (List<Map<String, Object>>) configuration.getOrDefault("fields", List.of());
+        fields.forEach(
+                r -> {
+                    String name = ConfigurationUtils.getString("name", "", r);
+                    String expression = ConfigurationUtils.getString("expression", "", r);
+                    log.info("Emitting field with name {} computed as {}", name, expression);
+                    JstlEvaluator<Object> expressionEvaluator =
+                            new JstlEvaluator<>("${" + expression + "}", Object.class);
+                    this.fields.add(new FieldDefinition(name, expressionEvaluator));
+                });
+        executorService =
+                Executors.newSingleThreadScheduledExecutor(
+                        new ThreadFactory() {
+                            @Override
+                            public Thread newThread(Runnable r) {
+                                return new Thread(r, "timer-source-" + agentId());
+                            }
+                        });
+    }
+
+    private void emitRecord() {
+        try {
+            MutableRecord mutableRecord =
+                    MutableRecord.recordToMutableRecord(SimpleRecord.of("{}", "{}"), true);
+            Map<String, Object> values = new HashMap<>();
+            for (FieldDefinition field : fields) {
+                values.put(field.name, field.expressionEvaluator.evaluate(mutableRecord));
+            }
+            values.forEach(
+                    (fieldName, fieldValue) -> {
+                        mutableRecord.setResultField(fieldValue, fieldName, null, null, null);
+                    });
+            Record record = MutableRecord.mutableRecordToRecord(mutableRecord).orElseThrow();
+            queue.put(record);
+            log.info("Generated record {}", record);
+        } catch (Throwable e) {
+            log.error("Error while emitting record", e);
+        }
+    }
+
+    @Override
+    public void start() throws Exception {
+        executorService.scheduleAtFixedRate(
+                this::emitRecord, periodSeconds, periodSeconds, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public List<Record> read() throws Exception {
+        try {
+            // this is a blocking operation, the main loop will be blocked until a record is
+            // available, but we want to let the agent shutdown gracefully in case there it nothing
+            // to serve
+            Record taken = queue.poll(500, TimeUnit.MILLISECONDS);
+            if (taken == null) {
+                return List.of();
+            }
+            return List.of(taken);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return List.of();
+        }
+    }
+
+    @Override
+    public void commit(List<Record> records) throws Exception {
+        // ignore
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (executorService != null) {
+            executorService.shutdown();
+            executorService = null;
+        }
+    }
+}
diff --git a/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TriggerEventProcessor.java b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TriggerEventProcessor.java
new file mode 100644
index 000000000..ebcfbc813
--- /dev/null
+++ b/langstream-agents/langstream-agents-flow-control/src/main/java/ai/langstream/agents/flow/TriggerEventProcessor.java
@@ -0,0 +1,155 @@
+/*
+ * 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.agents.flow;
+
+import ai.langstream.ai.agents.commons.MutableRecord;
+import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
+import ai.langstream.ai.agents.commons.jstl.predicate.JstlPredicate;
+import ai.langstream.api.runner.code.AbstractAgentCode;
+import ai.langstream.api.runner.code.AgentContext;
+import ai.langstream.api.runner.code.AgentProcessor;
+import ai.langstream.api.runner.code.Record;
+import ai.langstream.api.runner.code.RecordSink;
+import ai.langstream.api.runner.topics.TopicProducer;
+import ai.langstream.api.runtime.ComponentType;
+import ai.langstream.api.util.ConfigurationUtils;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TriggerEventProcessor extends AbstractAgentCode implements AgentProcessor {
+
+    record FieldDefinition(String name, JstlEvaluator<Object> expressionEvaluator) {}
+
+    private final List<FieldDefinition> fields = new ArrayList<>();
+    private TopicProducer topicProducer;
+    private String destination;
+    private JstlPredicate predicate;
+    private AgentContext agentContext;
+
+    private boolean continueProcessing;
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public void init(Map<String, Object> configuration) {
+        destination = ConfigurationUtils.getString("destination", "", configuration);
+        String when = ConfigurationUtils.getString("when", "true", configuration);
+        continueProcessing =
+                ConfigurationUtils.getBoolean("continue-processing", true, configuration);
+        predicate = new JstlPredicate(when);
+        List<Map<String, Object>> fields =
+                (List<Map<String, Object>>) configuration.getOrDefault("fields", List.of());
+        fields.forEach(
+                r -> {
+                    String name = ConfigurationUtils.getString("name", "", r);
+                    String expression = ConfigurationUtils.getString("expression", "", r);
+                    log.info("Emitting field with name {} computed as {}", name, expression);
+                    JstlEvaluator<Object> expressionEvaluator =
+                            new JstlEvaluator<>("${" + expression + "}", Object.class);
+                    this.fields.add(new FieldDefinition(name, expressionEvaluator));
+                });
+    }
+
+    @Override
+    public ComponentType componentType() {
+        return ComponentType.PROCESSOR;
+    }
+
+    @Override
+    public void setContext(AgentContext context) throws Exception {
+        this.agentContext = context;
+    }
+
+    private Record emitRecord(Record originalRecord) {
+        MutableRecord mutableRecord =
+                MutableRecord.recordToMutableRecord(originalRecord, true).copy();
+        if (!predicate.test(mutableRecord)) {
+            return null;
+        }
+        Map<String, Object> values = new HashMap<>();
+        for (FieldDefinition field : fields) {
+            values.put(field.name, field.expressionEvaluator.evaluate(mutableRecord));
+        }
+        values.forEach(
+                (fieldName, fieldValue) -> {
+                    mutableRecord.setResultField(fieldValue, fieldName, null, null, null);
+                });
+        Record record = MutableRecord.mutableRecordToRecord(mutableRecord).orElseThrow();
+        log.info("Generated record {}", record);
+        return record;
+    }
+
+    @Override
+    public void process(List<Record> records, RecordSink recordSink) {
+        for (Record r : records) {
+            processRecordAsync(r, recordSink);
+        }
+    }
+
+    private void processRecordAsync(Record r, RecordSink recordSink) {
+        Record newRecord;
+        try {
+            newRecord = emitRecord(r);
+            if (newRecord == null) {
+                log.info("Record {} does not match the predicate, skipping", r);
+                // nothing to do
+                recordSink.emitSingleResult(r, r);
+                return;
+            }
+        } catch (RuntimeException error) {
+            recordSink.emitError(r, error);
+            return;
+        }
+        topicProducer
+                .write(newRecord)
+                .whenComplete(
+                        (v, e) -> {
+                            if (e != null) {
+                                log.error("Error while writing record to topic", e);
+                                recordSink.emitError(r, e);
+                            } else {
+                                if (continueProcessing) {
+                                    // pass the source record as the result
+                                    // the new record has been written to the side topic
+                                    recordSink.emitSingleResult(r, r);
+                                } else {
+                                    // stop processing the message
+                                    recordSink.emitEmptyList(r);
+                                }
+                            }
+                        });
+    }
+
+    @Override
+    public void start() throws Exception {
+        topicProducer =
+                agentContext
+                        .getTopicConnectionProvider()
+                        .createProducer(agentContext.getGlobalAgentId(), destination, Map.of());
+        topicProducer.start();
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (topicProducer != null) {
+            topicProducer.close();
+            topicProducer = null;
+        }
+    }
+}
diff --git a/langstream-agents/langstream-agents-flow-control/src/main/resources/META-INF/ai.langstream.agents.index b/langstream-agents/langstream-agents-flow-control/src/main/resources/META-INF/ai.langstream.agents.index
index 554505981..5576be744 100644
--- a/langstream-agents/langstream-agents-flow-control/src/main/resources/META-INF/ai.langstream.agents.index
+++ b/langstream-agents/langstream-agents-flow-control/src/main/resources/META-INF/ai.langstream.agents.index
@@ -1,1 +1,3 @@
-dispatch
\ No newline at end of file
+dispatch
+trigger-event
+timer-source
\ No newline at end of file
diff --git a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TextSplitterAgent.java b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TextSplitterAgent.java
index bca1f7e7a..66098d4ed 100644
--- a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TextSplitterAgent.java
+++ b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TextSplitterAgent.java
@@ -96,6 +96,7 @@ public List<Record> processRecord(Record record) {
         String text = Utils.toText(value);
         List<String> chunks = textSplitter.splitText(text);
         int chunkId = 0;
+        int numChunks = chunks.size();
         List<Record> result = new ArrayList<>();
         for (String chunk : chunks) {
             List<Header> chunkHeaders = new ArrayList<>(record.headers());
@@ -106,6 +107,8 @@ public List<Record> processRecord(Record record) {
             int numTokens = lengthFunction.length(chunk);
             chunkHeaders.add(
                     new SimpleRecord.SimpleHeader("chunk_num_tokens", String.valueOf(numTokens)));
+            chunkHeaders.add(
+                    new SimpleRecord.SimpleHeader("text_num_chunks", String.valueOf(numChunks)));
 
             // here we are setting as key the original key
             // this is to ensure that all the chunks will be processed in order downstream
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/GenAIToolKitAgent.java b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/GenAIToolKitAgent.java
index 4516f7e7d..c74048e64 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/GenAIToolKitAgent.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/GenAIToolKitAgent.java
@@ -15,10 +15,10 @@
  */
 package ai.langstream.ai.agents;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
-import static ai.langstream.ai.agents.commons.TransformContext.transformContextToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.mutableRecordToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.datasource.DataSourceProviderRegistry;
 import ai.langstream.ai.agents.services.ServiceProviderRegistry;
 import ai.langstream.api.runner.code.AbstractAgentCode;
@@ -98,15 +98,14 @@ public CompletableFuture<List<Record>> processRecord(Record record) {
         if (log.isDebugEnabled()) {
             log.debug("Processing {}", record);
         }
-        TransformContext context =
-                recordToTransformContext(record, config.isAttemptJsonConversion());
+        MutableRecord context = recordToMutableRecord(record, config.isAttemptJsonConversion());
 
         CompletableFuture<?> handle = processStep(context, step);
         return handle.thenApply(
                 ___ -> {
                     try {
                         context.convertMapToStringOrBytes();
-                        Optional<Record> recordResult = transformContextToRecord(context);
+                        Optional<Record> recordResult = mutableRecordToRecord(context);
                         if (log.isDebugEnabled()) {
                             log.debug("Result {}", recordResult);
                         }
@@ -119,11 +118,11 @@ public CompletableFuture<List<Record>> processRecord(Record record) {
     }
 
     private static CompletableFuture<?> processStep(
-            TransformContext transformContext, StepPredicatePair pair) {
+            MutableRecord mutableRecord, StepPredicatePair pair) {
         TransformStep step = pair.getTransformStep();
-        Predicate<TransformContext> predicate = pair.getPredicate();
-        if (predicate == null || predicate.test(transformContext)) {
-            return step.processAsync(transformContext);
+        Predicate<MutableRecord> predicate = pair.getPredicate();
+        if (predicate == null || predicate.test(mutableRecord)) {
+            return step.processAsync(mutableRecord);
         } else {
             return CompletableFuture.completedFuture(null);
         }
@@ -208,8 +207,8 @@ public TopicStreamingAnswersConsumer(TopicProducer topicProducer) {
 
         @Override
         public void streamAnswerChunk(
-                int index, String message, boolean last, TransformContext outputMessage) {
-            Optional<Record> record = transformContextToRecord(outputMessage);
+                int index, String message, boolean last, MutableRecord outputMessage) {
+            Optional<Record> record = mutableRecordToRecord(outputMessage);
             if (record.isPresent()) {
                 if (log.isDebugEnabled()) {
                     log.debug(
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/flare/FlareControllerAgent.java b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/flare/FlareControllerAgent.java
index ca11dec90..7a12c8497 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/flare/FlareControllerAgent.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/flare/FlareControllerAgent.java
@@ -15,10 +15,10 @@
  */
 package ai.langstream.ai.agents.flare;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
-import static ai.langstream.ai.agents.commons.TransformContext.transformContextToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.mutableRecordToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.runner.code.AbstractAgentCode;
 import ai.langstream.api.runner.code.AgentCode;
@@ -117,9 +117,9 @@ public void processRecord(Record record, RecordSink recordSink) {
             recordSink.emitEmptyList(record);
             return;
         }
-        TransformContext transformContext = recordToTransformContext(record, true).copy();
+        MutableRecord mutableRecord = recordToMutableRecord(record, true).copy();
 
-        Integer iterationCount = numIterationsAccessor.evaluate(transformContext);
+        Integer iterationCount = numIterationsAccessor.evaluate(mutableRecord);
         if (iterationCount == null) {
             iterationCount = 0;
         }
@@ -130,9 +130,9 @@ public void processRecord(Record record, RecordSink recordSink) {
         } else {
             log.info("Record did {} iterations in Flare controller", iterationCount);
         }
-        List<String> tokens = tokensAccessor.evaluate(transformContext);
+        List<String> tokens = tokensAccessor.evaluate(mutableRecord);
         log.info("Flare: tokens: {}", tokens);
-        List<Double> logProbs = logsProbsAccessor.evaluate(transformContext);
+        List<Double> logProbs = logsProbsAccessor.evaluate(mutableRecord);
 
         List<String> logConfidenceSpans =
                 lowConfidenceSpans(tokens, logProbs, minProb, minTokenGap, numPadTokens);
@@ -149,14 +149,14 @@ public void processRecord(Record record, RecordSink recordSink) {
         }
         log.info("The LLM needs more content about these spans: {}", logConfidenceSpans);
 
-        transformContext.setResultField(
+        mutableRecord.setResultField(
                 logConfidenceSpans,
                 retrieveDocumentsField,
                 Schema.createArray(Schema.create(Schema.Type.STRING)),
                 avroKeySchemaCache,
                 avroValueSchemaCache);
 
-        Record forLoopTopic = transformContextToRecord(transformContext).orElseThrow();
+        Record forLoopTopic = mutableRecordToRecord(mutableRecord).orElseThrow();
 
         loopTopicProducer
                 .write(forLoopTopic)
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/rerank/ReRankAgent.java b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/rerank/ReRankAgent.java
index e14524902..cca2fe7ef 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/rerank/ReRankAgent.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/ai/langstream/ai/agents/rerank/ReRankAgent.java
@@ -15,10 +15,10 @@
  */
 package ai.langstream.ai.agents.rerank;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
-import static ai.langstream.ai.agents.commons.TransformContext.transformContextToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.mutableRecordToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.runner.code.Record;
 import ai.langstream.api.runner.code.SingleRecordAgentProcessor;
@@ -101,18 +101,18 @@ public List<Record> processRecord(Record record) throws Exception {
         if (record == null) {
             return List.of();
         }
-        TransformContext transformContext = recordToTransformContext(record, true).copy();
+        MutableRecord mutableRecord = recordToMutableRecord(record, true).copy();
 
-        List<Object> currentList = (List<Object>) fieldAccessor.evaluate(transformContext);
+        List<Object> currentList = (List<Object>) fieldAccessor.evaluate(mutableRecord);
 
         TextWithEmbeddings query =
                 new TextWithEmbeddings(
-                        queryAccessor.evaluate(transformContext),
-                        toArrayOfFloat(queryEmbeddingsAccessor.evaluate(transformContext)));
+                        queryAccessor.evaluate(mutableRecord),
+                        toArrayOfFloat(queryEmbeddingsAccessor.evaluate(mutableRecord)));
 
         Function<Object, TextWithEmbeddings> recordExtractor =
                 (Object o) -> {
-                    TransformContext context = new TransformContext();
+                    MutableRecord context = new MutableRecord();
                     context.setRecordObject(o);
                     String text = recordTextAccessor.evaluate(context);
                     float[] embeddings = toArrayOfFloat(recordEmbeddingsAccessor.evaluate(context));
@@ -123,10 +123,10 @@ public List<Record> processRecord(Record record) throws Exception {
                     return new TextWithEmbeddings(text, embeddings);
                 };
         List<Object> result = rerank(currentList, query, recordExtractor);
-        transformContext.setResultField(
+        mutableRecord.setResultField(
                 result, outputField, null, avroKeySchemaCache, avroValueSchemaCache);
-        transformContext.convertMapToStringOrBytes();
-        Optional<Record> recordResult = transformContextToRecord(transformContext);
+        mutableRecord.convertMapToStringOrBytes();
+        Optional<Record> recordResult = mutableRecordToRecord(mutableRecord);
         return recordResult.map(List::of).orElse(List.of());
     }
 
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/CastStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/CastStep.java
index 23d97e5b8..81b205ace 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/CastStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/CastStep.java
@@ -15,9 +15,9 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import static ai.langstream.ai.agents.commons.TransformContext.attemptJsonConversion;
+import static ai.langstream.ai.agents.commons.MutableRecord.attemptJsonConversion;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import ai.langstream.ai.agents.commons.jstl.JstlTypeConverter;
 import java.sql.Time;
@@ -37,18 +37,18 @@ public class CastStep implements TransformStep {
     private final boolean attemptJsonConversion;
 
     @Override
-    public void process(TransformContext transformContext) {
-        if (transformContext.getKeySchemaType() != null
+    public void process(MutableRecord mutableRecord) {
+        if (mutableRecord.getKeySchemaType() != null
                 && keySchemaType != null
-                && transformContext.getKeySchemaType() != keySchemaType) {
-            Object value = convertValue(transformContext.getKeyObject(), keySchemaType);
-            transformContext.setKeySchemaType(keySchemaType);
-            transformContext.setKeyObject(value);
+                && mutableRecord.getKeySchemaType() != keySchemaType) {
+            Object value = convertValue(mutableRecord.getKeyObject(), keySchemaType);
+            mutableRecord.setKeySchemaType(keySchemaType);
+            mutableRecord.setKeyObject(value);
         }
-        if (valueSchemaType != null && transformContext.getValueSchemaType() != valueSchemaType) {
-            Object value = convertValue(transformContext.getValueObject(), valueSchemaType);
-            transformContext.setValueSchemaType(valueSchemaType);
-            transformContext.setValueObject(value);
+        if (valueSchemaType != null && mutableRecord.getValueSchemaType() != valueSchemaType) {
+            Object value = convertValue(mutableRecord.getValueObject(), valueSchemaType);
+            mutableRecord.setValueSchemaType(valueSchemaType);
+            mutableRecord.setValueObject(value);
         }
     }
 
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ChatCompletionsStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ChatCompletionsStep.java
index 020bb5f89..90de48c29 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ChatCompletionsStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ChatCompletionsStep.java
@@ -18,7 +18,7 @@
 import static com.datastax.oss.streaming.ai.util.TransformFunctionUtil.convertToMap;
 
 import ai.langstream.ai.agents.commons.JsonRecord;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import com.datastax.oss.streaming.ai.completions.ChatChoice;
 import com.datastax.oss.streaming.ai.completions.ChatCompletions;
 import com.datastax.oss.streaming.ai.completions.ChatMessage;
@@ -66,7 +66,7 @@ public void streamAnswerChunk(
                                 int index,
                                 String message,
                                 boolean last,
-                                TransformContext outputMessage) {
+                                MutableRecord outputMessage) {
                             log.info("index: {}, message: {}, last: {}", index, message, last);
                         }
                     };
@@ -112,8 +112,8 @@ public void close() throws Exception {
     }
 
     @Override
-    public CompletableFuture<?> processAsync(TransformContext transformContext) {
-        JsonRecord jsonRecord = transformContext.toJsonRecord();
+    public CompletableFuture<?> processAsync(MutableRecord mutableRecord) {
+        JsonRecord jsonRecord = mutableRecord.toJsonRecord();
 
         List<ChatMessage> messages =
                 config.getMessages().stream()
@@ -140,7 +140,7 @@ public void consumeChunk(
                                 // we must copy the context because the same context is used for all
                                 // chunks
                                 // and also for the final answer
-                                TransformContext copy = transformContext.copy();
+                                MutableRecord copy = mutableRecord.copy();
 
                                 copy.getProperties().put("stream-id", answerId);
                                 copy.getProperties().put("stream-index", index + "");
@@ -157,7 +157,7 @@ public void consumeChunk(
         return chatCompletionsHandle.thenApply(
                 chatCompletions -> {
                     ChatChoice chatChoice = chatCompletions.getChoices().get(0);
-                    applyResultFieldToContext(transformContext, chatChoice.content(), false);
+                    applyResultFieldToContext(mutableRecord, chatChoice.content(), false);
 
                     String logField = config.getLogField();
                     if (logField != null && !logField.isEmpty()) {
@@ -165,8 +165,8 @@ public void consumeChunk(
                         logMap.put("model", config.getModel());
                         logMap.put("options", options);
                         logMap.put("messages", messages);
-                        transformContext.setResultField(
-                                TransformContext.toJson(logMap),
+                        mutableRecord.setResultField(
+                                MutableRecord.toJson(logMap),
                                 logField,
                                 org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING),
                                 avroKeySchemaCache,
@@ -177,7 +177,7 @@ public void consumeChunk(
     }
 
     private void applyResultFieldToContext(
-            TransformContext transformContext, String content, boolean streamingAnswer) {
+            MutableRecord mutableRecord, String content, boolean streamingAnswer) {
         String fieldName = config.getFieldName();
 
         // maybe we want a different field in the streaming answer
@@ -187,7 +187,7 @@ private void applyResultFieldToContext(
                 && !config.getStreamResponseCompletionField().isEmpty()) {
             fieldName = config.getStreamResponseCompletionField();
         }
-        transformContext.setResultField(
+        mutableRecord.setResultField(
                 content,
                 fieldName,
                 Schema.create(Schema.Type.STRING),
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsStep.java
index d30c6d127..cf47223bf 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsStep.java
@@ -16,7 +16,7 @@
 package com.datastax.oss.streaming.ai;
 
 import ai.langstream.ai.agents.commons.JsonRecord;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.util.OrderedAsyncBatchExecutor;
 import com.datastax.oss.streaming.ai.embeddings.EmbeddingsService;
@@ -106,7 +106,7 @@ public ComputeAIEmbeddingsStep(
     }
 
     private static int computeHashForRecord(RecordHolder record) {
-        Object key = record.transformContext.getKeyObject();
+        Object key = record.mutableRecord.getKeyObject();
         if (key != null) {
             return Objects.hashCode(key);
         } else {
@@ -129,8 +129,8 @@ private void processBatch(List<RecordHolder> records, CompletableFuture<?> compl
 
         try {
             for (RecordHolder holder : records) {
-                TransformContext transformContext = holder.transformContext();
-                JsonRecord jsonRecord = transformContext.toJsonRecord();
+                MutableRecord mutableRecord = holder.mutableRecord();
+                JsonRecord jsonRecord = mutableRecord.toJsonRecord();
                 if (loopOverAccessor == null) {
                     String text = template.execute(jsonRecord);
                     texts.add(text);
@@ -138,7 +138,7 @@ private void processBatch(List<RecordHolder> records, CompletableFuture<?> compl
                             new TextAndReference(
                                     text,
                                     (List<Double> embeddingsForText) -> {
-                                        transformContext.setResultField(
+                                        mutableRecord.setResultField(
                                                 embeddingsForText,
                                                 embeddingsFieldName,
                                                 Schema.createArray(
@@ -148,7 +148,7 @@ private void processBatch(List<RecordHolder> records, CompletableFuture<?> compl
                                         holder.handle().complete(null);
                                     }));
                 } else {
-                    List<Object> nestedRecords = loopOverAccessor.evaluate(transformContext);
+                    List<Object> nestedRecords = loopOverAccessor.evaluate(mutableRecord);
                     List<Map<String, Object>> newList = new CopyOnWriteArrayList<>();
                     AtomicInteger remaining = new AtomicInteger(nestedRecords.size());
 
@@ -174,7 +174,7 @@ private void processBatch(List<RecordHolder> records, CompletableFuture<?> compl
                                                 log.info("final list {}", newList);
                                                 // all the nested records are done, we can override
                                                 // the field, in the original record
-                                                transformContext.setResultField(
+                                                mutableRecord.setResultField(
                                                         newList,
                                                         loopOverFieldName,
                                                         Schema.createArray(
@@ -239,12 +239,12 @@ public void close() throws Exception {
         }
     }
 
-    record RecordHolder(TransformContext transformContext, CompletableFuture<?> handle) {}
+    record RecordHolder(MutableRecord mutableRecord, CompletableFuture<?> handle) {}
 
     @Override
-    public CompletableFuture<?> processAsync(TransformContext transformContext) {
+    public CompletableFuture<?> processAsync(MutableRecord mutableRecord) {
         CompletableFuture<?> handle = new CompletableFuture<>();
-        batchExecutor.add(new RecordHolder(transformContext, handle));
+        batchExecutor.add(new RecordHolder(mutableRecord, handle));
         return handle;
     }
 }
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeStep.java
index 634ac4752..5ea87c7a6 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/ComputeStep.java
@@ -16,7 +16,7 @@
 package com.datastax.oss.streaming.ai;
 
 import ai.langstream.ai.agents.commons.AvroUtil;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import ai.langstream.ai.agents.commons.jstl.JstlTypeConverter;
 import com.datastax.oss.streaming.ai.model.ComputeField;
@@ -56,35 +56,35 @@ public class ComputeStep implements TransformStep {
             new ConcurrentHashMap<>();
 
     @Override
-    public void process(TransformContext transformContext) {
+    public void process(MutableRecord mutableRecord) {
         computePrimitiveField(
                 fields.stream()
                         .filter(f -> "primitive".equals(f.getScope()))
                         .collect(Collectors.toList()),
-                transformContext);
+                mutableRecord);
         computeKeyFields(
                 fields.stream()
                         .filter(f -> "key".equals(f.getScope()))
                         .collect(Collectors.toList()),
-                transformContext);
+                mutableRecord);
         computeValueFields(
                 fields.stream()
                         .filter(f -> "value".equals(f.getScope()))
                         .collect(Collectors.toList()),
-                transformContext);
+                mutableRecord);
         computeHeaderFields(
                 fields.stream()
                         .filter(f -> "header".equals(f.getScope()))
                         .collect(Collectors.toList()),
-                transformContext);
+                mutableRecord);
         computeHeaderPropertiesFields(
                 fields.stream()
                         .filter(f -> "header.properties".equals(f.getScope()))
                         .collect(Collectors.toList()),
-                transformContext);
+                mutableRecord);
     }
 
-    public void computeValueFields(List<ComputeField> fields, TransformContext context) {
+    public void computeValueFields(List<ComputeField> fields, MutableRecord context) {
         TransformSchemaType schemaType = context.getValueSchemaType();
         if (context.getValueObject() instanceof Map) {
             getEvaluatedFields(fields, context)
@@ -98,7 +98,7 @@ public void computeValueFields(List<ComputeField> fields, TransformContext conte
         }
     }
 
-    public void computePrimitiveField(List<ComputeField> fields, TransformContext context) {
+    public void computePrimitiveField(List<ComputeField> fields, MutableRecord context) {
         fields.stream()
                 .filter(f -> "key".equals(f.getName()))
                 .findFirst()
@@ -137,7 +137,7 @@ public void computePrimitiveField(List<ComputeField> fields, TransformContext co
                         });
     }
 
-    public void computeKeyFields(List<ComputeField> fields, TransformContext context) {
+    public void computeKeyFields(List<ComputeField> fields, MutableRecord context) {
         Object keyObject = context.getKeyObject();
         if (keyObject != null) {
             if (keyObject instanceof Map) {
@@ -154,7 +154,7 @@ public void computeKeyFields(List<ComputeField> fields, TransformContext context
         }
     }
 
-    public void computeHeaderFields(List<ComputeField> fields, TransformContext context) {
+    public void computeHeaderFields(List<ComputeField> fields, MutableRecord context) {
         fields.forEach(
                 field -> {
                     switch (field.getName()) {
@@ -173,13 +173,13 @@ public void computeHeaderFields(List<ComputeField> fields, TransformContext cont
                 });
     }
 
-    public void computeHeaderPropertiesFields(List<ComputeField> fields, TransformContext context) {
+    public void computeHeaderPropertiesFields(List<ComputeField> fields, MutableRecord context) {
         fields.forEach(
                 field ->
                         context.setProperty(field.getName(), validateAndGetString(field, context)));
     }
 
-    private String validateAndGetString(ComputeField field, TransformContext context) {
+    private String validateAndGetString(ComputeField field, MutableRecord context) {
         Object value = field.getEvaluator().evaluate(context);
         if (value instanceof String) {
             return (String) value;
@@ -194,7 +194,7 @@ private String validateAndGetString(ComputeField field, TransformContext context
     }
 
     private Map<Schema.Field, Object> getEvaluatedFields(
-            List<ComputeField> fields, TransformContext context) {
+            List<ComputeField> fields, MutableRecord context) {
         Map<Schema.Field, Object> evaluatedFields =
                 new LinkedHashMap<>(); // preserves the insertion order of keys
         for (ComputeField field : fields) {
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropFieldStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropFieldStep.java
index 6d4e5a243..a88701f56 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropFieldStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropFieldStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -35,10 +35,10 @@ public class DropFieldStep implements TransformStep {
             new ConcurrentHashMap<>();
 
     @Override
-    public void process(TransformContext transformContext) {
-        if (transformContext.getKeyObject() != null) {
-            transformContext.dropKeyFields(keyFields, keySchemaCache);
+    public void process(MutableRecord mutableRecord) {
+        if (mutableRecord.getKeyObject() != null) {
+            mutableRecord.dropKeyFields(keyFields, keySchemaCache);
         }
-        transformContext.dropValueFields(valueFields, valueSchemaCache);
+        mutableRecord.dropValueFields(valueFields, valueSchemaCache);
     }
 }
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropStep.java
index 0d47325d5..22bdf194c 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/DropStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 
 /**
  * Drops a message from further processing. Works in conjunctions with predicates to selectively
@@ -23,7 +23,7 @@
  */
 public class DropStep implements TransformStep {
     @Override
-    public void process(TransformContext transformContext) throws Exception {
-        transformContext.setDropCurrentRecord(true);
+    public void process(MutableRecord mutableRecord) throws Exception {
+        mutableRecord.setDropCurrentRecord(true);
     }
 }
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/FlattenStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/FlattenStep.java
index b2eb54b96..f296126b0 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/FlattenStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/FlattenStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import java.util.ArrayList;
 import java.util.List;
@@ -42,23 +42,23 @@ public class FlattenStep implements TransformStep {
     private final String part;
 
     @Override
-    public void process(TransformContext transformContext) throws Exception {
+    public void process(MutableRecord mutableRecord) throws Exception {
         if (part != null && !part.equals("key") && !part.equals("value")) {
             throw new IllegalArgumentException("Unsupported part for Flatten: " + part);
         }
         if ("key".equals(part) || part == null) {
-            validateAvro(transformContext.getKeySchemaType());
+            validateAvro(mutableRecord.getKeySchemaType());
             GenericRecord flattenedKey =
-                    flattenGenericRecord((GenericRecord) transformContext.getKeyObject());
-            transformContext.setKeyObject(flattenedKey);
-            transformContext.setKeyNativeSchema(flattenedKey.getSchema());
+                    flattenGenericRecord((GenericRecord) mutableRecord.getKeyObject());
+            mutableRecord.setKeyObject(flattenedKey);
+            mutableRecord.setKeyNativeSchema(flattenedKey.getSchema());
         }
         if ("value".equals(part) || part == null) {
-            validateAvro(transformContext.getValueSchemaType());
+            validateAvro(mutableRecord.getValueSchemaType());
             GenericRecord flattenedValue =
-                    flattenGenericRecord((GenericRecord) transformContext.getValueObject());
-            transformContext.setValueObject(flattenedValue);
-            transformContext.setValueNativeSchema(flattenedValue.getSchema());
+                    flattenGenericRecord((GenericRecord) mutableRecord.getValueObject());
+            mutableRecord.setValueObject(flattenedValue);
+            mutableRecord.setValueNativeSchema(flattenedValue.getSchema());
         }
     }
 
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/MergeKeyValueStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/MergeKeyValueStep.java
index 15eb31718..e36cd36bc 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/MergeKeyValueStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/MergeKeyValueStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -33,20 +33,20 @@ public class MergeKeyValueStep implements TransformStep {
             schemaCache = new ConcurrentHashMap<>();
 
     @Override
-    public void process(TransformContext transformContext) {
-        TransformSchemaType keySchemaType = transformContext.getKeySchemaType();
+    public void process(MutableRecord mutableRecord) {
+        TransformSchemaType keySchemaType = mutableRecord.getKeySchemaType();
         if (keySchemaType == null) {
             return;
         }
-        Object keyObject = transformContext.getKeyObject();
-        Object valueObject = transformContext.getValueObject();
+        Object keyObject = mutableRecord.getKeyObject();
+        Object valueObject = mutableRecord.getValueObject();
         if (keyObject instanceof Map && valueObject instanceof Map) {
             Map<Object, Object> value = (Map<Object, Object>) valueObject;
             Map<String, Object> keyCopy =
                     OBJECT_MAPPER.convertValue(keyObject, new TypeReference<>() {});
             keyCopy.forEach(value::putIfAbsent);
         } else if (keySchemaType == TransformSchemaType.AVRO
-                && transformContext.getValueSchemaType() == TransformSchemaType.AVRO) {
+                && mutableRecord.getValueSchemaType() == TransformSchemaType.AVRO) {
             GenericRecord avroKeyRecord = (GenericRecord) keyObject;
             org.apache.avro.Schema avroKeySchema = avroKeyRecord.getSchema();
 
@@ -65,24 +65,24 @@ public void process(TransformContext transformContext) {
                     newRecord.put(field.name(), avroKeyRecord.get(field.name()));
                 }
             }
-            transformContext.setValueObject(newRecord);
-            transformContext.setValueNativeSchema(newRecord.getSchema());
+            mutableRecord.setValueObject(newRecord);
+            mutableRecord.setValueNativeSchema(newRecord.getSchema());
         } else if (keySchemaType == TransformSchemaType.JSON
-                && transformContext.getValueSchemaType() == TransformSchemaType.JSON) {
+                && mutableRecord.getValueSchemaType() == TransformSchemaType.JSON) {
             org.apache.avro.Schema avroKeySchema =
-                    (org.apache.avro.Schema) transformContext.getKeyNativeSchema();
+                    (org.apache.avro.Schema) mutableRecord.getKeyNativeSchema();
             org.apache.avro.Schema avroValueSchema =
-                    (org.apache.avro.Schema) transformContext.getValueNativeSchema();
+                    (org.apache.avro.Schema) mutableRecord.getValueNativeSchema();
             if (avroValueSchema != null && avroKeySchema != null) {
                 org.apache.avro.Schema mergedSchema =
                         getMergedSchema(avroKeySchema, avroValueSchema);
-                transformContext.setValueNativeSchema(mergedSchema);
+                mutableRecord.setValueNativeSchema(mergedSchema);
             } else {
-                transformContext.setValueNativeSchema(null);
+                mutableRecord.setValueNativeSchema(null);
             }
             ObjectNode newValue = ((ObjectNode) keyObject).deepCopy();
             newValue.setAll(((ObjectNode) valueObject).deepCopy());
-            transformContext.setValueObject(newValue);
+            mutableRecord.setValueObject(newValue);
         }
     }
 
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/QueryStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/QueryStep.java
index 72b655fee..a4b45a2e4 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/QueryStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/QueryStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import com.datastax.oss.streaming.ai.datasource.QueryStepDataSource;
 import java.util.ArrayList;
@@ -82,13 +82,13 @@ public class QueryStep implements TransformStep {
     }
 
     @Override
-    public void process(TransformContext transformContext) {
+    public void process(MutableRecord mutableRecord) {
         Schema schema;
         Object finalResult;
 
         switch (mode) {
             case MODE_QUERY:
-                List<Map<String, Object>> results = processQuery(transformContext);
+                List<Map<String, Object>> results = processQuery(mutableRecord);
                 if (onlyFirst) {
                     schema = Schema.createMap(Schema.create(Schema.Type.STRING));
                     if (results.isEmpty()) {
@@ -103,7 +103,7 @@ public void process(TransformContext transformContext) {
                 }
                 break;
             case MODE_EXECUTE:
-                finalResult = processExecute(transformContext);
+                finalResult = processExecute(mutableRecord);
                 if (finalResult instanceof Map) {
                     schema = Schema.createMap(Schema.create(Schema.Type.STRING));
                 } else if (finalResult instanceof List) {
@@ -116,22 +116,21 @@ public void process(TransformContext transformContext) {
             default:
                 throw new IllegalStateException("Unknown mode " + mode);
         }
-
-        transformContext.setResultField(
+        mutableRecord.setResultField(
                 finalResult, outputFieldName, schema, avroKeySchemaCache, avroValueSchemaCache);
     }
 
-    private List<Map<String, Object>> processQuery(TransformContext transformContext) {
+    private List<Map<String, Object>> processQuery(MutableRecord mutableRecord) {
         List<Map<String, Object>> results;
         if (loopOverAccessor == null) {
-            results = performQuery(transformContext);
+            results = performQuery(mutableRecord);
         } else {
             // loop over a list
             // for each item we name if "record" and we perform the query
-            List<Object> nestedRecords = loopOverAccessor.evaluate(transformContext);
+            List<Object> nestedRecords = loopOverAccessor.evaluate(mutableRecord);
             results = new ArrayList<>();
             for (Object document : nestedRecords) {
-                TransformContext nestedRecordContext = new TransformContext();
+                MutableRecord nestedRecordContext = new MutableRecord();
                 nestedRecordContext.setRecordObject(document);
                 // set in the item the result
                 List<Map<String, Object>> resultsForDocument = performQuery(nestedRecordContext);
@@ -141,17 +140,17 @@ private List<Map<String, Object>> processQuery(TransformContext transformContext
         return results;
     }
 
-    private Object processExecute(TransformContext transformContext) {
+    private Object processExecute(MutableRecord mutableRecord) {
         Object res;
         if (loopOverAccessor == null) {
-            res = executeStatement(transformContext);
+            res = executeStatement(mutableRecord);
         } else {
             // loop over a list
             // for each item we name if "record" and we perform the query
-            List<Object> nestedRecords = loopOverAccessor.evaluate(transformContext);
+            List<Object> nestedRecords = loopOverAccessor.evaluate(mutableRecord);
             List<Map<String, Object>> results = new ArrayList<>();
             for (Object document : nestedRecords) {
-                TransformContext nestedRecordContext = new TransformContext();
+                MutableRecord nestedRecordContext = new MutableRecord();
                 nestedRecordContext.setRecordObject(document);
                 // set in the item the result
                 Map<String, Object> resultsForDocument = executeStatement(nestedRecordContext);
@@ -162,14 +161,14 @@ private Object processExecute(TransformContext transformContext) {
         return res;
     }
 
-    private List<Map<String, Object>> performQuery(TransformContext transformContext) {
+    private List<Map<String, Object>> performQuery(MutableRecord mutableRecord) {
         List<Object> params = new ArrayList<>();
         fieldsEvaluators.forEach(
                 field -> {
                     if (log.isDebugEnabled()) {
                         log.debug("Evaluating {}", field);
                     }
-                    Object value = field.evaluate(transformContext);
+                    Object value = field.evaluate(mutableRecord);
                     if (log.isDebugEnabled()) {
                         log.debug(
                                 "Result {} type {}",
@@ -188,14 +187,14 @@ private List<Map<String, Object>> performQuery(TransformContext transformContext
         return results;
     }
 
-    private Map<String, Object> executeStatement(TransformContext transformContext) {
+    private Map<String, Object> executeStatement(MutableRecord mutableRecord) {
         List<Object> params = new ArrayList<>();
         fieldsEvaluators.forEach(
                 field -> {
                     if (log.isDebugEnabled()) {
                         log.debug("Evaluating {}", field);
                     }
-                    Object value = field.evaluate(transformContext);
+                    Object value = field.evaluate(mutableRecord);
                     if (log.isDebugEnabled()) {
                         log.debug(
                                 "Result {} type {}",
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TextCompletionsStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TextCompletionsStep.java
index 407b8f9d4..441788f64 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TextCompletionsStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TextCompletionsStep.java
@@ -18,7 +18,7 @@
 import static com.datastax.oss.streaming.ai.util.TransformFunctionUtil.convertToMap;
 
 import ai.langstream.ai.agents.commons.JsonRecord;
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import com.datastax.oss.streaming.ai.completions.Chunk;
 import com.datastax.oss.streaming.ai.completions.CompletionsService;
 import com.datastax.oss.streaming.ai.completions.TextCompletionResult;
@@ -80,8 +80,8 @@ public void close() throws Exception {
     }
 
     @Override
-    public CompletableFuture<?> processAsync(TransformContext transformContext) {
-        JsonRecord jsonRecord = transformContext.toJsonRecord();
+    public CompletableFuture<?> processAsync(MutableRecord mutableRecord) {
+        JsonRecord jsonRecord = mutableRecord.toJsonRecord();
 
         List<String> prompt =
                 config.getPrompt().stream()
@@ -102,7 +102,7 @@ public void consumeChunk(
                                 // we must copy the context because the same context is used for all
                                 // chunks
                                 // and also for the final answer
-                                TransformContext copy = transformContext.copy();
+                                MutableRecord copy = mutableRecord.copy();
 
                                 copy.getProperties().put("stream-id", answerId);
                                 copy.getProperties().put("stream-index", index + "");
@@ -118,7 +118,7 @@ public void consumeChunk(
 
         return chatCompletionsHandle.thenApply(
                 content -> {
-                    applyResultFieldToContext(transformContext, content.text(), false);
+                    applyResultFieldToContext(mutableRecord, content.text(), false);
 
                     String logField = config.getLogField();
                     if (logField != null && !logField.isEmpty()) {
@@ -126,8 +126,8 @@ public void consumeChunk(
                         logMap.put("model", config.getModel());
                         logMap.put("options", options);
                         logMap.put("messages", prompt);
-                        transformContext.setResultField(
-                                TransformContext.toJson(logMap),
+                        mutableRecord.setResultField(
+                                MutableRecord.toJson(logMap),
                                 logField,
                                 Schema.create(Schema.Type.STRING),
                                 avroKeySchemaCache,
@@ -151,7 +151,7 @@ public void consumeChunk(
                         logMap.put("tokens", tokens != null ? tokens : List.of());
                         logMap.put("logprobs", logprobs != null ? logprobs : List.of());
 
-                        transformContext.setResultField(
+                        mutableRecord.setResultField(
                                 logMap,
                                 logProbsField,
                                 Schema.createRecord(
@@ -180,7 +180,7 @@ public void consumeChunk(
     }
 
     private void applyResultFieldToContext(
-            TransformContext transformContext, String content, boolean streamingAnswer) {
+            MutableRecord mutableRecord, String content, boolean streamingAnswer) {
         String fieldName = config.getFieldName();
 
         // maybe we want a different field in the streaming answer
@@ -190,7 +190,7 @@ private void applyResultFieldToContext(
                 && !config.getStreamResponseCompletionField().isEmpty()) {
             fieldName = config.getStreamResponseCompletionField();
         }
-        transformContext.setResultField(
+        mutableRecord.setResultField(
                 content,
                 fieldName,
                 Schema.create(Schema.Type.STRING),
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TransformStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TransformStep.java
index b33c85e30..6c3edba45 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TransformStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/TransformStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
@@ -24,9 +24,9 @@ default void close() throws Exception {}
 
     default void start() throws Exception {}
 
-    default void process(TransformContext transformContext) throws Exception {
+    default void process(MutableRecord mutableRecord) throws Exception {
         try {
-            processAsync(transformContext).get();
+            processAsync(mutableRecord).get();
         } catch (ExecutionException err) {
             if (err.getCause() instanceof Exception) {
                 throw (Exception) err.getCause();
@@ -36,9 +36,9 @@ default void process(TransformContext transformContext) throws Exception {
         }
     }
 
-    default CompletableFuture<?> processAsync(TransformContext transformContext) {
+    default CompletableFuture<?> processAsync(MutableRecord mutableRecord) {
         try {
-            process(transformContext);
+            process(mutableRecord);
             return CompletableFuture.completedFuture(null);
         } catch (Throwable error) {
             return CompletableFuture.failedFuture(error);
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/UnwrapKeyValueStep.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/UnwrapKeyValueStep.java
index ae49d8456..9fe772698 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/UnwrapKeyValueStep.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/UnwrapKeyValueStep.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 
 public class UnwrapKeyValueStep implements TransformStep {
 
@@ -26,16 +26,16 @@ public UnwrapKeyValueStep(boolean unwrapKey) {
     }
 
     @Override
-    public void process(TransformContext transformContext) throws Exception {
-        if (transformContext.getKeySchemaType() != null) {
+    public void process(MutableRecord mutableRecord) throws Exception {
+        if (mutableRecord.getKeySchemaType() != null) {
             if (unwrapKey) {
-                transformContext.setValueSchemaType(transformContext.getKeySchemaType());
-                transformContext.setValueNativeSchema(transformContext.getKeyNativeSchema());
-                transformContext.setValueObject(transformContext.getKeyObject());
+                mutableRecord.setValueSchemaType(mutableRecord.getKeySchemaType());
+                mutableRecord.setValueNativeSchema(mutableRecord.getKeyNativeSchema());
+                mutableRecord.setValueObject(mutableRecord.getKeyObject());
             }
-            transformContext.setKeySchemaType(null);
-            transformContext.setKeyNativeSchema(null);
-            transformContext.setKeyObject(null);
+            mutableRecord.setKeySchemaType(null);
+            mutableRecord.setKeyNativeSchema(null);
+            mutableRecord.setKeyObject(null);
         }
     }
 }
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/streaming/StreamingAnswersConsumer.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/streaming/StreamingAnswersConsumer.java
index 010107495..e6182de78 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/streaming/StreamingAnswersConsumer.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/streaming/StreamingAnswersConsumer.java
@@ -15,10 +15,10 @@
  */
 package com.datastax.oss.streaming.ai.streaming;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 
 public interface StreamingAnswersConsumer {
-    void streamAnswerChunk(int index, String message, boolean last, TransformContext outputMessage);
+    void streamAnswerChunk(int index, String message, boolean last, MutableRecord outputMessage);
 
     default void close() {}
 }
diff --git a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java
index 92886a718..bd1e9c852 100644
--- a/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java
+++ b/langstream-agents/langstream-ai-agents/src/main/java/com/datastax/oss/streaming/ai/util/TransformFunctionUtil.java
@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.streaming.ai.util;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import ai.langstream.ai.agents.commons.jstl.predicate.JstlPredicate;
 import com.azure.ai.openai.OpenAIAsyncClient;
@@ -355,7 +355,7 @@ public static TransformStep newQuery(QueryConfig config, QueryStepDataSource dat
                             field -> {
                                 if (config.getLoopOver() != null
                                         && !config.getLoopOver().isEmpty()) {
-                                    if (!field.contains("record.")) {
+                                    if (!field.contains("record.") && !field.contains("fn:now")) {
                                         throw new IllegalArgumentException(
                                                 String.format(
                                                         "Invalid field name for query step (with loop-over you must use record.xxx: %s",
@@ -365,6 +365,7 @@ public static TransformStep newQuery(QueryConfig config, QueryStepDataSource dat
                                     if (!FIELD_NAMES.contains(field)
                                             && !field.contains("value.")
                                             && !field.contains("key.")
+                                            && !field.contains("fn:now")
                                             && !field.contains("properties.")) {
                                         throw new IllegalArgumentException(
                                                 String.format(
@@ -387,19 +388,18 @@ public static TransformStep newQuery(QueryConfig config, QueryStepDataSource dat
     }
 
     public static void processTransformSteps(
-            TransformContext transformContext, Collection<StepPredicatePair> steps)
-            throws Exception {
+            MutableRecord mutableRecord, Collection<StepPredicatePair> steps) throws Exception {
         for (StepPredicatePair pair : steps) {
-            processStep(transformContext, pair);
+            processStep(mutableRecord, pair);
         }
     }
 
-    public static void processStep(TransformContext transformContext, StepPredicatePair pair)
+    public static void processStep(MutableRecord mutableRecord, StepPredicatePair pair)
             throws Exception {
         TransformStep step = pair.getTransformStep();
-        Predicate<TransformContext> predicate = pair.getPredicate();
-        if (predicate == null || predicate.test(transformContext)) {
-            step.process(transformContext);
+        Predicate<MutableRecord> predicate = pair.getPredicate();
+        if (predicate == null || predicate.test(mutableRecord)) {
+            step.process(mutableRecord);
         }
     }
 
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/QueryVectorDBAgent.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/QueryVectorDBAgent.java
index 798cdcdf1..76d3a0ed5 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/QueryVectorDBAgent.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/QueryVectorDBAgent.java
@@ -15,10 +15,10 @@
  */
 package ai.langstream.agents.vector;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
-import static ai.langstream.ai.agents.commons.TransformContext.transformContextToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.mutableRecordToRecord;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.datasource.DataSourceProviderRegistry;
 import ai.langstream.api.runner.code.Record;
 import ai.langstream.api.runner.code.SingleRecordAgentProcessor;
@@ -65,10 +65,10 @@ public List<Record> processRecord(Record record) throws Exception {
         if (log.isDebugEnabled()) {
             log.debug("Processing record {}", record);
         }
-        TransformContext context = recordToTransformContext(record, true);
+        MutableRecord context = recordToMutableRecord(record, true);
         TransformFunctionUtil.processTransformSteps(context, steps);
         context.convertMapToStringOrBytes();
-        Optional<Record> recordResult = transformContextToRecord(context);
+        Optional<Record> recordResult = mutableRecordToRecord(context);
         if (log.isDebugEnabled()) {
             log.debug("recordResult {}", recordResult);
         }
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/jdbc/JdbcWriter.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/jdbc/JdbcWriter.java
index b7c6c2dad..226c202f4 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/jdbc/JdbcWriter.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/jdbc/JdbcWriter.java
@@ -15,9 +15,9 @@
  */
 package ai.langstream.agents.vector.jdbc;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.ai.agents.datasource.impl.JdbcDataSourceProvider;
 import ai.langstream.api.database.VectorDatabaseWriter;
@@ -141,10 +141,10 @@ public synchronized CompletableFuture<?> upsert(
                 Record record, Map<String, Object> context) {
             CompletableFuture<?> handle = new CompletableFuture<>();
             try {
-                TransformContext transformContext = recordToTransformContext(record, true);
+                MutableRecord mutableRecord = recordToMutableRecord(record, true);
 
-                List<Object> primaryKeyValues = prepareValueList(transformContext, primaryKey);
-                List<Object> otherValues = prepareValueList(transformContext, columns);
+                List<Object> primaryKeyValues = prepareValueList(mutableRecord, primaryKey);
+                List<Object> otherValues = prepareValueList(mutableRecord, columns);
                 if (record.value() != null) {
                     int i = 1;
                     for (Object value : otherValues) {
@@ -179,11 +179,11 @@ public synchronized CompletableFuture<?> upsert(
         }
 
         private List<Object> prepareValueList(
-                TransformContext transformContext, Map<String, JstlEvaluator> primaryKey) {
+                MutableRecord mutableRecord, Map<String, JstlEvaluator> primaryKey) {
             List<Object> result = new ArrayList<>();
             primaryKey.forEach(
                     (name, evaluator) -> {
-                        Object value = evaluator.evaluate(transformContext);
+                        Object value = evaluator.evaluate(mutableRecord);
                         if (log.isDebugEnabled()) {
                             log.debug(
                                     "setting value {} ({}) for field {}",
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/milvus/MilvusWriter.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/milvus/MilvusWriter.java
index 66663c4ec..aebe082ca 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/milvus/MilvusWriter.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/milvus/MilvusWriter.java
@@ -15,9 +15,9 @@
  */
 package ai.langstream.agents.vector.milvus;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.database.VectorDatabaseWriter;
 import ai.langstream.api.database.VectorDatabaseWriterProvider;
@@ -116,12 +116,12 @@ public void initialise(Map<String, Object> agentConfiguration) throws Exception
         public CompletableFuture<?> upsert(Record record, Map<String, Object> context) {
             CompletableFuture<?> handle = new CompletableFuture<>();
             try {
-                TransformContext transformContext = recordToTransformContext(record, true);
+                MutableRecord mutableRecord = recordToMutableRecord(record, true);
 
                 JSONObject row = new JSONObject();
                 fields.forEach(
                         (name, evaluator) -> {
-                            Object value = evaluator.evaluate(transformContext);
+                            Object value = evaluator.evaluate(mutableRecord);
                             if (value != null) {
                                 if (log.isDebugEnabled()) {
                                     log.debug(
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/opensearch/OpenSearchWriter.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/opensearch/OpenSearchWriter.java
index 8a1e7e03a..d29eb15ac 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/opensearch/OpenSearchWriter.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/opensearch/OpenSearchWriter.java
@@ -15,9 +15,9 @@
  */
 package ai.langstream.agents.vector.opensearch;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.database.VectorDatabaseWriter;
 import ai.langstream.api.database.VectorDatabaseWriterProvider;
@@ -304,14 +304,14 @@ public CompletableFuture<?> upsert(Record record, Map<String, Object> context) {
 
             CompletableFuture<?> handle = new CompletableFuture<>();
             try {
-                TransformContext transformContext = recordToTransformContext(record, true);
+                MutableRecord mutableRecord = recordToMutableRecord(record, true);
                 Map<String, Object> documentJson;
 
                 if (record.value() != null) {
                     documentJson = new HashMap<>();
                     fields.forEach(
                             (name, evaluator) -> {
-                                Object value = evaluator.evaluate(transformContext);
+                                Object value = evaluator.evaluate(mutableRecord);
                                 if (log.isDebugEnabled()) {
                                     log.debug(
                                             "setting value {} ({}) for field {}",
@@ -329,7 +329,7 @@ public CompletableFuture<?> upsert(Record record, Map<String, Object> context) {
                 if (id == null) {
                     documentId = null;
                 } else {
-                    final Object evaluate = id.evaluate(transformContext);
+                    final Object evaluate = id.evaluate(mutableRecord);
                     documentId = evaluate == null ? null : evaluate.toString();
                 }
                 if (documentJson == null && documentId == null) {
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/pinecone/PineconeWriter.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/pinecone/PineconeWriter.java
index 8b059cda7..d2e66fd86 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/pinecone/PineconeWriter.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/pinecone/PineconeWriter.java
@@ -15,9 +15,9 @@
  */
 package ai.langstream.agents.vector.pinecone;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.database.VectorDatabaseWriter;
 import ai.langstream.api.database.VectorDatabaseWriterProvider;
@@ -103,23 +103,22 @@ public void initialise(Map<String, Object> agentConfiguration) {
         public CompletableFuture<?> upsert(Record record, Map<String, Object> context) {
             CompletableFuture<?> handle = new CompletableFuture<>();
             try {
-                TransformContext transformContext = recordToTransformContext(record, true);
-                String id =
-                        idFunction != null ? (String) idFunction.evaluate(transformContext) : null;
+                MutableRecord mutableRecord = recordToMutableRecord(record, true);
+                String id = idFunction != null ? (String) idFunction.evaluate(mutableRecord) : null;
                 String namespace =
                         namespaceFunction != null
-                                ? (String) namespaceFunction.evaluate(transformContext)
+                                ? (String) namespaceFunction.evaluate(mutableRecord)
                                 : null;
                 List<Object> vector =
                         vectorFunction != null
-                                ? (List<Object>) vectorFunction.evaluate(transformContext)
+                                ? (List<Object>) vectorFunction.evaluate(mutableRecord)
                                 : null;
                 Map<String, Object> metadata =
                         metadataFunctions.entrySet().stream()
                                 .collect(
                                         Collectors.toMap(
                                                 Map.Entry::getKey,
-                                                e -> e.getValue().evaluate(transformContext)));
+                                                e -> e.getValue().evaluate(mutableRecord)));
                 Struct metadataStruct =
                         Struct.newBuilder()
                                 .putAllFields(
diff --git a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/solr/SolrWriter.java b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/solr/SolrWriter.java
index 3a70e144d..36ad0ad20 100644
--- a/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/solr/SolrWriter.java
+++ b/langstream-agents/langstream-vector-agents/src/main/java/ai/langstream/agents/vector/solr/SolrWriter.java
@@ -15,9 +15,9 @@
  */
 package ai.langstream.agents.vector.solr;
 
-import static ai.langstream.ai.agents.commons.TransformContext.recordToTransformContext;
+import static ai.langstream.ai.agents.commons.MutableRecord.recordToMutableRecord;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.database.VectorDatabaseWriter;
 import ai.langstream.api.database.VectorDatabaseWriterProvider;
@@ -95,12 +95,12 @@ public void initialise(Map<String, Object> agentConfiguration) throws Exception
         public CompletableFuture<?> upsert(Record record, Map<String, Object> context) {
             CompletableFuture<?> handle = new CompletableFuture<>();
             try {
-                TransformContext transformContext = recordToTransformContext(record, true);
+                MutableRecord mutableRecord = recordToMutableRecord(record, true);
 
                 SolrInputDocument document = new SolrInputDocument();
                 fields.forEach(
                         (name, evaluator) -> {
-                            Object value = evaluator.evaluate(transformContext);
+                            Object value = evaluator.evaluate(mutableRecord);
                             if (log.isDebugEnabled()) {
                                 log.debug(
                                         "setting value {} ({}) for field {}",
diff --git a/langstream-api/src/main/java/ai/langstream/api/doc/ConfigProperty.java b/langstream-api/src/main/java/ai/langstream/api/doc/ConfigProperty.java
index ae6091575..d78c2c365 100644
--- a/langstream-api/src/main/java/ai/langstream/api/doc/ConfigProperty.java
+++ b/langstream-api/src/main/java/ai/langstream/api/doc/ConfigProperty.java
@@ -28,4 +28,6 @@
     String description() default "";
 
     String defaultValue() default "";
+
+    ExtendedValidationType extendedValidationType() default ExtendedValidationType.NONE;
 }
diff --git a/langstream-api/src/main/java/ai/langstream/api/doc/ConfigPropertyModel.java b/langstream-api/src/main/java/ai/langstream/api/doc/ConfigPropertyModel.java
index c3a050322..4cf70bf2e 100644
--- a/langstream-api/src/main/java/ai/langstream/api/doc/ConfigPropertyModel.java
+++ b/langstream-api/src/main/java/ai/langstream/api/doc/ConfigPropertyModel.java
@@ -32,4 +32,5 @@ public class ConfigPropertyModel {
     private Map<String, ConfigPropertyModel> properties;
     private ConfigPropertyModel items;
     private Object defaultValue;
+    private ExtendedValidationType extendedValidationType;
 }
diff --git a/langstream-api/src/main/java/ai/langstream/api/doc/ExtendedValidationType.java b/langstream-api/src/main/java/ai/langstream/api/doc/ExtendedValidationType.java
new file mode 100644
index 000000000..753463783
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/doc/ExtendedValidationType.java
@@ -0,0 +1,31 @@
+/*
+ * 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.doc;
+
+public enum ExtendedValidationType {
+    EL_EXPRESSION("el-expression"),
+    NONE("");
+
+    private final String name;
+
+    ExtendedValidationType(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/topics/TopicConnectionProvider.java b/langstream-api/src/main/java/ai/langstream/api/runner/topics/TopicConnectionProvider.java
index 5da260edc..2af8c3963 100644
--- a/langstream-api/src/main/java/ai/langstream/api/runner/topics/TopicConnectionProvider.java
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/topics/TopicConnectionProvider.java
@@ -18,7 +18,11 @@
 import java.util.Map;
 
 public interface TopicConnectionProvider {
-    TopicConsumer createConsumer(String agentId, Map<String, Object> config);
+    default TopicConsumer createConsumer(String agentId, Map<String, Object> config) {
+        throw new UnsupportedOperationException("Consumer not supported");
+    }
 
-    TopicProducer createProducer(String agentId, String topic, Map<String, Object> config);
+    default TopicProducer createProducer(String agentId, String topic, Map<String, Object> config) {
+        throw new UnsupportedOperationException("Producer not supported");
+    }
 }
diff --git a/langstream-core/pom.xml b/langstream-core/pom.xml
index 968708c8a..ad93ab52f 100644
--- a/langstream-core/pom.xml
+++ b/langstream-core/pom.xml
@@ -35,7 +35,11 @@
       <artifactId>langstream-api</artifactId>
       <version>${project.version}</version>
     </dependency>
-
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>langstream-agents-commons</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.dataformat</groupId>
       <artifactId>jackson-dataformat-yaml</artifactId>
diff --git a/langstream-core/src/main/java/ai/langstream/impl/uti/ClassConfigValidator.java b/langstream-core/src/main/java/ai/langstream/impl/uti/ClassConfigValidator.java
index 286bb86bf..c6ab42e47 100644
--- a/langstream-core/src/main/java/ai/langstream/impl/uti/ClassConfigValidator.java
+++ b/langstream-core/src/main/java/ai/langstream/impl/uti/ClassConfigValidator.java
@@ -15,6 +15,7 @@
  */
 package ai.langstream.impl.uti;
 
+import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.api.doc.AgentConfig;
 import ai.langstream.api.doc.AgentConfigurationModel;
 import ai.langstream.api.doc.AssetConfig;
@@ -22,6 +23,7 @@
 import ai.langstream.api.doc.ConfigProperty;
 import ai.langstream.api.doc.ConfigPropertyIgnore;
 import ai.langstream.api.doc.ConfigPropertyModel;
+import ai.langstream.api.doc.ExtendedValidationType;
 import ai.langstream.api.doc.ResourceConfig;
 import ai.langstream.api.doc.ResourceConfigurationModel;
 import ai.langstream.api.model.AgentConfiguration;
@@ -50,8 +52,10 @@
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 
+@Slf4j
 public class ClassConfigValidator {
 
     static final ObjectMapper jsonWriter =
@@ -306,6 +310,9 @@ private static void validateProperty(
                     propertyValue.getProperties(),
                     false);
         }
+        if (propertyValue.getExtendedValidationType() != null) {
+            validateExtendedValidationType(propertyValue.getExtendedValidationType(), actualValue);
+        }
     }
 
     @Data
@@ -314,6 +321,7 @@ public static class ConfigPropertyModel {
         private boolean required;
         private String description;
         private String defaultValue;
+        private ExtendedValidationType extendedValidationType;
     }
 
     @Data
@@ -370,6 +378,10 @@ private static ai.langstream.api.doc.ConfigPropertyModel parseProp(
             newProp.setDescription(
                     property.getDescription() == null ? null : property.getDescription().strip());
             newProp.setDefaultValue(property.getDefaultValue());
+            if (property.getExtendedValidationType() != null
+                    && property.getExtendedValidationType() != ExtendedValidationType.NONE) {
+                newProp.setExtendedValidationType(property.getExtendedValidationType());
+            }
         }
 
         if (value.getProperties() != null) {
@@ -423,6 +435,10 @@ public String apply(FieldScope fieldScope) {
                                                 && !annotation.defaultValue().isEmpty()) {
                                             model.setDefaultValue(annotation.defaultValue());
                                         }
+                                        if (annotation.extendedValidationType() != null) {
+                                            model.setExtendedValidationType(
+                                                    annotation.extendedValidationType());
+                                        }
                                     }
                                 }
 
@@ -507,4 +523,16 @@ private static void validateConversion(
             }
         }
     }
+
+    private static void validateExtendedValidationType(
+            ExtendedValidationType extendedValidationType, Object actualValue) {
+        switch (extendedValidationType) {
+            case EL_EXPRESSION -> {
+                String expression = actualValue.toString();
+                log.info("Validating EL expression: {}", expression);
+                new JstlEvaluator(actualValue.toString(), Object.class);
+            }
+            case NONE -> {}
+        }
+    }
 }
diff --git a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/agents/FlowControlAgentsProvider.java b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/agents/FlowControlAgentsProvider.java
index d5603f13b..b9f147ee5 100644
--- a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/agents/FlowControlAgentsProvider.java
+++ b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/agents/FlowControlAgentsProvider.java
@@ -17,6 +17,7 @@
 
 import ai.langstream.api.doc.AgentConfig;
 import ai.langstream.api.doc.ConfigProperty;
+import ai.langstream.api.doc.ExtendedValidationType;
 import ai.langstream.api.model.AgentConfiguration;
 import ai.langstream.api.model.Module;
 import ai.langstream.api.model.Pipeline;
@@ -28,6 +29,7 @@
 import ai.langstream.impl.agents.AbstractComposableAgentProvider;
 import ai.langstream.impl.uti.ClassConfigValidator;
 import ai.langstream.runtime.impl.k8s.KubernetesClusterRuntime;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -39,7 +41,10 @@
 public class FlowControlAgentsProvider extends AbstractComposableAgentProvider {
 
     protected static final String DISPATCH = "dispatch";
-    private static final Set<String> SUPPORTED_AGENT_TYPES = Set.of(DISPATCH);
+    protected static final String TIMER_SOURCE = "timer-source";
+    protected static final String TRIGGER_EVENT = "trigger-event";
+    private static final Set<String> SUPPORTED_AGENT_TYPES =
+            Set.of(DISPATCH, TIMER_SOURCE, TRIGGER_EVENT);
 
     public FlowControlAgentsProvider() {
         super(SUPPORTED_AGENT_TYPES, List.of(KubernetesClusterRuntime.CLUSTER_TYPE, "none"));
@@ -47,7 +52,13 @@ public FlowControlAgentsProvider() {
 
     @Override
     protected ComponentType getComponentType(AgentConfiguration agentConfiguration) {
-        return ComponentType.PROCESSOR;
+        return switch (agentConfiguration.getType()) {
+            case DISPATCH -> ComponentType.PROCESSOR;
+            case TIMER_SOURCE -> ComponentType.SOURCE;
+            case TRIGGER_EVENT -> ComponentType.PROCESSOR;
+            default -> throw new IllegalArgumentException(
+                    "Unsupported agent type: " + agentConfiguration.getType());
+        };
     }
 
     @Override
@@ -58,27 +69,47 @@ protected Map<String, Object> computeAgentConfiguration(
             ExecutionPlan executionPlan,
             ComputeClusterRuntime clusterRuntime,
             PluginsRegistry pluginsRegistry) {
-        DispatchConfig dispatchConfig =
-                ClassConfigValidator.convertValidatedConfiguration(
-                        agentConfiguration.getConfiguration(), DispatchConfig.class);
-        List<RouteConfiguration> routes = dispatchConfig.getRoutes();
-        if (routes != null) {
-            for (RouteConfiguration routeConfiguration : routes) {
-                String action = routeConfiguration.getAction();
-                ConfigurationUtils.validateEnumValue(
-                        "action",
-                        Set.of("dispatch", "drop"),
-                        action,
-                        () -> "route " + routeConfiguration);
-                String destination = routeConfiguration.getDestination();
-                if (destination != null && !destination.isEmpty()) {
-                    if (action.equals("drop")) {
-                        throw new IllegalArgumentException("drop action cannot have a destination");
+        switch (agentConfiguration.getType()) {
+            case DISPATCH:
+                {
+                    DispatchConfig dispatchConfig =
+                            ClassConfigValidator.convertValidatedConfiguration(
+                                    agentConfiguration.getConfiguration(), DispatchConfig.class);
+                    List<RouteConfiguration> routes = dispatchConfig.getRoutes();
+                    if (routes != null) {
+                        for (RouteConfiguration routeConfiguration : routes) {
+                            String action = routeConfiguration.getAction();
+                            ConfigurationUtils.validateEnumValue(
+                                    "action",
+                                    Set.of("dispatch", "drop"),
+                                    action,
+                                    () -> "route " + routeConfiguration);
+                            String destination = routeConfiguration.getDestination();
+                            if (destination != null && !destination.isEmpty()) {
+                                if (action.equals("drop")) {
+                                    throw new IllegalArgumentException(
+                                            "drop action cannot have a destination");
+                                }
+                                log.info("Validating topic reference {}", destination);
+                                module.resolveTopic(destination);
+                            }
+                        }
+                    }
+                    break;
+                }
+            case TRIGGER_EVENT:
+                {
+                    TriggerEventProcessorConfig triggerEventProcessorConfig =
+                            ClassConfigValidator.convertValidatedConfiguration(
+                                    agentConfiguration.getConfiguration(),
+                                    TriggerEventProcessorConfig.class);
+                    String destination = triggerEventProcessorConfig.getDestination();
+                    if (destination != null && !destination.isEmpty()) {
+                        log.info("Validating topic reference {}", destination);
+                        module.resolveTopic(destination);
                     }
-                    log.info("Validating topic reference {}", destination);
-                    module.resolveTopic(destination);
+                    break;
                 }
-            }
         }
         return super.computeAgentConfiguration(
                 agentConfiguration,
@@ -93,10 +124,84 @@ protected Map<String, Object> computeAgentConfiguration(
     protected Class getAgentConfigModelClass(String type) {
         return switch (type) {
             case DISPATCH -> DispatchConfig.class;
+            case TIMER_SOURCE -> TimeSourceConfig.class;
+            case TRIGGER_EVENT -> TriggerEventProcessorConfig.class;
             default -> throw new IllegalArgumentException("Unsupported agent type: " + type);
         };
     }
 
+    @AgentConfig(
+            name = "Timer source",
+            description =
+                    """
+            Periodically emits records to trigger the execution of pipelines.
+            """)
+    @Data
+    public static class TimeSourceConfig {
+        @ConfigProperty(
+                description =
+                        """
+                        Fields of the generated records.
+                                """)
+        List<FieldConfiguration> fields;
+
+        @ConfigProperty(
+                description =
+                        """
+                        Period of the timer in seconds.
+                                """,
+                defaultValue = "60")
+        @JsonProperty("period-seconds")
+        int periodInSeconds;
+    }
+
+    @AgentConfig(
+            name = "Trigger event",
+            description =
+                    """
+            Emits a record on a side destination when a record is received.
+            """)
+    @Data
+    public static class TriggerEventProcessorConfig {
+        @ConfigProperty(
+                description =
+                        """
+                        Fields of the generated records.
+                                """)
+        List<FieldConfiguration> fields;
+
+        @ConfigProperty(
+                description =
+                        """
+                        Condition to trigger the event. This is a standard EL expression.
+                                """,
+                defaultValue = "true",
+                extendedValidationType = ExtendedValidationType.EL_EXPRESSION)
+        @JsonProperty("when")
+        String when;
+
+        @ConfigProperty(
+                description =
+                        """
+                        Whether to continue processing the record downstream after emitting the event.
+                        If the when condition is false, the record is passed downstream anyway.
+                        This flag allows you to stop processing system events and trigger a different pipeline.
+                                """,
+                defaultValue = "true")
+        @JsonProperty("continue-processing")
+        boolean continueProcessing;
+
+        @ConfigProperty(
+                description =
+                        """
+                        Destination of the message.
+                                """,
+                defaultValue = "",
+                required = true)
+        @JsonProperty("destination")
+        String destination;
+    }
+
     @AgentConfig(
             name = "Dispatch agent",
             description =
@@ -119,7 +224,8 @@ public static class RouteConfiguration {
                 description =
                         """
                         Condition to activate the route. This is a standard EL expression.
-                                """)
+                                """,
+                extendedValidationType = ExtendedValidationType.EL_EXPRESSION)
         String when;
 
         @ConfigProperty(
@@ -137,4 +243,24 @@ public static class RouteConfiguration {
                 defaultValue = "dispatch")
         String action = "dispatch";
     }
+
+    @Data
+    public static class FieldConfiguration {
+        @ConfigProperty(
+                description =
+                        """
+                        Name of the field like value.xx, key.xxx, properties.xxx
+                                """,
+                required = true)
+        String name;
+
+        @ConfigProperty(
+                description =
+                        """
+                        Expression to compute the value of the field. This is a standard EL expression.
+                                """,
+                required = true,
+                extendedValidationType = ExtendedValidationType.EL_EXPRESSION)
+        String expression;
+    }
 }
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
index 110dfec42..fe8d04609 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
+++ b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
@@ -376,7 +376,7 @@ public TopicProducer createProducer(
                             new PendingRecordsCounterSource(source, sink.handlesCommit());
 
                     statsScheduler.scheduleAtFixedRate(
-                            pendingRecordsCounterSource::dumpStats, 5, 5, TimeUnit.SECONDS);
+                            pendingRecordsCounterSource::dumpStats, 30, 30, TimeUnit.SECONDS);
 
                     runMainLoop(
                             pendingRecordsCounterSource,


Test patch:
diff --git a/langstream-agents/langstream-agents-flow-control/src/test/java/ai/langstream/agents/flow/FlowControlAgentsTest.java b/langstream-agents/langstream-agents-flow-control/src/test/java/ai/langstream/agents/flow/FlowControlAgentsTest.java
new file mode 100644
index 000000000..27d72fced
--- /dev/null
+++ b/langstream-agents/langstream-agents-flow-control/src/test/java/ai/langstream/agents/flow/FlowControlAgentsTest.java
@@ -0,0 +1,352 @@
+/*
+ * 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.agents.flow;
+
+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.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import ai.langstream.api.runner.code.AgentContext;
+import ai.langstream.api.runner.code.AgentProcessor;
+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.TopicConnectionProvider;
+import ai.langstream.api.runner.topics.TopicProducer;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+public class FlowControlAgentsTest {
+
+    @Test
+    public void testTimerSource() throws Exception {
+        try (TimerSource timerSource = new TimerSource(); ) {
+            timerSource.init(
+                    Map.of(
+                            "period-seconds",
+                            1,
+                            "fields",
+                            List.of(
+                                    Map.of("name", "value.now", "expression", "fn:now()"),
+                                    Map.of("name", "key.someid", "expression", "fn:uuid()"),
+                                    Map.of(
+                                            "name",
+                                            "properties.someprop",
+                                            "expression",
+                                            "fn:random(1000)"))));
+            timerSource.start();
+            List<Record> all = new ArrayList<>();
+            while (all.size() < 10) {
+                List<Record> read = timerSource.read();
+                if (read.isEmpty()) {
+                    continue;
+                }
+                assertEquals(1, read.size());
+                all.addAll(read);
+                Record r = read.get(0);
+                log.info("Received record {}", r);
+                Map<String, Object> value = (Map<String, Object>) r.value();
+                assertNotNull(value.get("now"));
+                Map<String, Object> key = (Map<String, Object>) r.key();
+                assertNotNull(key.get("someid"));
+                Header someprop = r.getHeader("someprop");
+                assertNotNull(someprop);
+                assertNotNull(someprop.value());
+            }
+        }
+    }
+
+    @Test
+    public void testTriggerEventProcessor() throws Exception {
+        try (TriggerEventProcessor processor = new TriggerEventProcessor(); ) {
+            processor.init(
+                    Map.of(
+                            "when",
+                            "value.activator > 5",
+                            "destination",
+                            "other-topic",
+                            "fields",
+                            List.of(
+                                    Map.of(
+                                            "name",
+                                            "value.computed",
+                                            "expression",
+                                            "fn:uppercase(value.original)"),
+                                    Map.of(
+                                            "name",
+                                            "key.computed",
+                                            "expression",
+                                            "fn:lowercase(key.original)"),
+                                    Map.of(
+                                            "name",
+                                            "properties.computed",
+                                            "expression",
+                                            "fn:lowercase(properties.original)"))));
+
+            List<Record> emittedRecords = new CopyOnWriteArrayList<>();
+            TopicProducer dummyProducer =
+                    new TopicProducer() {
+                        @Override
+                        public long getTotalIn() {
+                            return 0;
+                        }
+
+                        @Override
+                        public CompletableFuture<?> write(Record record) {
+                            emittedRecords.add(record);
+                            return CompletableFuture.completedFuture(null);
+                        }
+                    };
+            TopicConnectionProvider dummyProvider =
+                    new TopicConnectionProvider() {
+                        @Override
+                        public TopicProducer createProducer(
+                                String agentId, String topic, Map<String, Object> config) {
+                            return dummyProducer;
+                        }
+                    };
+            AgentContext context = mock(AgentContext.class);
+            when(context.getTopicConnectionProvider()).thenReturn(dummyProvider);
+            processor.setContext(context);
+            processor.start();
+            List<Record> all = new ArrayList<>();
+            for (int i = 0; i < 10; i++) {
+                SimpleRecord someRecord =
+                        SimpleRecord.builder()
+                                .key(
+                                        """
+                        {"original": "Hello World %s"}
+                        """
+                                                .formatted(i))
+                                .value(
+                                        """
+                        {"original": "Hello Folks %s", "activator": %s}
+                        """
+                                                .formatted(i, i))
+                                .headers(
+                                        List.of(
+                                                new SimpleRecord.SimpleHeader(
+                                                        "original",
+                                                        "Some session id %s".formatted(i))))
+                                .build();
+
+                List<Record> read = new ArrayList<>();
+                processor.process(
+                        List.of(someRecord),
+                        (sourceRecordAndResult) ->
+                                read.addAll(sourceRecordAndResult.resultRecords()));
+                assertEquals(1, read.size());
+                all.addAll(read);
+                Record emittedToDownstream = read.get(0);
+                // the processor must pass downstream the original record
+                assertSame(emittedToDownstream, someRecord);
+                log.info("Received record {}", emittedToDownstream);
+
+                if (i > 5) {
+                    assertEquals(1, emittedRecords.size());
+                    Record r = emittedRecords.remove(0);
+                    Map<String, Object> value = (Map<String, Object>) r.value();
+                    assertEquals(
+                            "Hello Folks %s".formatted(i).toUpperCase(), value.get("computed"));
+                    Map<String, Object> key = (Map<String, Object>) r.key();
+                    assertEquals("Hello World %s".formatted(i).toLowerCase(), key.get("computed"));
+
+                    Header someprop = r.getHeader("computed");
+                    assertNotNull(someprop);
+                    assertEquals("Some session id %s".formatted(i).toLowerCase(), someprop.value());
+                } else {
+                    assertEquals(0, emittedRecords.size());
+                }
+            }
+            assertEquals(10, all.size());
+        }
+    }
+
+    @Test
+    public void testTriggerEventProcessorWithTopicProducerErrors() throws Exception {
+        try (TriggerEventProcessor processor = new TriggerEventProcessor(); ) {
+            processor.init(
+                    Map.of(
+                            "destination",
+                            "other-topic",
+                            "fields",
+                            List.of(Map.of("name", "value", "expression", "fn:uppercase(value)"))));
+
+            List<Record> emittedRecords = new CopyOnWriteArrayList<>();
+            TopicProducer dummyProducer =
+                    new TopicProducer() {
+                        @Override
+                        public long getTotalIn() {
+                            return 0;
+                        }
+
+                        @Override
+                        public CompletableFuture<?> write(Record record) {
+                            // here the value is UPPERCASE because the TopicProducer sees only the
+                            // record
+                            // to send to the side topic
+                            if (record.value().toString().contains("FAIL-ME")) {
+                                return CompletableFuture.failedFuture(
+                                        new IOException("Simulated error"));
+                            }
+                            emittedRecords.add(record);
+                            return CompletableFuture.completedFuture(null);
+                        }
+                    };
+            TopicConnectionProvider dummyProvider =
+                    new TopicConnectionProvider() {
+                        @Override
+                        public TopicProducer createProducer(
+                                String agentId, String topic, Map<String, Object> config) {
+                            return dummyProducer;
+                        }
+                    };
+            AgentContext context = mock(AgentContext.class);
+            when(context.getTopicConnectionProvider()).thenReturn(dummyProvider);
+            processor.setContext(context);
+            processor.start();
+
+            for (int i = 0; i < 10; i++) {
+                boolean fail = i % 2 == 0;
+                String content = (fail ? "fail-me" : "keep-me") + " " + i;
+                SimpleRecord someRecord = SimpleRecord.of(null, content);
+
+                List<AgentProcessor.SourceRecordAndResult> resultsForRecord = new ArrayList<>();
+                processor.process(List.of(someRecord), resultsForRecord::add);
+                // we always have an outcome
+                assertEquals(1, resultsForRecord.size());
+                // the processor must pass downstream the original record
+                Record emittedToDownstream = resultsForRecord.get(0).sourceRecord();
+                assertSame(emittedToDownstream, someRecord);
+
+                if (fail) {
+                    assertNotNull(resultsForRecord.get(0).error());
+                } else {
+                    assertNull(resultsForRecord.get(0).error());
+                    log.info("Received record {}", emittedToDownstream);
+                    assertEquals(1, emittedRecords.size());
+                    Record r = emittedRecords.remove(0);
+                    assertEquals(content.toUpperCase(), r.value());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testTriggerEventProcessorStopProcessing() throws Exception {
+        try (TriggerEventProcessor processor = new TriggerEventProcessor(); ) {
+            processor.init(
+                    Map.of(
+                            "continue-processing",
+                            false,
+                            "destination",
+                            "other-topic",
+                            "fields",
+                            List.of(
+                                    Map.of(
+                                            "name",
+                                            "value.computed",
+                                            "expression",
+                                            "fn:uppercase(value.original)"),
+                                    Map.of(
+                                            "name",
+                                            "key.computed",
+                                            "expression",
+                                            "fn:lowercase(key.original)"),
+                                    Map.of(
+                                            "name",
+                                            "properties.computed",
+                                            "expression",
+                                            "fn:lowercase(properties.original)"))));
+
+            List<Record> emittedRecords = new CopyOnWriteArrayList<>();
+            TopicProducer dummyProducer =
+                    new TopicProducer() {
+                        @Override
+                        public long getTotalIn() {
+                            return 0;
+                        }
+
+                        @Override
+                        public CompletableFuture<?> write(Record record) {
+                            emittedRecords.add(record);
+                            return CompletableFuture.completedFuture(null);
+                        }
+                    };
+            TopicConnectionProvider dummyProvider =
+                    new TopicConnectionProvider() {
+                        @Override
+                        public TopicProducer createProducer(
+                                String agentId, String topic, Map<String, Object> config) {
+                            return dummyProducer;
+                        }
+                    };
+            AgentContext context = mock(AgentContext.class);
+            when(context.getTopicConnectionProvider()).thenReturn(dummyProvider);
+            processor.setContext(context);
+            processor.start();
+
+            for (int i = 0; i < 10; i++) {
+                SimpleRecord someRecord =
+                        SimpleRecord.builder()
+                                .key(
+                                        """
+                        {"original": "Hello World %s"}
+                        """
+                                                .formatted(i))
+                                .value(
+                                        """
+                        {"original": "Hello Folks %s", "activator": %s}
+                        """
+                                                .formatted(i, i))
+                                .headers(
+                                        List.of(
+                                                new SimpleRecord.SimpleHeader(
+                                                        "original",
+                                                        "Some session id %s".formatted(i))))
+                                .build();
+
+                List<Record> read = new ArrayList<>();
+                processor.process(
+                        List.of(someRecord),
+                        (sourceRecordAndResult) ->
+                                read.addAll(sourceRecordAndResult.resultRecords()));
+                assertEquals(0, read.size());
+
+                assertEquals(1, emittedRecords.size());
+                Record r = emittedRecords.remove(0);
+                Map<String, Object> value = (Map<String, Object>) r.value();
+                assertEquals("Hello Folks %s".formatted(i).toUpperCase(), value.get("computed"));
+                Map<String, Object> key = (Map<String, Object>) r.key();
+                assertEquals("Hello World %s".formatted(i).toLowerCase(), key.get("computed"));
+
+                Header someprop = r.getHeader("computed");
+                assertNotNull(someprop);
+                assertEquals("Some session id %s".formatted(i).toLowerCase(), someprop.value());
+            }
+        }
+    }
+}
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunction.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunction.java
index 9616ebfa5..4ee7c8c28 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunction.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunction.java
@@ -15,10 +15,10 @@
  */
 package com.datastax.oss.pulsar.functions.transforms;
 
-import static ai.langstream.ai.agents.commons.TransformContext.attemptJsonConversion;
+import static ai.langstream.ai.agents.commons.MutableRecord.attemptJsonConversion;
 import static com.datastax.oss.streaming.ai.util.TransformFunctionUtil.buildStep;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import ai.langstream.ai.agents.services.impl.HuggingFaceProvider;
 import com.datastax.oss.streaming.ai.StepPredicatePair;
@@ -267,25 +267,25 @@ public Record<GenericObject> process(GenericObject input, Context context) throw
                     currentRecord);
         }
 
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(
                         context, nativeObject, transformConfig.isAttemptJsonConversion());
-        process(transformContext);
-        return send(context, transformContext);
+        process(mutableRecord);
+        return send(context, mutableRecord);
     }
 
-    public static TransformContext newTransformContext(
+    public static MutableRecord newTransformContext(
             Context context, Object value, boolean attemptJsonConversion) {
         Record<?> currentRecord = context.getCurrentRecord();
-        TransformContext transformContext = new TransformContext();
-        transformContext.setProperties(new HashMap<>());
-        transformContext.setInputTopic(currentRecord.getTopicName().orElse(null));
-        transformContext.setOutputTopic(currentRecord.getDestinationTopic().orElse(null));
-        transformContext.setKey(currentRecord.getKey().orElse(null));
-        transformContext.setEventTime(currentRecord.getEventTime().orElse(null));
+        MutableRecord mutableRecord = new MutableRecord();
+        mutableRecord.setProperties(new HashMap<>());
+        mutableRecord.setInputTopic(currentRecord.getTopicName().orElse(null));
+        mutableRecord.setOutputTopic(currentRecord.getDestinationTopic().orElse(null));
+        mutableRecord.setKey(currentRecord.getKey().orElse(null));
+        mutableRecord.setEventTime(currentRecord.getEventTime().orElse(null));
 
         if (currentRecord.getProperties() != null) {
-            transformContext.setProperties(new HashMap<>(currentRecord.getProperties()));
+            mutableRecord.setProperties(new HashMap<>(currentRecord.getProperties()));
         }
 
         Schema<?> schema = currentRecord.getSchema();
@@ -294,32 +294,31 @@ public static TransformContext newTransformContext(
             KeyValue<?, ?> kv = (KeyValue<?, ?>) value;
             Schema<?> keySchema = kvSchema.getKeySchema();
             Schema<?> valueSchema = kvSchema.getValueSchema();
-            transformContext.setKeySchemaType(pulsarSchemaToTransformSchemaType(keySchema));
-            transformContext.setKeyNativeSchema(getNativeSchema(keySchema));
-            transformContext.setKeyObject(
+            mutableRecord.setKeySchemaType(pulsarSchemaToTransformSchemaType(keySchema));
+            mutableRecord.setKeyNativeSchema(getNativeSchema(keySchema));
+            mutableRecord.setKeyObject(
                     keySchema.getSchemaInfo().getType().isStruct()
                             ? ((GenericObject) kv.getKey()).getNativeObject()
                             : kv.getKey());
-            transformContext.setValueSchemaType(pulsarSchemaToTransformSchemaType(valueSchema));
-            transformContext.setValueNativeSchema(getNativeSchema(valueSchema));
-            transformContext.setValueObject(
+            mutableRecord.setValueSchemaType(pulsarSchemaToTransformSchemaType(valueSchema));
+            mutableRecord.setValueNativeSchema(getNativeSchema(valueSchema));
+            mutableRecord.setValueObject(
                     valueSchema.getSchemaInfo().getType().isStruct()
                             ? ((GenericObject) kv.getValue()).getNativeObject()
                             : kv.getValue());
-            transformContext
+            mutableRecord
                     .getCustomContext()
                     .put("keyValueEncodingType", kvSchema.getKeyValueEncodingType());
         } else {
-            transformContext.setValueSchemaType(pulsarSchemaToTransformSchemaType(schema));
-            transformContext.setValueNativeSchema(getNativeSchema(schema));
-            transformContext.setValueObject(value);
+            mutableRecord.setValueSchemaType(pulsarSchemaToTransformSchemaType(schema));
+            mutableRecord.setValueNativeSchema(getNativeSchema(schema));
+            mutableRecord.setValueObject(value);
         }
         if (attemptJsonConversion) {
-            transformContext.setKeyObject(attemptJsonConversion(transformContext.getKeyObject()));
-            transformContext.setValueObject(
-                    attemptJsonConversion(transformContext.getValueObject()));
+            mutableRecord.setKeyObject(attemptJsonConversion(mutableRecord.getKeyObject()));
+            mutableRecord.setValueObject(attemptJsonConversion(mutableRecord.getValueObject()));
         }
-        return transformContext;
+        return mutableRecord;
     }
 
     private static Object getNativeSchema(Schema<?> schema) {
@@ -379,44 +378,44 @@ private static TransformSchemaType pulsarSchemaToTransformSchemaType(Schema<?> s
     }
 
     @Override
-    public void process(TransformContext transformContext) throws Exception {
-        TransformFunctionUtil.processTransformSteps(transformContext, steps);
+    public void process(MutableRecord mutableRecord) throws Exception {
+        TransformFunctionUtil.processTransformSteps(mutableRecord, steps);
     }
 
-    public static Record<GenericObject> send(Context context, TransformContext transformContext)
+    public static Record<GenericObject> send(Context context, MutableRecord mutableRecord)
             throws IOException {
-        if (transformContext.isDropCurrentRecord()) {
+        if (mutableRecord.isDropCurrentRecord()) {
             return null;
         }
-        transformContext.convertAvroToBytes();
-        transformContext.convertMapToStringOrBytes();
+        mutableRecord.convertAvroToBytes();
+        mutableRecord.convertMapToStringOrBytes();
 
         Schema outputSchema;
         Object outputObject;
-        if (transformContext.getKeySchemaType() != null) {
+        if (mutableRecord.getKeySchemaType() != null) {
             KeyValueEncodingType keyValueEncodingType =
                     (KeyValueEncodingType)
-                            transformContext.getCustomContext().get("keyValueEncodingType");
+                            mutableRecord.getCustomContext().get("keyValueEncodingType");
             outputSchema =
                     Schema.KeyValue(
                             buildSchema(
-                                    transformContext.getKeySchemaType(),
-                                    transformContext.getKeyNativeSchema()),
+                                    mutableRecord.getKeySchemaType(),
+                                    mutableRecord.getKeyNativeSchema()),
                             buildSchema(
-                                    transformContext.getValueSchemaType(),
-                                    transformContext.getValueNativeSchema()),
+                                    mutableRecord.getValueSchemaType(),
+                                    mutableRecord.getValueNativeSchema()),
                             keyValueEncodingType != null
                                     ? keyValueEncodingType
                                     : KeyValueEncodingType.INLINE);
-            Object outputKeyObject = transformContext.getKeyObject();
-            Object outputValueObject = transformContext.getValueObject();
+            Object outputKeyObject = mutableRecord.getKeyObject();
+            Object outputValueObject = mutableRecord.getValueObject();
             outputObject = new KeyValue<>(outputKeyObject, outputValueObject);
         } else {
             outputSchema =
                     buildSchema(
-                            transformContext.getValueSchemaType(),
-                            transformContext.getValueNativeSchema());
-            outputObject = transformContext.getValueObject();
+                            mutableRecord.getValueSchemaType(),
+                            mutableRecord.getValueNativeSchema());
+            outputObject = mutableRecord.getValueObject();
         }
 
         if (log.isDebugEnabled()) {
@@ -425,12 +424,12 @@ public static Record<GenericObject> send(Context context, TransformContext trans
 
         FunctionRecord.FunctionRecordBuilder<GenericObject> recordBuilder =
                 context.newOutputRecordBuilder(outputSchema)
-                        .destinationTopic(transformContext.getOutputTopic())
+                        .destinationTopic(mutableRecord.getOutputTopic())
                         .value(outputObject)
-                        .properties(transformContext.getProperties());
+                        .properties(mutableRecord.getProperties());
 
-        if (transformContext.getKeySchemaType() == null && transformContext.getKey() != null) {
-            recordBuilder.key(transformContext.getKey());
+        if (mutableRecord.getKeySchemaType() == null && mutableRecord.getKey() != null) {
+            recordBuilder.key(mutableRecord.getKey());
         }
 
         return recordBuilder.build();
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunctionTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunctionTest.java
index bdeac3827..af3b3e923 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunctionTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/pulsar/functions/transforms/TransformFunctionTest.java
@@ -21,7 +21,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import java.util.List;
@@ -539,19 +539,18 @@ void testComputeFilterList() throws Exception {
         Utils.TestRecord<String> testRecord = new Utils.TestRecord<>(Schema.STRING, value, null);
         Utils.TestContext context = new Utils.TestContext(testRecord, config);
         transformFunction.initialize(context);
-        TransformContext transformContext =
-                TransformFunction.newTransformContext(context, value, true);
-        transformFunction.process(transformContext);
+        MutableRecord mutableRecord = TransformFunction.newTransformContext(context, value, true);
+        transformFunction.process(mutableRecord);
         log.info(
                 "result: {} {}",
-                transformContext.getValueObject(),
-                transformContext.getValueObject().getClass());
+                mutableRecord.getValueObject(),
+                mutableRecord.getValueObject().getClass());
         assertEquals(
                 Map.of(
                         "queryResults",
                         List.of(Map.of("v1", "v2")),
                         "newQueryResults",
                         List.of(Map.of("v1", "v2"))),
-                transformContext.getValueObject());
+                mutableRecord.getValueObject());
     }
 }
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsTest.java
index ffea9f50e..5bf8bb059 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/ComputeAIEmbeddingsTest.java
@@ -18,7 +18,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.api.runner.code.SimpleRecord;
 import com.datastax.oss.streaming.ai.embeddings.MockEmbeddingsService;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -173,10 +173,10 @@ void testLoopOver() throws Exception {
                 }
                 """);
 
-        TransformContext transformContext = TransformContext.recordToTransformContext(record, true);
-        step.process(transformContext);
+        MutableRecord mutableRecord = MutableRecord.recordToMutableRecord(record, true);
+        step.process(mutableRecord);
         ai.langstream.api.runner.code.Record result =
-                TransformContext.transformContextToRecord(transformContext).orElseThrow();
+                MutableRecord.mutableRecordToRecord(mutableRecord).orElseThrow();
 
         Object value = result.value();
 
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/QueryStepTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/QueryStepTest.java
index 41436449a..c2d6b60fa 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/QueryStepTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/QueryStepTest.java
@@ -17,7 +17,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.api.runner.code.SimpleRecord;
 import com.datastax.oss.streaming.ai.datasource.QueryStepDataSource;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -82,18 +82,19 @@ public List<Map<String, Object>> fetchData(String query, List<Object> params) {
                                         "test-input-topic",
                                         42L);
                         assertEquals(params, expectedParams);
-                        return List.of(Map.of());
+                        return List.of(Map.of("foo", "bar"));
                     }
                 };
         QueryStep queryStep =
                 QueryStep.builder()
                         .dataSource(dataSource)
-                        .outputFieldName("value.result")
+                        .outputFieldName("value")
                         .query("select 1")
                         .fields(fields)
                         .build();
 
-        Utils.process(record, queryStep);
+        Record<Object> result = Utils.process(record, queryStep);
+        assertEquals(List.of(Map.of("foo", "bar")), result.getValue());
     }
 
     @Test
@@ -414,12 +415,12 @@ public List<Map<String, Object>> fetchData(String query, List<Object> params) {
                         .build();
         queryStep.start();
 
-        TransformContext context =
-                TransformContext.recordToTransformContext(SimpleRecord.of(null, value), true);
+        MutableRecord context =
+                MutableRecord.recordToMutableRecord(SimpleRecord.of(null, value), true);
 
         queryStep.process(context);
         ai.langstream.api.runner.code.Record record =
-                TransformContext.transformContextToRecord(context).orElseThrow();
+                MutableRecord.mutableRecordToRecord(context).orElseThrow();
         Map<String, Object> result = (Map<String, Object>) record.value();
         log.info("Result: {}", result);
 
@@ -471,12 +472,12 @@ public List<Map<String, Object>> fetchData(String query, List<Object> params) {
                         .build();
         queryStep.start();
 
-        TransformContext context =
-                TransformContext.recordToTransformContext(SimpleRecord.of(null, value), true);
+        MutableRecord context =
+                MutableRecord.recordToMutableRecord(SimpleRecord.of(null, value), true);
 
         queryStep.process(context);
         ai.langstream.api.runner.code.Record record =
-                TransformContext.transformContextToRecord(context).orElseThrow();
+                MutableRecord.mutableRecordToRecord(context).orElseThrow();
         Map<String, Object> result = (Map<String, Object>) record.value();
         log.info("Result: {}", result);
 
@@ -538,12 +539,12 @@ public List<Map<String, Object>> fetchData(String query, List<Object> params) {
                         .build();
         queryStep.start();
 
-        TransformContext context =
-                TransformContext.recordToTransformContext(SimpleRecord.of(null, value), true);
+        MutableRecord context =
+                MutableRecord.recordToMutableRecord(SimpleRecord.of(null, value), true);
 
         queryStep.process(context);
         ai.langstream.api.runner.code.Record record =
-                TransformContext.transformContextToRecord(context).orElseThrow();
+                MutableRecord.mutableRecordToRecord(context).orElseThrow();
         Map<String, Object> result = (Map<String, Object>) record.value();
         log.info("Result: {}", result);
 
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/Utils.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/Utils.java
index b4875e56a..d049e6dee 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/Utils.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/Utils.java
@@ -15,14 +15,14 @@
  */
 package com.datastax.oss.streaming.ai;
 
-import static ai.langstream.ai.agents.commons.TransformContext.attemptJsonConversion;
+import static ai.langstream.ai.agents.commons.MutableRecord.attemptJsonConversion;
 import static com.datastax.oss.streaming.ai.FlattenStep.AVRO_READ_OFFSET_PROP;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.TransformSchemaType;
 import com.datastax.oss.pulsar.functions.transforms.JsonNodeSchema;
 import java.io.ByteArrayOutputStream;
@@ -83,14 +83,14 @@ public class Utils {
     @SneakyThrows
     public static <T> Record<T> process(Record<GenericObject> record, TransformStep step) {
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
         try {
-            step.processAsync(transformContext).get();
+            step.processAsync(mutableRecord).get();
         } catch (ExecutionException error) {
             throw error.getCause();
         }
-        return send(context, transformContext);
+        return send(context, mutableRecord);
     }
 
     public static GenericData.Record getRecord(Schema<?> schema, byte[] value) throws IOException {
@@ -340,7 +340,7 @@ public static GenericAvroRecord createNestedAvroRecord(int levels) {
         return new GenericAvroRecord(new byte[0], nextLevelSchema, pulsarFields, nextLevelRecord);
     }
 
-    public static TransformContext createContextWithPrimitiveRecord(
+    public static MutableRecord createContextWithPrimitiveRecord(
             Schema<?> schema, Object value, String key) {
         Record<GenericObject> record =
                 new Utils.TestRecord<>(
@@ -384,22 +384,22 @@ static void assertNonOptionalField(
         assertEquals(record.get(fieldName), expectedValue);
     }
 
-    public static TransformContext newTransformContext(Context context, Object value) {
+    public static MutableRecord newTransformContext(Context context, Object value) {
         return newTransformContext(context, value, true);
     }
 
-    public static TransformContext newTransformContext(
+    public static MutableRecord newTransformContext(
             Context context, Object value, boolean attemptJsonConversion) {
         Record<?> currentRecord = context.getCurrentRecord();
-        TransformContext transformContext = new TransformContext();
-        transformContext.setProperties(new HashMap<>());
-        transformContext.setInputTopic(currentRecord.getTopicName().orElse(null));
-        transformContext.setOutputTopic(currentRecord.getDestinationTopic().orElse(null));
-        transformContext.setKey(currentRecord.getKey().orElse(null));
-        transformContext.setEventTime(currentRecord.getEventTime().orElse(null));
+        MutableRecord mutableRecord = new MutableRecord();
+        mutableRecord.setProperties(new HashMap<>());
+        mutableRecord.setInputTopic(currentRecord.getTopicName().orElse(null));
+        mutableRecord.setOutputTopic(currentRecord.getDestinationTopic().orElse(null));
+        mutableRecord.setKey(currentRecord.getKey().orElse(null));
+        mutableRecord.setEventTime(currentRecord.getEventTime().orElse(null));
 
         if (currentRecord.getProperties() != null) {
-            transformContext.setProperties(new HashMap<>(currentRecord.getProperties()));
+            mutableRecord.setProperties(new HashMap<>(currentRecord.getProperties()));
         }
 
         Schema<?> schema = currentRecord.getSchema();
@@ -408,32 +408,31 @@ public static TransformContext newTransformContext(
             KeyValue<?, ?> kv = (KeyValue<?, ?>) value;
             Schema<?> keySchema = kvSchema.getKeySchema();
             Schema<?> valueSchema = kvSchema.getValueSchema();
-            transformContext.setKeySchemaType(pulsarSchemaToTransformSchemaType(keySchema));
-            transformContext.setKeyNativeSchema(getNativeSchema(keySchema));
-            transformContext.setKeyObject(
+            mutableRecord.setKeySchemaType(pulsarSchemaToTransformSchemaType(keySchema));
+            mutableRecord.setKeyNativeSchema(getNativeSchema(keySchema));
+            mutableRecord.setKeyObject(
                     keySchema.getSchemaInfo().getType().isStruct()
                             ? ((GenericObject) kv.getKey()).getNativeObject()
                             : kv.getKey());
-            transformContext.setValueSchemaType(pulsarSchemaToTransformSchemaType(valueSchema));
-            transformContext.setValueNativeSchema(getNativeSchema(valueSchema));
-            transformContext.setValueObject(
+            mutableRecord.setValueSchemaType(pulsarSchemaToTransformSchemaType(valueSchema));
+            mutableRecord.setValueNativeSchema(getNativeSchema(valueSchema));
+            mutableRecord.setValueObject(
                     valueSchema.getSchemaInfo().getType().isStruct()
                             ? ((GenericObject) kv.getValue()).getNativeObject()
                             : kv.getValue());
-            transformContext
+            mutableRecord
                     .getCustomContext()
                     .put("keyValueEncodingType", kvSchema.getKeyValueEncodingType());
         } else {
-            transformContext.setValueSchemaType(pulsarSchemaToTransformSchemaType(schema));
-            transformContext.setValueNativeSchema(getNativeSchema(schema));
-            transformContext.setValueObject(value);
+            mutableRecord.setValueSchemaType(pulsarSchemaToTransformSchemaType(schema));
+            mutableRecord.setValueNativeSchema(getNativeSchema(schema));
+            mutableRecord.setValueObject(value);
         }
         if (attemptJsonConversion) {
-            transformContext.setKeyObject(attemptJsonConversion(transformContext.getKeyObject()));
-            transformContext.setValueObject(
-                    attemptJsonConversion(transformContext.getValueObject()));
+            mutableRecord.setKeyObject(attemptJsonConversion(mutableRecord.getKeyObject()));
+            mutableRecord.setValueObject(attemptJsonConversion(mutableRecord.getValueObject()));
         }
-        return transformContext;
+        return mutableRecord;
     }
 
     private static Object getNativeSchema(Schema<?> schema) {
@@ -492,40 +491,40 @@ private static TransformSchemaType pulsarSchemaToTransformSchemaType(Schema<?> s
         }
     }
 
-    public static <T> Record<T> send(Context context, TransformContext transformContext)
+    public static <T> Record<T> send(Context context, MutableRecord mutableRecord)
             throws IOException {
-        if (transformContext.isDropCurrentRecord()) {
+        if (mutableRecord.isDropCurrentRecord()) {
             return null;
         }
-        transformContext.convertAvroToBytes();
-        transformContext.convertMapToStringOrBytes();
+        mutableRecord.convertAvroToBytes();
+        mutableRecord.convertMapToStringOrBytes();
 
         Schema outputSchema;
         Object outputObject;
-        if (transformContext.getKeySchemaType() != null) {
+        if (mutableRecord.getKeySchemaType() != null) {
             KeyValueEncodingType keyValueEncodingType =
                     (KeyValueEncodingType)
-                            transformContext.getCustomContext().get("keyValueEncodingType");
+                            mutableRecord.getCustomContext().get("keyValueEncodingType");
             outputSchema =
                     Schema.KeyValue(
                             buildSchema(
-                                    transformContext.getKeySchemaType(),
-                                    transformContext.getKeyNativeSchema()),
+                                    mutableRecord.getKeySchemaType(),
+                                    mutableRecord.getKeyNativeSchema()),
                             buildSchema(
-                                    transformContext.getValueSchemaType(),
-                                    transformContext.getValueNativeSchema()),
+                                    mutableRecord.getValueSchemaType(),
+                                    mutableRecord.getValueNativeSchema()),
                             keyValueEncodingType != null
                                     ? keyValueEncodingType
                                     : KeyValueEncodingType.INLINE);
-            Object outputKeyObject = transformContext.getKeyObject();
-            Object outputValueObject = transformContext.getValueObject();
+            Object outputKeyObject = mutableRecord.getKeyObject();
+            Object outputValueObject = mutableRecord.getValueObject();
             outputObject = new KeyValue<>(outputKeyObject, outputValueObject);
         } else {
             outputSchema =
                     buildSchema(
-                            transformContext.getValueSchemaType(),
-                            transformContext.getValueNativeSchema());
-            outputObject = transformContext.getValueObject();
+                            mutableRecord.getValueSchemaType(),
+                            mutableRecord.getValueNativeSchema());
+            outputObject = mutableRecord.getValueObject();
         }
 
         if (log.isDebugEnabled()) {
@@ -534,12 +533,12 @@ public static <T> Record<T> send(Context context, TransformContext transformCont
 
         FunctionRecord.FunctionRecordBuilder<T> recordBuilder =
                 context.newOutputRecordBuilder(outputSchema)
-                        .destinationTopic(transformContext.getOutputTopic())
+                        .destinationTopic(mutableRecord.getOutputTopic())
                         .value(outputObject)
-                        .properties(transformContext.getProperties());
+                        .properties(mutableRecord.getProperties());
 
-        if (transformContext.getKeySchemaType() == null && transformContext.getKey() != null) {
-            recordBuilder.key(transformContext.getKey());
+        if (mutableRecord.getKeySchemaType() == null && mutableRecord.getKey() != null) {
+            recordBuilder.key(mutableRecord.getKey());
         }
 
         return recordBuilder.build();
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluatorTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluatorTest.java
index 51b6877d1..15f0f8c51 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluatorTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluatorTest.java
@@ -18,7 +18,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlEvaluator;
 import ai.langstream.ai.agents.commons.jstl.JstlFunctions;
 import com.datastax.oss.streaming.ai.Utils;
@@ -36,7 +36,7 @@ public class JstlEvaluatorTest {
 
     @ParameterizedTest
     @MethodSource("methodInvocationExpressionProvider")
-    void testMethodInvocationsDisabled(String expression, TransformContext context) {
+    void testMethodInvocationsDisabled(String expression, MutableRecord context) {
         assertThrows(
                 MethodNotFoundException.class,
                 () -> {
@@ -47,7 +47,7 @@ void testMethodInvocationsDisabled(String expression, TransformContext context)
 
     @ParameterizedTest
     @MethodSource("functionExpressionProvider")
-    void testFunctions(String expression, TransformContext context, Object expectedValue) {
+    void testFunctions(String expression, MutableRecord context, Object expectedValue) {
         assertEquals(
                 expectedValue,
                 new JstlEvaluator<>(String.format("${%s}", expression), Object.class)
@@ -56,7 +56,7 @@ void testFunctions(String expression, TransformContext context, Object expectedV
 
     @Test
     void testPrimitiveValue() {
-        TransformContext primitiveStringContext =
+        MutableRecord primitiveStringContext =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "");
 
         String value =
@@ -67,7 +67,7 @@ void testPrimitiveValue() {
 
     @Test
     void testNowFunction() {
-        TransformContext primitiveStringContext =
+        MutableRecord primitiveStringContext =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "");
 
         long expectedMillis = 123L;
@@ -82,7 +82,7 @@ void testNowFunction() {
 
     @Test
     void testTimestampAddFunctionsNow() {
-        TransformContext primitiveStringContext =
+        MutableRecord primitiveStringContext =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "");
 
         long nowMillis = 5000L;
@@ -100,7 +100,7 @@ void testTimestampAddFunctionsNow() {
      * @return {"expression", "transform context"}
      */
     public static Object[][] methodInvocationExpressionProvider() {
-        TransformContext primitiveStringContext =
+        MutableRecord primitiveStringContext =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "header-key");
 
         return new Object[][] {
@@ -116,13 +116,13 @@ public static Object[][] methodInvocationExpressionProvider() {
      * @return {"expression", "context", "expected value"}
      */
     public static Object[][] functionExpressionProvider() {
-        TransformContext primitiveBytesContext =
+        MutableRecord primitiveBytesContext =
                 Utils.createContextWithPrimitiveRecord(
                         Schema.BYTES, "Test-Message ".getBytes(StandardCharsets.UTF_8), "");
-        TransformContext primitiveInstantContext =
+        MutableRecord primitiveInstantContext =
                 Utils.createContextWithPrimitiveRecord(
                         Schema.INSTANT, Instant.parse("2017-01-02T00:01:02Z"), "");
-        TransformContext chronoUnitBytesContext =
+        MutableRecord chronoUnitBytesContext =
                 Utils.createContextWithPrimitiveRecord(
                         Schema.BYTES, "millis".getBytes(StandardCharsets.UTF_8), "");
         long millis = Instant.parse("2017-01-02T00:01:02Z").toEpochMilli();
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlTransformContextAdapterTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlMutableRecordAdapterTest.java
similarity index 94%
rename from langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlTransformContextAdapterTest.java
rename to langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlMutableRecordAdapterTest.java
index f58be475d..dd233c288 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlTransformContextAdapterTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlMutableRecordAdapterTest.java
@@ -20,7 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.JstlTransformContextAdapter;
 import com.datastax.oss.streaming.ai.Utils;
 import java.time.LocalDate;
@@ -42,7 +42,7 @@
 import org.apache.pulsar.functions.api.Record;
 import org.junit.jupiter.api.Test;
 
-public class JstlTransformContextAdapterTest {
+public class JstlMutableRecordAdapterTest {
     private static final org.apache.avro.Schema dateType =
             LogicalTypes.date()
                     .addToSchema(org.apache.avro.Schema.create(org.apache.avro.Schema.Type.INT));
@@ -58,11 +58,11 @@ void testAdapterForKeyValueRecord() {
          }
         */
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertTrue(adapter.getKey() instanceof Map);
@@ -89,11 +89,11 @@ void testAdapterForPrimitiveKeyValueRecord() {
                 Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.SEPARATED);
         KeyValue<String, Integer> keyValue = new KeyValue<>("key", 42);
 
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 Utils.createContextWithPrimitiveRecord(keyValueSchema, keyValue, "header-key");
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         assertEquals(adapter.getKey(), "key");
         assertEquals(adapter.adaptValue(), 42);
@@ -103,11 +103,11 @@ void testAdapterForPrimitiveKeyValueRecord() {
     @Test
     void testAdapterForPrimitiveRecord() {
         // given
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "header-key");
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertEquals(adapter.getHeader().get("messageKey"), "header-key");
@@ -130,11 +130,11 @@ void testAdapterForNestedValueRecord() {
          "level4Union": "level4_2" } } } }
         */
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertEquals(adapter.getHeader().get("messageKey"), "header-key");
@@ -160,11 +160,11 @@ void testAdapterForNestedKeyValueRecord() {
          "level4Union": "level4_2" } } } }
         */
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertTrue(adapter.getKey() instanceof Map);
@@ -194,11 +194,11 @@ void testAdapterForRecordHeaders() {
                         .build();
 
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertEquals(adapter.getHeader().get("messageKey"), "test-key");
@@ -238,11 +238,11 @@ public void testAdapterForLogicalTypes() {
         Record<GenericObject> record =
                 new Utils.TestRecord<>(pulsarValueSchema, valueRecord, "key");
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
         // when
-        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(transformContext);
+        JstlTransformContextAdapter adapter = new JstlTransformContextAdapter(mutableRecord);
 
         // then
         assertTrue(adapter.adaptValue() instanceof Map);
diff --git a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/predicate/JstlPredicateTest.java b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/predicate/JstlPredicateTest.java
index cf97dc269..0fb6f351e 100644
--- a/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/predicate/JstlPredicateTest.java
+++ b/langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/predicate/JstlPredicateTest.java
@@ -19,7 +19,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-import ai.langstream.ai.agents.commons.TransformContext;
+import ai.langstream.ai.agents.commons.MutableRecord;
 import ai.langstream.ai.agents.commons.jstl.predicate.JstlPredicate;
 import com.datastax.oss.streaming.ai.Utils;
 import java.util.HashMap;
@@ -41,10 +41,10 @@ void testKeyValueAvro(String when, boolean match) {
 
         Record<GenericObject> record = Utils.createNestedAvroKeyValueRecord(2);
         Utils.TestContext context = new Utils.TestContext(record, new HashMap<>());
-        TransformContext transformContext =
+        MutableRecord mutableRecord =
                 newTransformContext(context, record.getValue().getNativeObject());
 
-        assertEquals(predicate.test(transformContext), match);
+        assertEquals(predicate.test(mutableRecord), match);
     }
 
     @Test
@@ -61,14 +61,14 @@ void testInvalidWhen() {
 
     @ParameterizedTest
     @MethodSource("primitiveKeyValuePredicates")
-    void testPrimitiveKeyValueAvro(String when, TransformContext context, boolean match) {
+    void testPrimitiveKeyValueAvro(String when, MutableRecord context, boolean match) {
         JstlPredicate predicate = new JstlPredicate(when);
         assertEquals(predicate.test(context), match);
     }
 
     @ParameterizedTest
     @MethodSource("primitivePredicates")
-    void testPrimitiveValueAvro(String when, TransformContext context, boolean match) {
+    void testPrimitiveValueAvro(String when, MutableRecord context, boolean match) {
         JstlPredicate predicate = new JstlPredicate(when);
         assertEquals(predicate.test(context), match);
     }
@@ -82,7 +82,7 @@ public static Object[][] primitiveKeyValuePredicates() {
 
         KeyValue<String, Integer> keyValue = new KeyValue<>("key", 42);
 
-        TransformContext primitiveKVContext =
+        MutableRecord primitiveKVContext =
                 Utils.createContextWithPrimitiveRecord(keyValueSchema, keyValue, "");
 
         return new Object[][] {
@@ -101,11 +101,11 @@ public static Object[][] primitivePredicates() {
                 Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.SEPARATED);
         KeyValue<String, Integer> keyValue = new KeyValue<>("key", 42);
 
-        TransformContext primitiveStringContext =
+        MutableRecord primitiveStringContext =
                 Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "header-key");
-        TransformContext primitiveIntContext =
+        MutableRecord primitiveIntContext =
                 Utils.createContextWithPrimitiveRecord(Schema.INT32, 33, "header-key");
-        TransformContext primitiveKVContext =
+        MutableRecord primitiveKVContext =
                 Utils.createContextWithPrimitiveRecord(keyValueSchema, keyValue, "header-key");
 
         return new Object[][] {
diff --git a/langstream-core/src/test/java/ai/langstream/impl/uti/ClassConfigValidatorTest.java b/langstream-core/src/test/java/ai/langstream/impl/uti/ClassConfigValidatorTest.java
index 743a1a08c..ebb43a4d1 100644
--- a/langstream-core/src/test/java/ai/langstream/impl/uti/ClassConfigValidatorTest.java
+++ b/langstream-core/src/test/java/ai/langstream/impl/uti/ClassConfigValidatorTest.java
@@ -19,6 +19,7 @@
 import ai.langstream.api.doc.AgentConfigurationModel;
 import ai.langstream.api.doc.ConfigProperty;
 import ai.langstream.api.doc.ConfigPropertyIgnore;
+import ai.langstream.api.doc.ExtendedValidationType;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyDescription;
@@ -49,6 +50,9 @@ public static class MyClass {
         @ConfigProperty(required = true)
         private List<MyClass2> theList;
 
+        @ConfigProperty(extendedValidationType = ExtendedValidationType.EL_EXPRESSION)
+        private String expression;
+
         @ConfigProperty(required = true)
         private Map<String, MyClass2> theMap;
 
@@ -74,61 +78,66 @@ public void testParseClass() throws Exception {
 
         Assertions.assertEquals(
                 """
-                        {
-                          "name" : "my super agent",
-                          "description" : "this agent is AWESOME!",
-                          "properties" : {
-                            "doNotIgnoreMe" : {
-                              "required" : false
-                            },
-                            "my-int" : {
-                              "description" : "my description",
-                              "required" : true,
-                              "type" : "integer",
-                              "defaultValue" : "11"
-                            },
-                            "myclass2" : {
-                              "required" : false,
-                              "type" : "object",
-                              "properties" : {
-                                "innerDoc" : {
-                                  "description" : "my description",
-                                  "required" : true,
-                                  "type" : "string",
-                                  "defaultValue" : "110.1a"
-                                },
-                                "inner" : {
-                                  "required" : false,
-                                  "type" : "string"
-                                }
-                              }
-                            },
-                            "theList" : {
-                              "required" : true,
-                              "type" : "array",
-                              "items" : {
-                                "required" : true,
-                                "type" : "object",
+                              {
+                                "name" : "my super agent",
+                                "description" : "this agent is AWESOME!",
                                 "properties" : {
-                                  "innerDoc" : {
+                                  "doNotIgnoreMe" : {
+                                    "required" : false
+                                  },
+                                  "expression" : {
+                                    "required" : false,
+                                    "type" : "string",
+                                    "extendedValidationType" : "EL_EXPRESSION"
+                                  },
+                                  "my-int" : {
                                     "description" : "my description",
                                     "required" : true,
-                                    "type" : "string",
-                                    "defaultValue" : "110.1a"
+                                    "type" : "integer",
+                                    "defaultValue" : "11"
                                   },
-                                  "inner" : {
+                                  "myclass2" : {
                                     "required" : false,
-                                    "type" : "string"
+                                    "type" : "object",
+                                    "properties" : {
+                                      "innerDoc" : {
+                                        "description" : "my description",
+                                        "required" : true,
+                                        "type" : "string",
+                                        "defaultValue" : "110.1a"
+                                      },
+                                      "inner" : {
+                                        "required" : false,
+                                        "type" : "string"
+                                      }
+                                    }
+                                  },
+                                  "theList" : {
+                                    "required" : true,
+                                    "type" : "array",
+                                    "items" : {
+                                      "required" : true,
+                                      "type" : "object",
+                                      "properties" : {
+                                        "innerDoc" : {
+                                          "description" : "my description",
+                                          "required" : true,
+                                          "type" : "string",
+                                          "defaultValue" : "110.1a"
+                                        },
+                                        "inner" : {
+                                          "required" : false,
+                                          "type" : "string"
+                                        }
+                                      }
+                                    }
+                                  },
+                                  "theMap" : {
+                                    "required" : true,
+                                    "type" : "object"
                                   }
                                 }
-                              }
-                            },
-                            "theMap" : {
-                              "required" : true,
-                              "type" : "object"
-                            }
-                          }
-                        }""",
+                              }""",
                 new ObjectMapper()
                         .configure(SerializationFeature.INDENT_OUTPUT, true)
                         .setSerializationInclusion(
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
index 0a0c33732..58347c6e4 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
@@ -48,12 +48,14 @@
 import lombok.extern.slf4j.Slf4j;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 @Slf4j
 public abstract class AbstractApplicationRunner {
 
+    private static final int DEDAULT_NUM_LOOPS = 5;
     public static final Path agentsDirectory;
 
     static {
@@ -67,14 +69,16 @@ public abstract class AbstractApplicationRunner {
     private static NarFileHandler narFileHandler;
     private static TopicConnectionsRuntimeRegistry topicConnectionsRuntimeRegistry;
 
-    private Path codeDirectory;
+    private static Path codeDirectory;
 
-    public Path getCodeDirectory() {
-        return codeDirectory;
+    private int maxNumLoops = DEDAULT_NUM_LOOPS;
+
+    public int getMaxNumLoops() {
+        return maxNumLoops;
     }
 
-    public void setCodeDirectory(Path codeDirectory) {
-        this.codeDirectory = codeDirectory;
+    public void setMaxNumLoops(int maxNumLoops) {
+        this.maxNumLoops = maxNumLoops;
     }
 
     protected record ApplicationRuntime(
@@ -140,7 +144,7 @@ protected ApplicationRuntime deployApplicationWithSecrets(
 
     @BeforeAll
     public static void setup() throws Exception {
-        Path codeDirectory = Paths.get("target/test-jdbc-drivers");
+        codeDirectory = Paths.get("target/test-jdbc-drivers");
         narFileHandler =
                 new NarFileHandler(
                         agentsDirectory,
@@ -212,8 +216,9 @@ protected AgentRunResult executeAgentRunners(ApplicationRuntime runtime) throws
                                         agentsDirectory,
                                         agentInfo,
                                         () -> {
-                                            log.info("Num loops {}", numLoops.get());
-                                            return numLoops.incrementAndGet() <= 10;
+                                            log.info(
+                                                    "Num loops {}/{}", numLoops.get(), maxNumLoops);
+                                            return numLoops.incrementAndGet() <= maxNumLoops;
                                         },
                                         () -> validateAgentInfoBeforeStop(agentInfo),
                                         false,
@@ -266,6 +271,11 @@ protected AgentRunResult executeAgentRunners(ApplicationRuntime runtime) throws
 
     protected void validateAgentInfoBeforeStop(AgentInfo agentInfo) {}
 
+    @AfterEach
+    public void resetNumLoops() {
+        setMaxNumLoops(DEDAULT_NUM_LOOPS);
+    }
+
     @AfterAll
     public static void teardown() {
         if (applicationDeployer != null) {
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/AbstractKafkaApplicationRunner.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/AbstractKafkaApplicationRunner.java
index 85b560c8c..d6307a601 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/AbstractKafkaApplicationRunner.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/AbstractKafkaApplicationRunner.java
@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.AdminClient;
@@ -113,7 +114,9 @@ protected void sendMessage(
         producer.flush();
     }
 
-    protected List<ConsumerRecord> waitForMessages(KafkaConsumer consumer, List<?> expected) {
+    protected List<ConsumerRecord> waitForMessages(
+            KafkaConsumer consumer,
+            BiConsumer<List<ConsumerRecord>, List<Object>> assertionOnReceivedMessages) {
         List<ConsumerRecord> result = new ArrayList<>();
         List<Object> received = new ArrayList<>();
 
@@ -131,25 +134,33 @@ protected List<ConsumerRecord> waitForMessages(KafkaConsumer consumer, List<?> e
                             log.info("Result:  {}", received);
                             received.forEach(r -> log.info("Received |{}|", r));
 
-                            assertEquals(expected.size(), received.size());
-                            for (int i = 0; i < expected.size(); i++) {
-                                Object expectedValue = expected.get(i);
-                                Object actualValue = received.get(i);
-                                if (expectedValue instanceof Consumer fn) {
-                                    fn.accept(actualValue);
-                                } else if (expectedValue instanceof byte[]) {
-                                    assertArrayEquals((byte[]) expectedValue, (byte[]) actualValue);
-                                } else {
-                                    log.info("expected: {}", expectedValue);
-                                    log.info("got: {}", actualValue);
-                                    assertEquals(expectedValue, actualValue);
-                                }
-                            }
+                            assertionOnReceivedMessages.accept(result, received);
                         });
 
         return result;
     }
 
+    protected List<ConsumerRecord> waitForMessages(KafkaConsumer consumer, List<?> expected) {
+        return waitForMessages(
+                consumer,
+                (result, received) -> {
+                    assertEquals(expected.size(), received.size());
+                    for (int i = 0; i < expected.size(); i++) {
+                        Object expectedValue = expected.get(i);
+                        Object actualValue = received.get(i);
+                        if (expectedValue instanceof Consumer fn) {
+                            fn.accept(actualValue);
+                        } else if (expectedValue instanceof byte[]) {
+                            assertArrayEquals((byte[]) expectedValue, (byte[]) actualValue);
+                        } else {
+                            log.info("expected: {}", expectedValue);
+                            log.info("got: {}", actualValue);
+                            assertEquals(expectedValue, actualValue);
+                        }
+                    }
+                });
+    }
+
     protected List<ConsumerRecord> waitForMessagesInAnyOrder(
             KafkaConsumer consumer, Collection<String> expected) {
         List<ConsumerRecord> result = new ArrayList<>();
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/FlowControlRunnerIT.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/FlowControlRunnerIT.java
index 5da32bacd..a2923f26d 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/FlowControlRunnerIT.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/FlowControlRunnerIT.java
@@ -15,9 +15,18 @@
  */
 package ai.langstream.kafka;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.common.header.internals.RecordHeader;
@@ -29,7 +38,7 @@
 class FlowControlRunnerIT extends AbstractKafkaApplicationRunner {
 
     @Test
-    public void testSimpleFlowControl() throws Exception {
+    public void testDispatch() throws Exception {
         String tenant = "tenant";
         String[] expectedAgents = {"app-step1"};
 
@@ -99,7 +108,7 @@ public void testSimpleFlowControl() throws Exception {
      * "output" the processor behaves like a sink (no output)
      */
     @Test
-    public void testSimpleFlowControlNoDefaultDestination() throws Exception {
+    public void testDispatchNoDefaultDestination() throws Exception {
         String tenant = "tenant";
         String[] expectedAgents = {"app-step1"};
 
@@ -161,7 +170,7 @@ public void testSimpleFlowControlNoDefaultDestination() throws Exception {
     }
 
     @Test
-    public void testSimpleFlowControlDefaultToAnotherAgent() throws Exception {
+    public void testDispatchDefaultToAnotherAgent() throws Exception {
         String tenant = "tenant";
         String[] expectedAgents = {"app-step1"};
 
@@ -202,7 +211,6 @@ public void testSimpleFlowControlDefaultToAnotherAgent() throws Exception {
                                            expression: "'modified'"
                                 """);
 
-        // query the database with re-rank
         try (ApplicationRuntime applicationRuntime =
                 deployApplication(
                         tenant, "app", application, buildInstanceYaml(), expectedAgents)) {
@@ -234,4 +242,124 @@ public void testSimpleFlowControlDefaultToAnotherAgent() throws Exception {
             }
         }
     }
+
+    @Test
+    public void testTimerSource() throws Exception {
+        setMaxNumLoops(10);
+        String tenant = "tenant";
+        String[] expectedAgents = {"app-step1"};
+
+        Map<String, String> application =
+                Map.of(
+                        "module.yaml",
+                        """
+                                topics:
+                                  - name: "timer-source-output-topic"
+                                    creation-mode: create-if-not-exists
+                                pipeline:
+                                  - name: "Timer"
+                                    type: "timer-source"
+                                    id: step1
+                                    output: timer-source-output-topic
+                                    configuration:
+                                      period-seconds: 1
+                                      fields:
+                                         - name: "key.id"
+                                           expression: "fn:uuid()"
+                                         - name: "value.stringpayload"
+                                           expression: "'constant-payload'"
+                                         - name: "value.intpayload"
+                                           expression: "42"
+                                         - name: "properties.foo"
+                                           expression: "'bar'"
+                                """);
+        try (ApplicationRuntime applicationRuntime =
+                deployApplication(
+                        tenant, "app", application, buildInstanceYaml(), expectedAgents)) {
+            try (KafkaConsumer<String, String> consumer =
+                    createConsumer("timer-source-output-topic"); ) {
+                executeAgentRunners(applicationRuntime);
+                waitForMessages(
+                        consumer,
+                        new BiConsumer<List<ConsumerRecord>, List<Object>>() {
+                            @Override
+                            @SneakyThrows
+                            public void accept(
+                                    List<ConsumerRecord> consumerRecords, List<Object> objects) {
+                                assertTrue(objects.size() > 1);
+                                Object o = objects.get(0);
+                                log.info("Received {}", o);
+                                assertEquals(
+                                        "{\"intpayload\":42,\"stringpayload\":\"constant-payload\"}",
+                                        o);
+                                ConsumerRecord consumerRecord = consumerRecords.get(0);
+                                Object key = consumerRecord.key();
+                                log.info("Received key {}", key);
+                                Map<String, Object> jsonKey =
+                                        new ObjectMapper().readValue(key.toString(), Map.class);
+                                // try to parse the UUID
+                                UUID.fromString(jsonKey.get("id").toString());
+                                assertEquals(
+                                        "bar",
+                                        new String(
+                                                consumerRecord.headers().lastHeader("foo").value(),
+                                                StandardCharsets.UTF_8));
+                            }
+                        });
+            }
+        }
+    }
+
+    @Test
+    public void testTriggerEventProcessor() throws Exception {
+        String tenant = "tenant";
+        String[] expectedAgents = {"app-step1"};
+
+        Map<String, String> application =
+                Map.of(
+                        "module.yaml",
+                        """
+                                topics:
+                                  - name: "input-topic-splitter"
+                                    creation-mode: create-if-not-exists
+                                  - name: "side-topic"
+                                    creation-mode: create-if-not-exists
+                                  - name: "output-topic-chunks"
+                                    creation-mode: create-if-not-exists
+                                pipeline:
+                                  - name: "Chunk some text"
+                                    id: step1
+                                    type: "text-splitter"
+                                    input: input-topic-splitter
+                                    configuration:
+                                      chunk_size: 5
+                                      chunk_overlap: 0
+                                  - name: "Trigger event on last chunk"
+                                    type: "trigger-event"
+                                    output: output-topic-chunks
+                                    configuration:
+                                      destination: side-topic
+                                      when: fn:toInt(properties.text_num_chunks) == (fn:toInt(properties.chunk_id) + 1)
+                                      fields:
+                                         - name: "properties.foo"
+                                           expression: "'bar'"
+                                """);
+        try (ApplicationRuntime applicationRuntime =
+                deployApplication(
+                        tenant, "app", application, buildInstanceYaml(), expectedAgents)) {
+            try (KafkaProducer<String, String> producer = createProducer();
+                    KafkaConsumer<String, String> consumer = createConsumer("output-topic-chunks");
+                    KafkaConsumer<String, String> consumerSideTopic =
+                            createConsumer("side-topic"); ) {
+
+                sendMessage("input-topic-splitter", "some very long text. end", producer);
+
+                executeAgentRunners(applicationRuntime);
+                waitForMessages(consumer, List.of("some very long", "text. end"));
+
+                // the side topic receives only the last chunk
+                waitForMessages(consumerSideTopic, List.of("text. end"));
+            }
+        }
+    }
 }
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/TextProcessingAgentsRunnerIT.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/TextProcessingAgentsRunnerIT.java
index e5cd39140..22f64a224 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/TextProcessingAgentsRunnerIT.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/TextProcessingAgentsRunnerIT.java
@@ -188,9 +188,9 @@ public void testSplitThenJson(boolean useIntermediateTopic) throws Exception {
                 waitForMessages(
                         consumer,
                         List.of(
-                                "{\"chunk_text_length\":\"47\",\"text\":\"This text is written in English, but it is very\",\"chunk_id\":\"0\",\"chunk_num_tokens\":\"47\"}",
-                                "{\"chunk_text_length\":\"5\",\"text\":\"long,\",\"chunk_id\":\"1\",\"chunk_num_tokens\":\"5\"}",
-                                "{\"chunk_text_length\":\"40\",\"text\":\"so you may want to split it into chunks.\",\"chunk_id\":\"2\",\"chunk_num_tokens\":\"40\"}"));
+                                "{\"chunk_text_length\":\"47\",\"text\":\"This text is written in English, but it is very\",\"text_num_chunks\":\"3\",\"chunk_id\":\"0\",\"chunk_num_tokens\":\"47\"}",
+                                "{\"chunk_text_length\":\"5\",\"text\":\"long,\",\"text_num_chunks\":\"3\",\"chunk_id\":\"1\",\"chunk_num_tokens\":\"5\"}",
+                                "{\"chunk_text_length\":\"40\",\"text\":\"so you may want to split it into chunks.\",\"text_num_chunks\":\"3\",\"chunk_id\":\"2\",\"chunk_num_tokens\":\"40\"}"));
             }
         }
     }