Repository: LangStream/langstream (language: python)

Issue:
[BUG] document-to-json agent copy-properties option not copying correct value
Given this order of pipeline steps

```yaml
  - name: "Split into chunks"
    type: "text-splitter"
    input: html-render
    output: text-split
    configuration:
      splitter_type: "RecursiveCharacterTextSplitter"
      chunk_size: 10
      separators: ["\n\n", "\n", " ", ""]
      keep_separator: false
      chunk_overlap: 3
      length_function: "cl100k_base"

  - name: "Convert to json"
    type: "document-to-json"
    input: text-split
    output: json-text
    configuration:
      text-field: text
      copy-properties: true
```

The properties added by text-splitter are not getting decoded correctly by document-to-json.

Here are the headers added by splitter:
`{ "chunk_text_length": "10", "chunk_id": "4", "chunk_num_tokens": "4" }`

Here is the result after document-to-json:
```
headers: { "chunk_text_length": "21", "chunk_id": "1", "chunk_num_tokens": "4" }
body:
{
  "chunk_text_length": "MjE=",
  "text": "some test text.",
  "chunk_id": "MQ==",
  "chunk_num_tokens": "NA=="
}
```

Notice the numeric values in the body are encoded.


Discussion / hints:

Fix patch:
diff --git a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/DocumentToJsonAgent.java b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/DocumentToJsonAgent.java
index 11bda6b6e..206e8f5f4 100644
--- a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/DocumentToJsonAgent.java
+++ b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/DocumentToJsonAgent.java
@@ -19,6 +19,7 @@
 import ai.langstream.api.runner.code.SimpleRecord;
 import ai.langstream.api.runner.code.SingleRecordAgentProcessor;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -50,7 +51,18 @@ public List<Record> processRecord(Record record) throws Exception {
         Map<String, Object> asJson = new HashMap<>();
         asJson.put(textField, stream);
         if (copyProperties) {
-            record.headers().forEach(h -> asJson.put(h.key(), h.value()));
+            record.headers()
+                    .forEach(
+                            h -> {
+                                Object headerValue = h.value();
+                                // JSON doesn't deal well with byte arrays
+                                if (headerValue instanceof byte[]) {
+                                    headerValue =
+                                            new String(
+                                                    (byte[]) headerValue, StandardCharsets.UTF_8);
+                                }
+                                asJson.put(h.key(), headerValue);
+                            });
         }
 
         return List.of(
diff --git a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TikaTextExtractorAgent.java b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TikaTextExtractorAgent.java
index 77888a888..dc945f668 100644
--- a/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TikaTextExtractorAgent.java
+++ b/langstream-agents/langstream-agents-text-processing/src/main/java/ai/langstream/agents/text/TikaTextExtractorAgent.java
@@ -33,7 +33,6 @@
 
 @Slf4j
 public class TikaTextExtractorAgent extends SingleRecordAgentProcessor {
-
     @Override
     public List<Record> processRecord(Record record) throws Exception {
         if (record == null) {
@@ -44,15 +43,19 @@ public List<Record> processRecord(Record record) throws Exception {
         final InputStream stream = Utils.toStream(value);
         Metadata metadata = new Metadata();
         ParseContext parseContext = new ParseContext();
+        StringWriter valueAsString;
+
+        // ParsingReader starts new threads, so we need to make sure they are cleaned up
         try (Reader reader = new ParsingReader(parser, stream, metadata, parseContext)) {
-            StringWriter valueAsString = new StringWriter();
-            String[] names = metadata.names();
-            log.info(
-                    "Document type: {} Content {}",
-                    Stream.of(names).collect(Collectors.toMap(Function.identity(), metadata::get)),
-                    valueAsString);
+            valueAsString = new StringWriter();
             reader.transferTo(valueAsString);
-            return List.of(SimpleRecord.copyFrom(record).value(valueAsString.toString()).build());
         }
+
+        String[] names = metadata.names();
+        log.info(
+                "Document type: {} Content {}",
+                Stream.of(names).collect(Collectors.toMap(Function.identity(), metadata::get)),
+                valueAsString);
+        return List.of(SimpleRecord.copyFrom(record).value(valueAsString.toString()).build());
     }
 }
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/code/SimpleRecord.java b/langstream-api/src/main/java/ai/langstream/api/runner/code/SimpleRecord.java
index caae2a740..fafce5c40 100644
--- a/langstream-api/src/main/java/ai/langstream/api/runner/code/SimpleRecord.java
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/code/SimpleRecord.java
@@ -15,6 +15,7 @@
  */
 package ai.langstream.api.runner.code;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.Set;
 import lombok.AllArgsConstructor;
@@ -79,12 +80,12 @@ public static SimpleRecord of(Object key, Object value) {
     @AllArgsConstructor
     public static class SimpleHeader implements Header {
 
-        public static SimpleHeader of(String key, String value) {
+        public static SimpleHeader of(String key, Object value) {
             return new SimpleHeader(key, value);
         }
 
         final String key;
-        final String value;
+        final Object value;
 
         @Override
         public String key() {
@@ -92,13 +93,19 @@ public String key() {
         }
 
         @Override
-        public String value() {
+        public Object value() {
             return value;
         }
 
         @Override
         public String valueAsString() {
-            return value;
+            if (value == null) {
+                return null;
+            } else if (value instanceof byte[] b) {
+                return new String(b, StandardCharsets.UTF_8);
+            } else {
+                return value.toString();
+            }
         }
     }
 }
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 aeb15da7f..675ac7d9f 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
@@ -111,7 +111,8 @@ public static void runAgent(
             Path agentsDirectory,
             AgentInfo agentInfo,
             int maxLoops,
-            Runnable beforeStopSource)
+            Runnable beforeStopSource,
+            boolean startHttpServer)
             throws Exception {
         new AgentRunner()
                 .run(
@@ -121,7 +122,8 @@ public static void runAgent(
                         agentsDirectory,
                         agentInfo,
                         maxLoops,
-                        beforeStopSource);
+                        beforeStopSource,
+                        startHttpServer);
     }
 
     public void run(
@@ -131,7 +133,8 @@ public void run(
             Path agentsDirectory,
             AgentInfo agentInfo,
             int maxLoops,
-            Runnable beforeStopSource)
+            Runnable beforeStopSource,
+            boolean startHttpServer)
             throws Exception {
         log.info("Pod Configuration {}", configuration);
 
@@ -161,10 +164,10 @@ public void run(
                 Server server = null;
                 try {
                     if (PythonCodeAgentProvider.isPythonCodeAgent(agentCode.agentCode())) {
-                        server = bootstrapHttpServer(null);
+                        server = startHttpServer ? bootstrapHttpServer(null) : null;
                         runPythonAgent(podRuntimeConfiguration, codeDirectory);
                     } else {
-                        server = bootstrapHttpServer(agentInfo);
+                        server = startHttpServer ? bootstrapHttpServer(agentInfo) : null;
                         runJavaAgent(
                                 configuration,
                                 maxLoops,
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunnerStarter.java b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunnerStarter.java
index 30f2a2ff0..928e6fbac 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunnerStarter.java
+++ b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunnerStarter.java
@@ -105,6 +105,7 @@ public void start(String... args) throws Exception {
                 agentsDirectory,
                 new AgentInfo(),
                 -1,
-                null);
+                null,
+                true);
     }
 }


Test patch:
diff --git a/langstream-agents/langstream-agents-text-processing/src/test/java/ai/langstream/agents/text/DocumentToJsonTest.java b/langstream-agents/langstream-agents-text-processing/src/test/java/ai/langstream/agents/text/DocumentToJsonTest.java
index 0e0c828f0..5b9f20493 100644
--- a/langstream-agents/langstream-agents-text-processing/src/test/java/ai/langstream/agents/text/DocumentToJsonTest.java
+++ b/langstream-agents/langstream-agents-text-processing/src/test/java/ai/langstream/agents/text/DocumentToJsonTest.java
@@ -36,7 +36,23 @@ public void textConvertToJson() throws Exception {
         instance.init(Map.of("text-field", "document", "copy-properties", "true"));
 
         assertEquals(
-                "{\"detected-language\":\"en\",\"document\":\"This is a English\"}",
+                "{\"detected-language\":\"en\",\"document\":\"This is a English\",\"other-header\":\"bytearray-value\"}",
+                convertToJson(instance, "This is a English"));
+
+        instance.init(Map.of("text-field", "document", "copy-properties", "false"));
+        assertEquals(
+                "{\"document\":\"This is a English\"}",
+                convertToJson(instance, "This is a English"));
+    }
+
+    @Test
+    public void textConvertToJsonWithRawHeaders() throws Exception {
+        TextProcessingAgentsCodeProvider provider = new TextProcessingAgentsCodeProvider();
+        SingleRecordAgentProcessor instance = provider.createInstance("document-to-json");
+        instance.init(Map.of("text-field", "document", "copy-properties", "true"));
+
+        assertEquals(
+                "{\"detected-language\":\"en\",\"document\":\"This is a English\",\"other-header\":\"bytearray-value\"}",
                 convertToJson(instance, "This is a English"));
 
         instance.init(Map.of("text-field", "document", "copy-properties", "false"));
@@ -52,7 +68,13 @@ private static String convertToJson(SingleRecordAgentProcessor instance, String
                         .key("filename.txt")
                         .value(text.getBytes(StandardCharsets.UTF_8))
                         .origin("origin")
-                        .headers(List.of(new SimpleRecord.SimpleHeader("detected-language", "en")))
+                        .headers(
+                                List.of(
+                                        new SimpleRecord.SimpleHeader("detected-language", "en"),
+                                        new SimpleRecord.SimpleHeader(
+                                                "other-header",
+                                                "bytearray-value"
+                                                        .getBytes(StandardCharsets.UTF_8))))
                         .timestamp(System.currentTimeMillis())
                         .build();
 
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 7c25c5bee..20c6e3f6c 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
@@ -289,7 +289,8 @@ protected AgentRunResult executeAgentRunners(ApplicationRuntime runtime) throws
                                         agentsDirectory,
                                         agentInfo,
                                         10,
-                                        () -> validateAgentInfoBeforeStop(agentInfo));
+                                        () -> validateAgentInfoBeforeStop(agentInfo),
+                                        false);
                                 List<?> infos = agentInfo.serveWorkerStatus();
                                 log.info(
                                         "{} AgentPod {} AgentInfo {}",
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 a96dad510..79e3e1695 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
@@ -18,10 +18,13 @@
 import ai.langstream.AbstractApplicationRunner;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 @Slf4j
 class TextProcessingAgentsRunnerIT extends AbstractApplicationRunner {
@@ -88,4 +91,108 @@ public void testFullLanguageProcessingPipeline() throws Exception {
             }
         }
     }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testSplitThenJson(boolean useIntermediateTopic) throws Exception {
+        String tenant = "tenant";
+        String[] expectedAgents =
+                useIntermediateTopic
+                        ? new String[] {"app-step1", "app-step2"}
+                        : new String[] {"app-step1"};
+        String inputTopic = "input-topic-" + UUID.randomUUID();
+        String outputTopic = "output-topic-" + UUID.randomUUID();
+        String intermediateTopic = "intermediate-topic-" + UUID.randomUUID();
+
+        Map<String, String> application =
+                Map.of(
+                        "module.yaml",
+                        useIntermediateTopic
+                                ? """
+                                module: "module-1"
+                                id: "pipeline-1"
+                                topics:
+                                  - name: "%s"
+                                    creation-mode: create-if-not-exists
+                                  - name: "%s"
+                                    creation-mode: create-if-not-exists
+                                  - name: "%s"
+                                    creation-mode: create-if-not-exists
+                                pipeline:
+                                  - name: "Split into chunks"
+                                    id: step1
+                                    type: "text-splitter"
+                                    input: %s
+                                    output: %s
+                                    configuration:
+                                      chunk_size: 50
+                                      chunk_overlap: 0
+                                      keep_separator: true
+                                      length_function: "length"
+                                  - name: "Convert chunks to JSON"
+                                    type: "document-to-json"
+                                    id: step2
+                                    input: %s
+                                    output: %s
+                                    configuration:
+                                        text-field: text
+                                        copy-properties: true
+                                """
+                                        .formatted(
+                                                inputTopic,
+                                                outputTopic,
+                                                intermediateTopic,
+                                                inputTopic,
+                                                intermediateTopic,
+                                                intermediateTopic,
+                                                outputTopic)
+                                : """
+                                module: "module-1"
+                                id: "pipeline-1"
+                                topics:
+                                  - name: "%s"
+                                    creation-mode: create-if-not-exists
+                                  - name: "%s"
+                                    creation-mode: create-if-not-exists
+                                pipeline:
+                                  - name: "Split into chunks"
+                                    id: step1
+                                    type: "text-splitter"
+                                    input: %s
+                                    configuration:
+                                      chunk_size: 50
+                                      chunk_overlap: 0
+                                      keep_separator: true
+                                      length_function: "length"
+                                  - name: "Convert chunks to JSON"
+                                    type: "document-to-json"
+                                    output: %s
+                                    configuration:
+                                        text-field: text
+                                        copy-properties: true
+                                """
+                                        .formatted(
+                                                inputTopic, outputTopic, inputTopic, outputTopic));
+
+        try (ApplicationRuntime applicationRuntime =
+                deployApplication(
+                        tenant, "app", application, buildInstanceYaml(), expectedAgents)) {
+            try (KafkaProducer<String, String> producer = createProducer();
+                    KafkaConsumer<String, String> consumer = createConsumer(outputTopic)) {
+
+                sendMessage(
+                        inputTopic,
+                        "This text is written in English, but it is very long,\nso you may want to split it into chunks.",
+                        producer);
+
+                executeAgentRunners(applicationRuntime);
+                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\"}"));
+            }
+        }
+    }
 }
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/pulsar/PulsarRunnerDockerTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/pulsar/PulsarRunnerDockerTest.java
index 5a9af3f64..6937039c1 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/pulsar/PulsarRunnerDockerTest.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/pulsar/PulsarRunnerDockerTest.java
@@ -143,7 +143,8 @@ public void testRunAITools() throws Exception {
                     AbstractApplicationRunner.agentsDirectory,
                     new AgentInfo(),
                     5,
-                    null);
+                    null,
+                    false);
 
             // receive one message from the output-topic (written by the PodJavaRuntime)
             Message<byte[]> record = consumer.receive();
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
index 8853b47fe..60b28cf53 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
@@ -15,6 +15,8 @@
  */
 package ai.langstream.runtime.agent;
 
+import static org.mockito.ArgumentMatchers.eq;
+
 import ai.langstream.runtime.api.agent.AgentRunnerConstants;
 import ai.langstream.runtime.api.agent.RuntimePodConfiguration;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -87,7 +89,8 @@ private void runTest(boolean expectOk, Map<String, String> env, String... args)
                         Mockito.any(),
                         Mockito.any(),
                         Mockito.anyInt(),
-                        Mockito.any());
+                        Mockito.any(),
+                        eq(true));
     }
 
     static class TestDeployer extends AgentRunnerStarter {