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() + } +}