combine json chunks from requests

[jiapingzeng]

Description

Combine request chunks in agent execute stream.

Related Issues

4314

Check List

• New functionality includes testing.
• New functionality has been documented.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff.
• Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[nathaliellenaa]

Thanks for the fix @jiapingzeng! Wondering if you see any additional latency due to combining the chunks?

[jiapingzeng]

I didn't notice additional latency from my side. There are some concerns of collectList() running out of memory when waiting for an infinite upstream: https://stackoverflow.com/a/72042404, but I don't think that applies here as agent execute requests are a few kBs at most.

[jiapingzeng]

Tests with failures:
 - org.opensearch.ml.rest.RestMLRAGSearchProcessorIT.testBM25WithBedrockWithConversation
131 tests completed, 1 failed, 11 skipped

=== Standard output of node `node{:opensearch-ml-plugin:integTest-0}` ===

»    ↓ errors and warnings from /__w/ml-commons/ml-commons/plugin/build/testclusters/integTest-0/logs/opensearch.stdout.log ↓
» WARN ][stderr                   ] [integTest-0] Oct 14, 2025 7:57:42 PM org.opensearch.javaagent.bootstrap.AgentPolicy setPolicy
» WARN ][stderr                   ] [integTest-0] INFO: Policy attached successfully: org.opensearch.bootstrap.OpenSearchPolicy@20de05e5
» WARN ][o.o.p.PluginsService     ] [integTest-0] Missing plugin [opensearch-security], dependency of [opensearch-ml]
» WARN ][o.o.p.PluginsService     ] [integTest-0] Some features of this plugin may not function without the dependencies being installed.
»  
» WARN ][stderr                   ] [integTest-0] Oct 14, 2025 7:57:42 PM org.apache.lucene.internal.vectorization.PanamaVectorizationProvider <init>
» WARN ][stderr                   ] [integTest-0] INFO: Java vector incubator API enabled; uses preferredBitSize=256; FMA enabled
» WARN ][o.o.g.DanglingIndicesState] [integTest-0] gateway.auto_import_dangling_indices is disabled, dangling indices will not be automatically detected or imported and must be managed manually
» WARN ][o.o.m.a.p.TransportPredictionStreamTaskAction] [integTest-0] StreamTransportService is not available.
» WARN ][o.o.m.a.e.TransportExecuteStreamTaskAction] [integTest-0] StreamTransportService is not available.
» WARN ][o.o.d.FileBasedSeedHostsProvider] [integTest-0] expected, but did not find, a dynamic hosts list at [/__w/ml-commons/ml-commons/plugin/build/testclusters/integTest-0/config/unicast_hosts.txt]
» WARN ][a.d.h.t.HuggingFaceTokenizer] [integTest-0] maxLength is not explicitly specified, use modelMaxLength: 512
» ERROR][o.o.m.a.MLModelAutoReDeployer] [integTest-0] Failed to query need undeploy models, no action will be performed
»  org.opensearch.index.IndexNotFoundException: no such index [.plugins-ml-model]
»  	at org.opensearch.cluster.metadata.IndexNameExpressionResolver$WildcardExpressionResolver.indexNotFoundException(IndexNameExpressionResolver.java:1101) ~[opensearch-3.3.0-SNAPSHOT.jar:3.3.0-SNAPSHOT]
»  	at org.opensearch.cluster.metadata.IndexNameExpressionResolver$WildcardExpressionResolver.innerResolve(IndexNameExpressionResolver.java:1038) ~[opensearch-3.3.0-SNAPSHOT.jar:3.3.0-SNAPSHOT]
»  	at org.opensearch.cluster.metadata.IndexNameExpressionResolver$WildcardExpressionResolver.resolve(IndexNameExpressionResolver.java:994) ~[opensearch-3.3.0-SNAPSHOT.jar:3.3.0-SNAPSHOT]

Don't think test failure is related to my change. Tried ./gradlew build locally and it passes.

[codecov]

Codecov Report

❌ Patch coverage is 19.23077% with 21 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.09%. Comparing base (f8b403b) to head (f5841e7).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
.../opensearch/ml/rest/RestMLExecuteStreamAction.java	19.23%	21 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #4317      +/-   ##
============================================
- Coverage     80.09%   80.09%   -0.01%     
- Complexity    10196    10200       +4     
============================================
  Files           855      855              
  Lines         44358    44373      +15     
  Branches       5133     5135       +2     
============================================
+ Hits          35530    35539       +9     
- Misses         6663     6670       +7     
+ Partials       2165     2164       -1     

Flag	Coverage Δ
ml-commons	80.09% <19.23%> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jiapingzeng]

Error handling sample requests:

curl -v -X POST "${OPENSEARCH_URL}/_plugins/_ml/agents/${AGENT_ID}/_execute/stream" \
  -H 'Content-Type: application/json' \
  -d '{"parameters":{"question":"test"'

data: {"error": "Failed to parse request: Unexpected end-of-input: expected close marker for Object (start marker at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 15])
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 33]"}

curl -v -X POST "${OPENSEARCH_URL}/_plugins/_ml/agents/${AGENT_ID}/_execute/stream" \
  -H 'Content-Type: application/json' \
  -d '{"parameters":::"question":"test"}}'

data: {"error": "Failed to parse request: Unexpected character (':' (code 58)): expected a valid value (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 15]"}

FIFO=$(mktemp -u)
mkfifo "$FIFO"

(
    echo -n '{"parameters":'
    sleep 0.1
    echo -n '{"question":'
    sleep 0.1
    echo -n '"test"'
    # Missing closing braces
) > "$FIFO" &

curl -v -X POST "${OPENSEARCH_URL}/_plugins/_ml/agents/${AGENT_ID}/_execute/stream" \
  -H 'Content-Type: application/json' \
  -H 'Transfer-Encoding: chunked' \
  --data-binary @"$FIFO"

data: {"error": "Failed to parse request: Unexpected end-of-input: expected close marker for Object (start marker at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 15])
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 33]"}

[mingshl]

would you address the code coverage?