Repository: LangStream/langstream (language: python)

Issue:
org.apache.avro.SchemaParseException: Illegal character in: query-result
When using JSON (not AVRO!) schema, some agents will apply the result using a field name configured in the pipeline. 
In many examples there is `query-result`, however AVRO doesn't support field names with dashes in it. To change the JSON schema, we use avro internally so then this happen:

```
[convhisthuman-pipeline-query-1-0] 08:01:12.765 [main] ERROR a.l.runtime.agent.AgentRunner -- Unrecoverable error while processing some the records, failing
[convhisthuman-pipeline-query-1-0] java.util.concurrent.CompletionException: org.apache.avro.SchemaParseException: Illegal character in: query-result
[convhisthuman-pipeline-query-1-0]      at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332)
[convhisthuman-pipeline-query-1-0]      at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:674)
[convhisthuman-pipeline-query-1-0]      at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662)
[convhisthuman-pipeline-query-1-0]      at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.ai.agents.GenAIToolKitAgent.processRecord(GenAIToolKitAgent.java:118)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.ai.agents.GenAIToolKitAgent.process(GenAIToolKitAgent.java:93)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader$3.lambda$process$0(AgentCodeAndLoader.java:208)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader.executeNoExceptionWithContextClassloader(AgentCodeAndLoader.java:56)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader$3.process(AgentCodeAndLoader.java:207)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.CompositeAgentProcessor.invokeProcessor(CompositeAgentProcessor.java:165)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.CompositeAgentProcessor.process(CompositeAgentProcessor.java:225)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader$3.lambda$process$0(AgentCodeAndLoader.java:208)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader.executeNoExceptionWithContextClassloader(AgentCodeAndLoader.java:56)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.api.runner.code.AgentCodeAndLoader$3.process(AgentCodeAndLoader.java:207)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunner.runProcessorAgent(AgentRunner.java:790)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunner.runMainLoop(AgentRunner.java:602)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunner.runJavaAgent(AgentRunner.java:370)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunner.run(AgentRunner.java:179)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunnerStarter.start(AgentRunnerStarter.java:111)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.agent.AgentRunnerStarter.main(AgentRunnerStarter.java:53)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.runtime.Main.main(Main.java:42)
[convhisthuman-pipeline-query-1-0] Caused by: org.apache.avro.SchemaParseException: Illegal character in: query-result
[convhisthuman-pipeline-query-1-0]      at org.apache.avro.Schema.validateName(Schema.java:1566)
[convhisthuman-pipeline-query-1-0]      at org.apache.avro.Schema.access$400(Schema.java:91)
[convhisthuman-pipeline-query-1-0]      at org.apache.avro.Schema$Field.<init>(Schema.java:546)
[convhisthuman-pipeline-query-1-0]      at org.apache.avro.Schema$Field.<init>(Schema.java:585)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.ai.agents.commons.TransformContext.setResultField(TransformContext.java:312)
[convhisthuman-pipeline-query-1-0]      at com.datastax.oss.streaming.ai.QueryStep.process(QueryStep.java:101)
[convhisthuman-pipeline-query-1-0]      at com.datastax.oss.streaming.ai.TransformStep.processAsync(TransformStep.java:41)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.ai.agents.GenAIToolKitAgent.processStep(GenAIToolKitAgent.java:139)
[convhisthuman-pipeline-query-1-0]      at ai.langstream.ai.agents.GenAIToolKitAgent.processRecord(GenAIToolKitAgent.java:117)
[convhisthuman-pipeline-query-1-0]      ... 16 common frames omitted
```


Discussion / hints:
We should add a validation at planner/deploy time.
Avro doesn't allow all the possible characters (for instance the '-' character.

Currently we are alwaysing trying to also maintain the Avro schema, maybe an alternative solution would be to not do it if the input object is not a AVRO object (we will also some CPU cyclesl

@cbornet 
I'm surprised that we use a Schema with Kafka's JSON 🤔 
This should only happen with Pulsar.

Fix patch:
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 a4b45a2e4..ce4d0b631 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
@@ -70,10 +70,12 @@ public class QueryStep implements TransformStep {
         this.dataSource = dataSource;
         this.loopOver = loopOver;
         this.loopOverAccessor = loopOverAccessor;
-        this.fields.forEach(
-                field -> {
-                    fieldsEvaluators.add(new JstlEvaluator<>("${" + field + "}", Object.class));
-                });
+        if (this.fields != null) {
+            this.fields.forEach(
+                    field -> {
+                        fieldsEvaluators.add(new JstlEvaluator<>("${" + field + "}", Object.class));
+                    });
+        }
         if (loopOver != null && !loopOver.isEmpty()) {
             this.loopOverAccessor = new JstlEvaluator<>("${" + loopOver + "}", List.class);
         } else {


Test patch:
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 c2d6b60fa..890bdfd38 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
@@ -552,4 +552,40 @@ public List<Map<String, Object>> fetchData(String query, List<Object> params) {
                 (List<Map<String, Object>>) result.get("command_results");
         assertEquals(List.of(Map.of("foo", "bar"), Map.of("foo", "bar2")), command_results);
     }
+
+    @Test
+    void testSetFieldWithDash() throws Exception {
+
+        String value = "{}";
+
+        QueryStepDataSource dataSource =
+                new QueryStepDataSource() {
+
+                    @Override
+                    public List<Map<String, Object>> fetchData(String query, List<Object> params) {
+                        return List.of(Map.of("foo", "bar"));
+                    }
+                };
+
+        QueryStep queryStep =
+                QueryStep.builder()
+                        .dataSource(dataSource)
+                        .outputFieldName("value.command-results")
+                        .query("select 1")
+                        .build();
+        queryStep.start();
+
+        MutableRecord context =
+                MutableRecord.recordToMutableRecord(SimpleRecord.of(null, value), true);
+
+        queryStep.process(context);
+        ai.langstream.api.runner.code.Record record =
+                MutableRecord.mutableRecordToRecord(context).orElseThrow();
+        Map<String, Object> result = (Map<String, Object>) record.value();
+        log.info("Result: {}", result);
+
+        List<Map<String, Object>> command_results =
+                (List<Map<String, Object>>) result.get("command-results");
+        assertEquals(List.of(Map.of("foo", "bar")), command_results);
+    }
 }