From 6ec61bdb57976811fcc3a1faeb6d94b35c4cbf1c Mon Sep 17 00:00:00 2001 From: yuga-hashimoto Date: Wed, 22 Apr 2026 13:07:07 +0900 Subject: [PATCH] fix(streaming): aggregate OpenAI tool-call delta fragments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Priority 2 — Local agent capabilities OpenAI-compatible SSE streams emit tool calls as *fragments*: 1. {id:"call_1", name:"get_weather", arguments:""} 2. {id:"", name:"", arguments:"{\"lo"} 3. {id:"", name:"", arguments:"cation\":\"Tokyo\"}"} ChatViewModel.sendMessage and OpenClawProvider.send both appended each delta to a toolCalls list verbatim, producing three broken tool-call requests per call — one with the real id but empty arguments, two with empty ids and partial JSON. The dispatcher then ran the first with empty args (tool fell back to defaults or errored) and the later two as "Unknown tool". StreamingToolCallAggregator glues fragments back together: a delta with a non-empty id starts a new call (committing the pending one); an empty-id delta appends its name/arguments fragment to the pending call. Sequential tool streaming (the OpenAI default) works; true multi-tool interleaved streaming would need explicit index keys and is documented as a future extension. Wired into ChatViewModel streaming and OpenClawProvider.send (the non-streaming path that still consumes per-event WS messages). --- .../agent/StreamingToolCallAggregator.kt | 90 ++++++++++++++++++ .../provider/openclaw/OpenClawProvider.kt | 6 +- .../com/opendash/app/ui/chat/ChatViewModel.kt | 7 +- .../agent/StreamingToolCallAggregatorTest.kt | 93 +++++++++++++++++++ 4 files changed, 190 insertions(+), 6 deletions(-) create mode 100644 app/src/main/java/com/opendash/app/assistant/agent/StreamingToolCallAggregator.kt create mode 100644 app/src/test/java/com/opendash/app/assistant/agent/StreamingToolCallAggregatorTest.kt diff --git a/app/src/main/java/com/opendash/app/assistant/agent/StreamingToolCallAggregator.kt b/app/src/main/java/com/opendash/app/assistant/agent/StreamingToolCallAggregator.kt new file mode 100644 index 00000000..95d31662 --- /dev/null +++ b/app/src/main/java/com/opendash/app/assistant/agent/StreamingToolCallAggregator.kt @@ -0,0 +1,90 @@ +package com.opendash.app.assistant.agent + +import com.opendash.app.assistant.model.ToolCallRequest + +/** + * OpenAI-compatible SSE streams emit tool calls as *fragments*: + * + * 1. `{id:"call_1", name:"get_weather", arguments:""}` + * 2. `{id:"", name:"", arguments:"{\"lo"}` + * 3. `{id:"", name:"", arguments:"cation\":\"Tokyo\"}"}` + * + * Consumers that append each delta to a list verbatim end up with + * three broken `ToolCallRequest`s — one with the real id but empty + * arguments, two with empty ids and partial JSON. The downstream + * dispatcher then runs the first one (empty args → tool falls back + * to defaults or errors) and the later two as "Unknown tool". + * + * This accumulator glues the fragments back together. Heuristic: + * - A delta with a **non-empty id** that differs from the pending + * call's id starts a new call (and commits the pending one). + * - A delta with an **empty id** is a continuation: its `name` + * fragment is appended to the pending name, its `arguments` + * fragment is appended to the pending arguments. + * - A delta with an empty id and no content is a no-op. + * - A fragment that arrives before any call is pending becomes a + * standalone call rather than being silently dropped. + * + * This works for the common case of sequential tool streaming + * (one tool's deltas arrive fully before the next). It does NOT + * handle true multi-tool interleaved streaming — for that a + * provider needs to expose the explicit `index` field and we'd + * key the pending-map by index. OpenAI's Chat Completions API + * streams sequentially in practice, so this is sufficient for + * the current providers. + */ +class StreamingToolCallAggregator { + + private val completed = mutableListOf() + private var pendingId: String = "" + private var pendingName = StringBuilder() + private var pendingArgs = StringBuilder() + private var hasPending = false + + fun accept(delta: ToolCallRequest) { + val isNewCall = delta.id.isNotEmpty() && delta.id != pendingId + if (isNewCall) { + commitPending() + pendingId = delta.id + pendingName.clear().append(delta.name) + pendingArgs.clear().append(delta.arguments) + hasPending = true + return + } + // Continuation of the pending call (or the provider emitted an + // args-only fragment before any id-bearing header). + if (!hasPending) { + if (delta.id.isEmpty() && delta.name.isEmpty() && delta.arguments.isEmpty()) { + return + } + pendingId = delta.id + pendingName.clear().append(delta.name) + pendingArgs.clear().append(delta.arguments) + hasPending = true + return + } + // Sticky append. + if (delta.name.isNotEmpty()) pendingName.append(delta.name) + if (delta.arguments.isNotEmpty()) pendingArgs.append(delta.arguments) + } + + fun complete(): List { + commitPending() + return completed.toList() + } + + private fun commitPending() { + if (!hasPending) return + completed.add( + ToolCallRequest( + id = pendingId, + name = pendingName.toString(), + arguments = pendingArgs.toString() + ) + ) + pendingId = "" + pendingName.clear() + pendingArgs.clear() + hasPending = false + } +} diff --git a/app/src/main/java/com/opendash/app/assistant/provider/openclaw/OpenClawProvider.kt b/app/src/main/java/com/opendash/app/assistant/provider/openclaw/OpenClawProvider.kt index c9b1672a..ddd55054 100644 --- a/app/src/main/java/com/opendash/app/assistant/provider/openclaw/OpenClawProvider.kt +++ b/app/src/main/java/com/opendash/app/assistant/provider/openclaw/OpenClawProvider.kt @@ -54,12 +54,12 @@ class OpenClawProvider( ws.send(payload) val responseBuilder = StringBuilder() - val toolCalls = mutableListOf() + val aggregator = com.opendash.app.assistant.agent.StreamingToolCallAggregator() var finished = false ws.messages.collect { raw -> if (finished) return@collect val parsed = parseResponse(raw) - parsed.toolCallDelta?.let { toolCalls.add(it) } + parsed.toolCallDelta?.let { aggregator.accept(it) } if (parsed.finishReason != null) { finished = true return@collect @@ -68,7 +68,7 @@ class OpenClawProvider( } return AssistantMessage.Assistant( content = responseBuilder.toString(), - toolCalls = toolCalls + toolCalls = aggregator.complete() ) } diff --git a/app/src/main/java/com/opendash/app/ui/chat/ChatViewModel.kt b/app/src/main/java/com/opendash/app/ui/chat/ChatViewModel.kt index 1dfe509b..0eb10a30 100644 --- a/app/src/main/java/com/opendash/app/ui/chat/ChatViewModel.kt +++ b/app/src/main/java/com/opendash/app/ui/chat/ChatViewModel.kt @@ -3,10 +3,10 @@ package com.opendash.app.ui.chat import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope import com.opendash.app.assistant.agent.AgentToolDispatcher +import com.opendash.app.assistant.agent.StreamingToolCallAggregator import com.opendash.app.assistant.model.AssistantMessage import com.opendash.app.assistant.model.AssistantSession import com.opendash.app.assistant.model.ConversationState -import com.opendash.app.assistant.model.ToolCallRequest import com.opendash.app.assistant.router.ConversationRouter import com.opendash.app.tool.ToolExecutor import com.opendash.app.voice.pipeline.VoicePipeline @@ -95,14 +95,15 @@ class ChatViewModel @Inject constructor( while (toolRounds < MAX_TOOL_ROUNDS) { _streamingContent.value = "" val responseBuilder = StringBuilder() - val toolCalls = mutableListOf() + val toolCallAggregator = StreamingToolCallAggregator() provider.sendStreaming(session!!, conversationMessages, tools) .collect { delta -> responseBuilder.append(delta.contentDelta) _streamingContent.value = responseBuilder.toString() - delta.toolCallDelta?.let { toolCalls.add(it) } + delta.toolCallDelta?.let { toolCallAggregator.accept(it) } } + val toolCalls = toolCallAggregator.complete() val assistantResponse = AssistantMessage.Assistant( content = responseBuilder.toString(), diff --git a/app/src/test/java/com/opendash/app/assistant/agent/StreamingToolCallAggregatorTest.kt b/app/src/test/java/com/opendash/app/assistant/agent/StreamingToolCallAggregatorTest.kt new file mode 100644 index 00000000..54b8e0b8 --- /dev/null +++ b/app/src/test/java/com/opendash/app/assistant/agent/StreamingToolCallAggregatorTest.kt @@ -0,0 +1,93 @@ +package com.opendash.app.assistant.agent + +import com.google.common.truth.Truth.assertThat +import com.opendash.app.assistant.model.ToolCallRequest +import org.junit.jupiter.api.Test + +class StreamingToolCallAggregatorTest { + + @Test + fun `single complete delta is returned as-is`() { + val agg = StreamingToolCallAggregator() + agg.accept(ToolCallRequest(id = "call_1", name = "get_weather", arguments = "{\"loc\":\"Tokyo\"}")) + val out = agg.complete() + assertThat(out).hasSize(1) + assertThat(out[0].id).isEqualTo("call_1") + assertThat(out[0].name).isEqualTo("get_weather") + assertThat(out[0].arguments).isEqualTo("{\"loc\":\"Tokyo\"}") + } + + @Test + fun `arguments fragments with empty id append to the pending call`() { + val agg = StreamingToolCallAggregator() + // Canonical OpenAI streaming sequence: + // {id:"call_1", name:"get_weather", arguments:""} + // {id:"", name:"", arguments:"{\"loc\":"} + // {id:"", name:"", arguments:"\"Tokyo\"}"} + agg.accept(ToolCallRequest(id = "call_1", name = "get_weather", arguments = "")) + agg.accept(ToolCallRequest(id = "", name = "", arguments = "{\"loc\":")) + agg.accept(ToolCallRequest(id = "", name = "", arguments = "\"Tokyo\"}")) + + val out = agg.complete() + assertThat(out).hasSize(1) + assertThat(out[0].id).isEqualTo("call_1") + assertThat(out[0].name).isEqualTo("get_weather") + assertThat(out[0].arguments).isEqualTo("{\"loc\":\"Tokyo\"}") + } + + @Test + fun `new non-empty id starts a new call`() { + val agg = StreamingToolCallAggregator() + agg.accept(ToolCallRequest(id = "call_1", name = "get_weather", arguments = "{}")) + agg.accept(ToolCallRequest(id = "call_2", name = "get_time", arguments = "{}")) + + val out = agg.complete() + assertThat(out).hasSize(2) + assertThat(out.map { it.id }).containsExactly("call_1", "call_2").inOrder() + assertThat(out.map { it.name }).containsExactly("get_weather", "get_time").inOrder() + } + + @Test + fun `name fragment arriving after id continues the pending call`() { + val agg = StreamingToolCallAggregator() + // Some providers emit name piecemeal too; sticky behaviour: + // when id is empty and name is non-empty, we treat the name as + // an override-or-append signal for the pending call. + agg.accept(ToolCallRequest(id = "call_1", name = "get_", arguments = "")) + agg.accept(ToolCallRequest(id = "", name = "weather", arguments = "")) + agg.accept(ToolCallRequest(id = "", name = "", arguments = "{}")) + + val out = agg.complete() + assertThat(out).hasSize(1) + assertThat(out[0].name).isEqualTo("get_weather") + } + + @Test + fun `fragment before any pending call becomes a standalone call`() { + val agg = StreamingToolCallAggregator() + // Edge: provider emits an args-only fragment first. We treat it + // as a full call with empty id/name rather than silently dropping + // it — downstream will surface it as a tool failure, which is + // strictly better than losing the call. + agg.accept(ToolCallRequest(id = "", name = "", arguments = "{}")) + val out = agg.complete() + assertThat(out).hasSize(1) + assertThat(out[0].arguments).isEqualTo("{}") + } + + @Test + fun `accepting empty chunk is a no-op`() { + val agg = StreamingToolCallAggregator() + agg.accept(ToolCallRequest(id = "call_1", name = "get_weather", arguments = "{}")) + agg.accept(ToolCallRequest(id = "", name = "", arguments = "")) + val out = agg.complete() + assertThat(out).hasSize(1) + assertThat(out[0].arguments).isEqualTo("{}") + } + + @Test + fun `complete on empty aggregator returns empty list`() { + val agg = StreamingToolCallAggregator() + assertThat(agg.complete()).isEmpty() + } +}