Add more tests and add back validation for some agents

langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/AbstractGrpcAgent.java

@@ -99,7 +99,10 @@ public void start() throws Exception {
             throw new IllegalStateException("Channel not initialized");
         }
         blockingStub =
-                AgentServiceGrpc.newBlockingStub(channel).withDeadlineAfter(30, TimeUnit.SECONDS);
+                AgentServiceGrpc.newBlockingStub(channel)
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE)
+                        .withDeadlineAfter(30, TimeUnit.SECONDS);
         asyncStub =
                 AgentServiceGrpc.newStub(channel)
                         .withWaitForReady()

langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/GrpcAgentProcessor.java

@@ -59,7 +59,12 @@ public synchronized void onNewSchemaToSend(Schema schema) {
     @Override
     public void start() throws Exception {
         super.start();
-        request = AgentServiceGrpc.newStub(channel).withWaitForReady().process(responseObserver);
+        request =
+                AgentServiceGrpc.newStub(channel)
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE)
+                        .withWaitForReady()
+                        .process(responseObserver);
         restarting.set(false);
         startFailedButDevelopmentMode = false;
     }

langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/GrpcAgentSink.java

@@ -53,7 +53,12 @@ public void onNewSchemaToSend(Schema schema) {
     @Override
     public void start() throws Exception {
         super.start();
-        request = AgentServiceGrpc.newStub(channel).withWaitForReady().write(responseObserver);
+        request =
+                AgentServiceGrpc.newStub(channel)
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE)
+                        .withWaitForReady()
+                        .write(responseObserver);
         restarting.set(false);
         startFailedButDevelopmentMode = false;
     }

langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/GrpcAgentSource.java

@@ -53,7 +53,12 @@ public void onNewSchemaToSend(Schema schema) {
     @Override
     public void start() throws Exception {
         super.start();
-        request = AgentServiceGrpc.newStub(channel).withWaitForReady().read(responseObserver);
+        request =
+                AgentServiceGrpc.newStub(channel)
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE)
+                        .withWaitForReady()
+                        .read(responseObserver);
         restarting.set(false);
         startFailedButDevelopmentMode = false;
     }

langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/PythonGrpcServer.java

@@ -90,7 +90,10 @@ public ManagedChannel start() throws Exception {
         ManagedChannel channel =
                 ManagedChannelBuilder.forAddress("localhost", port).usePlaintext().build();
         AgentServiceGrpc.AgentServiceBlockingStub stub =
-                AgentServiceGrpc.newBlockingStub(channel).withDeadlineAfter(30, TimeUnit.SECONDS);
+                AgentServiceGrpc.newBlockingStub(channel)
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE)
+                        .withDeadlineAfter(30, TimeUnit.SECONDS);
         for (int i = 0; ; i++) {
             try {
                 stub.agentInfo(Empty.getDefaultInstance());

langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlEvaluator.java

@@ -45,6 +45,9 @@ public JstlEvaluator(String expression, Class<? extends T> type) {
 
     @SneakyThrows
     private void registerFunctions() {
+        this.expressionContext
+                .getFunctionMapper()
+                .mapFunction("fn", "length", JstlFunctions.class.getMethod("length", Object.class));
         this.expressionContext
                 .getFunctionMapper()
                 .mapFunction("fn", "toJson", JstlFunctions.class.getMethod("toJson", Object.class));

langstream-agents/langstream-agents-commons/src/main/java/ai/langstream/ai/agents/commons/jstl/JstlFunctions.java

@@ -187,6 +187,10 @@ public static Map<String, Object> emptyMap() {
         return Map.of();
     }
 
+    public static long length(Object o) {
+        return o == null ? 0 : toString(o).length();
+    }
+
     public static Map<String, Object> mapOf(Object... field) {
         Map<String, Object> result = new HashMap<>();
         for (int i = 0; i < field.length; i += 2) {

langstream-agents/langstream-ai-agents/src/test/java/com/datastax/oss/streaming/ai/jstl/JstlEvaluatorTest.java

@@ -65,6 +65,18 @@ void testPrimitiveValue() {
         assertEquals("test-message", value);
     }
 
+    @Test
+    void testLength() {
+        MutableRecord primitiveStringContext =
+                Utils.createContextWithPrimitiveRecord(Schema.STRING, "test-message", "");
+
+        String value =
+                new JstlEvaluator<>("${fn:length(value)}", String.class)
+                        .evaluate(primitiveStringContext);
+
+        assertEquals("12", value);
+    }
+
     @Test
     void testNowFunction() {
         MutableRecord primitiveStringContext =

langstream-e2e-tests/src/test/java/ai/langstream/tests/PythonAgentsIT.java

@@ -24,6 +24,7 @@
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.shaded.com.google.common.base.Strings;
 
 @Slf4j
 @ExtendWith(BaseEndToEndTest.class)
@@ -115,8 +116,12 @@ public void testSource() {
                                 .formatted(applicationId)
                                 .split(" "));
         log.info("Output: {}", output);
+        String bigPayload = Strings.repeat("test", 10000);
+        String value = "the length is " + bigPayload.length();
         Assertions.assertTrue(
-                output.contains("{\"record\":{\"key\":null,\"value\":\"test\",\"headers\":{}}"));
+                output.contains(
+                        "{\"record\":{\"key\":null,\"value\":\"" + value + "\",\"headers\":{}}"),
+                "Output doesn't contain the expected payload: " + output);
 
         deleteAppAndAwaitCleanup(tenant, applicationId);
     }

langstream-e2e-tests/src/test/resources/apps/python-source/pipeline.yaml

@@ -22,12 +22,19 @@ topics:
     creation-mode: create-if-not-exists
     schema:
       type: string
+resources:
+  size: 2
 pipeline:
   - name: "Source using Python"
-    resources:
-      size: 2
     id: "test-python-source"
     type: "python-source"
+    configuration:
+      className: example.TestSource
+  - name: "Compute length"
+    id: "compute-length"
+    type: "compute"
     output: ls-test-output
     configuration:
-      className: example.TestSource
+      fields:
+        - name: "value"
+          expression: "fn:concat('the length is ', fn:len(value))"

langstream-e2e-tests/src/test/resources/apps/python-source/python/example.py

@@ -26,7 +26,7 @@ def read(self):
         if not self.sent:
             logging.info("Sending the record")
             self.sent = True
-            return [SimpleRecord("test")]
+            return [SimpleRecord("test" * 10000)]
         return []
 
     def commit(self, records):