diff --git a/internal/runtime/executor/devin_executor.go b/internal/runtime/executor/devin_executor.go index bcebb0284..e82ed4865 100644 --- a/internal/runtime/executor/devin_executor.go +++ b/internal/runtime/executor/devin_executor.go @@ -462,6 +462,9 @@ func (e *DevinExecutor) streamDevinFrames( name string } activeToolSlots := make(map[int]*devinActiveToolSlot) + activeCallByID := make(map[string]*devinActiveToolSlot) + var activeCallSlot *devinActiveToolSlot + toolCallCount := 0 thinkingBuf := &helps.UTF8SplitBuffer{} contentBuf := &helps.UTF8SplitBuffer{} var accumulatedThinking strings.Builder @@ -583,10 +586,6 @@ func (e *DevinExecutor) streamDevinFrames( } emitToolCall := func(tc helps.DevinToolCallDelta) bool { - if tc.Index < 0 || tc.Index >= maxDevinToolCalls { - log.Warnf("devin executor: tool call index %d out of bounds (max %d), dropping", tc.Index, maxDevinToolCalls) - return true - } if thoughtStarted { stopEvent, _ := sjson.SetBytes([]byte(`{"event_type":"step.stop","index":0}`), "index", stepIndex) _ = emitInteractionsEvent(stopEvent) @@ -600,16 +599,24 @@ func (e *DevinExecutor) streamDevinFrames( stepIndex++ } - slot, exists := activeToolSlots[tc.Index] - if exists && slot.id != "" && tc.ID != "" && tc.ID != slot.id { - stopEvent, _ := sjson.SetBytes([]byte(`{"event_type":"step.stop","index":0}`), "index", slot.stepIndex) - if !emitInteractionsEvent(stopEvent) { - return false - } - exists = false + argsChunk := tc.Arguments + if argsChunk == "" { + argsChunk = tc.InvalidJSONStr } - if !exists { + var slot *devinActiveToolSlot + if tc.ID != "" { + slot = activeCallByID[tc.ID] + } else if activeCallSlot != nil { + slot = activeCallSlot + } + + if slot == nil { + if toolCallCount >= maxDevinToolCalls { + log.Warnf("devin executor: total tool calls exceeded max %d, dropping", maxDevinToolCalls) + return true + } + toolCallCount++ sIdx := stepIndex stepIndex++ slot = &devinActiveToolSlot{ @@ -617,7 +624,11 @@ func (e *DevinExecutor) streamDevinFrames( id: tc.ID, name: tc.Name, } - activeToolSlots[tc.Index] = slot + activeToolSlots[sIdx] = slot + if tc.ID != "" { + activeCallByID[tc.ID] = slot + } + activeCallSlot = slot startEvent, _ := sjson.SetBytes([]byte(`{"event_type":"step.start","index":0,"step":{"type":"function_call","name":"","id":"","call_id":"","arguments":{}}}`), "index", sIdx) startEvent, _ = sjson.SetBytes(startEvent, "step.name", tc.Name) startEvent, _ = sjson.SetBytes(startEvent, "step.id", tc.ID) @@ -626,9 +637,11 @@ func (e *DevinExecutor) streamDevinFrames( return false } } else { + activeCallSlot = slot updated := false if slot.id == "" && tc.ID != "" { slot.id = tc.ID + activeCallByID[tc.ID] = slot updated = true } if slot.name == "" && tc.Name != "" { @@ -644,9 +657,9 @@ func (e *DevinExecutor) streamDevinFrames( } } - if tc.Arguments != "" { + if argsChunk != "" { deltaEvent, _ := sjson.SetBytes([]byte(`{"event_type":"step.delta","index":0,"delta":{"type":"arguments_delta","arguments":""}}`), "index", slot.stepIndex) - deltaEvent, _ = translatorcommon.SetStringWithoutHTMLEscape(deltaEvent, "delta.arguments", tc.Arguments) + deltaEvent, _ = translatorcommon.SetStringWithoutHTMLEscape(deltaEvent, "delta.arguments", argsChunk) if !emitInteractionsEvent(deltaEvent) { return false } @@ -674,6 +687,8 @@ func (e *DevinExecutor) streamDevinFrames( _ = emitInteractionsEvent(stopEvent) } clear(activeToolSlots) + clear(activeCallByID) + activeCallSlot = nil } } @@ -729,6 +744,9 @@ func (e *DevinExecutor) streamDevinFrames( if frameRes.Usage.CachedTokens > 0 { finalUsage.CachedTokens = frameRes.Usage.CachedTokens } + if frameRes.Usage.CacheWriteTokens > 0 { + finalUsage.CacheWriteTokens = frameRes.Usage.CacheWriteTokens + } if frameRes.Usage.RequestID != "" { finalUsage.RequestID = frameRes.Usage.RequestID } @@ -915,6 +933,9 @@ func (e *DevinExecutor) streamDevinFrames( completedEvent, _ = sjson.SetBytes(completedEvent, "interaction.usage.total_input_tokens", totalInput) completedEvent, _ = sjson.SetBytes(completedEvent, "interaction.usage.total_output_tokens", totalOutput) completedEvent, _ = sjson.SetBytes(completedEvent, "interaction.usage.total_cached_tokens", finalUsage.CachedTokens) + if finalUsage.CacheWriteTokens > 0 { + completedEvent, _ = sjson.SetBytes(completedEvent, "interaction.usage.cache_write_tokens", finalUsage.CacheWriteTokens) + } completedEvent, _ = sjson.SetBytes(completedEvent, "interaction.usage.total_tokens", totalTokens) if detail, ok := helps.ParseInteractionsStreamUsage(completedEvent); ok { if reporter != nil { @@ -976,7 +997,8 @@ func consumeDevinFramesToInteractions(body io.Reader, model, chatModelUID string args strings.Builder } var toolBuilders []*devinToolCallBuilder - slotToBuilderIndex := make(map[int]int) + callIDToBuilderIndex := make(map[string]int) + lastBuilderIdx := -1 getToolCalls := func() []helps.DevinToolCall { if len(toolBuilders) == 0 { @@ -1078,6 +1100,9 @@ func consumeDevinFramesToInteractions(body io.Reader, model, chatModelUID string if frameRes.Usage.CachedTokens > 0 { finalUsage.CachedTokens = frameRes.Usage.CachedTokens } + if frameRes.Usage.CacheWriteTokens > 0 { + finalUsage.CacheWriteTokens = frameRes.Usage.CacheWriteTokens + } if frameRes.Usage.RequestID != "" { finalUsage.RequestID = frameRes.Usage.RequestID } @@ -1123,32 +1148,47 @@ func consumeDevinFramesToInteractions(body io.Reader, model, chatModelUID string textParts = append(textParts, frameRes.ContentText) } for _, tc := range frameRes.ToolCallDeltas { - slotIdx := tc.Index - if slotIdx < 0 || slotIdx >= maxDevinToolCalls { - log.Warnf("devin executor: tool call index %d out of bounds (max %d), dropping", slotIdx, maxDevinToolCalls) - continue + argsChunk := tc.Arguments + if argsChunk == "" { + argsChunk = tc.InvalidJSONStr } - bIdx, exists := slotToBuilderIndex[slotIdx] - if exists && tc.ID != "" && toolBuilders[bIdx].id != "" && tc.ID != toolBuilders[bIdx].id { - exists = false + + var bIdx int + var exists bool + if tc.ID != "" { + bIdx, exists = callIDToBuilderIndex[tc.ID] + } else if lastBuilderIdx >= 0 { + bIdx = lastBuilderIdx + exists = true } + if !exists { if len(toolBuilders) >= maxDevinToolCalls { log.Warnf("devin executor: total tool calls exceeded max %d, dropping", maxDevinToolCalls) continue } bIdx = len(toolBuilders) - toolBuilders = append(toolBuilders, &devinToolCallBuilder{}) - slotToBuilderIndex[slotIdx] = bIdx + toolBuilders = append(toolBuilders, &devinToolCallBuilder{ + id: tc.ID, + name: tc.Name, + }) + if tc.ID != "" { + callIDToBuilderIndex[tc.ID] = bIdx + } + lastBuilderIdx = bIdx + } else { + lastBuilderIdx = bIdx + if toolBuilders[bIdx].id == "" && tc.ID != "" { + toolBuilders[bIdx].id = tc.ID + callIDToBuilderIndex[tc.ID] = bIdx + } + if tc.Name != "" { + toolBuilders[bIdx].name = tc.Name + } } - if tc.ID != "" { - toolBuilders[bIdx].id = tc.ID - } - if tc.Name != "" { - toolBuilders[bIdx].name = tc.Name - } - if tc.Arguments != "" { - toolBuilders[bIdx].args.WriteString(tc.Arguments) + + if argsChunk != "" { + toolBuilders[bIdx].args.WriteString(argsChunk) } } } @@ -1221,8 +1261,12 @@ func consumeDevinFramesToInteractions(body io.Reader, model, chatModelUID string fnStep, _ = sjson.SetBytes(fnStep, "name", tc.Name) fnStep, _ = sjson.SetBytes(fnStep, "id", tc.ID) fnStep, _ = sjson.SetBytes(fnStep, "call_id", tc.ID) - if tc.Arguments != "" && json.Valid([]byte(tc.Arguments)) { - fnStep, _ = sjson.SetRawBytes(fnStep, "arguments", []byte(tc.Arguments)) + if tc.Arguments != "" { + if json.Valid([]byte(tc.Arguments)) { + fnStep, _ = sjson.SetRawBytes(fnStep, "arguments", []byte(tc.Arguments)) + } else { + fnStep, _ = translatorcommon.SetStringWithoutHTMLEscape(fnStep, "arguments", tc.Arguments) + } } steps = append(steps, fnStep) } @@ -1248,6 +1292,9 @@ func consumeDevinFramesToInteractions(body io.Reader, model, chatModelUID string out, _ = sjson.SetBytes(out, "usage.total_input_tokens", totalInput) out, _ = sjson.SetBytes(out, "usage.total_output_tokens", totalOutput) out, _ = sjson.SetBytes(out, "usage.total_cached_tokens", finalUsage.CachedTokens) + if finalUsage.CacheWriteTokens > 0 { + out, _ = sjson.SetBytes(out, "usage.cache_write_tokens", finalUsage.CacheWriteTokens) + } out, _ = sjson.SetBytes(out, "usage.total_tokens", totalTokens) } diff --git a/internal/runtime/executor/devin_executor_test.go b/internal/runtime/executor/devin_executor_test.go index a9400a164..020ec4147 100644 --- a/internal/runtime/executor/devin_executor_test.go +++ b/internal/runtime/executor/devin_executor_test.go @@ -780,8 +780,6 @@ func TestConsumeDevinFramesToInteractions_MultiToolCallsNoPanic(t *testing.T) { tc0 = protowire.AppendString(tc0, "tool_0") tc0 = protowire.AppendTag(tc0, 3, protowire.BytesType) tc0 = protowire.AppendString(tc0, `{"a":`) - tc0 = protowire.AppendTag(tc0, 4, protowire.VarintType) - tc0 = protowire.AppendVarint(tc0, 0) // index 0 var f1 []byte f1 = protowire.AppendTag(f1, 6, protowire.BytesType) @@ -796,8 +794,6 @@ func TestConsumeDevinFramesToInteractions_MultiToolCallsNoPanic(t *testing.T) { tc1 = protowire.AppendString(tc1, "tool_1") tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) tc1 = protowire.AppendString(tc1, `{"b": 2}`) - tc1 = protowire.AppendTag(tc1, 4, protowire.VarintType) - tc1 = protowire.AppendVarint(tc1, 1) // index 1 var f2 []byte f2 = protowire.AppendTag(f2, 6, protowire.BytesType) @@ -806,10 +802,10 @@ func TestConsumeDevinFramesToInteractions_MultiToolCallsNoPanic(t *testing.T) { // Frame 3: tool call 0 continuation var tc0Cont []byte + tc0Cont = protowire.AppendTag(tc0Cont, 1, protowire.BytesType) + tc0Cont = protowire.AppendString(tc0Cont, "call_0") tc0Cont = protowire.AppendTag(tc0Cont, 3, protowire.BytesType) tc0Cont = protowire.AppendString(tc0Cont, `1}`) - tc0Cont = protowire.AppendTag(tc0Cont, 4, protowire.VarintType) - tc0Cont = protowire.AppendVarint(tc0Cont, 0) // index 0 var f3 []byte f3 = protowire.AppendTag(f3, 6, protowire.BytesType) @@ -954,14 +950,12 @@ func TestStreamDevinFrames_SequentialToolCallsSameIndexDifferentID(t *testing.T) tc0 = protowire.AppendString(tc0, "title") tc0 = protowire.AppendTag(tc0, 3, protowire.BytesType) tc0 = protowire.AppendString(tc0, `{"title": "Triage issue 5802"}`) - tc0 = protowire.AppendTag(tc0, 4, protowire.VarintType) - tc0 = protowire.AppendVarint(tc0, 0) // index 0 var f1 []byte f1 = protowire.AppendTag(f1, 6, protowire.BytesType) f1 = protowire.AppendBytes(f1, tc0) - // Frame 2: bash_1 (same index 0, but different ID) + // Frame 2: bash_1 var tc1 []byte tc1 = protowire.AppendTag(tc1, 1, protowire.BytesType) tc1 = protowire.AppendString(tc1, "bash_1") @@ -969,8 +963,6 @@ func TestStreamDevinFrames_SequentialToolCallsSameIndexDifferentID(t *testing.T) tc1 = protowire.AppendString(tc1, "bash") tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) tc1 = protowire.AppendString(tc1, `{"command": "gh issue view 5802 2>&1 | head -100"}`) - tc1 = protowire.AppendTag(tc1, 4, protowire.VarintType) - tc1 = protowire.AppendVarint(tc1, 0) // index 0 var f2 []byte f2 = protowire.AppendTag(f2, 6, protowire.BytesType) @@ -1072,8 +1064,6 @@ func TestConsumeDevinFramesToInteractions_SequentialToolCallsSameIndexDifferentI tc0 = protowire.AppendString(tc0, "title") tc0 = protowire.AppendTag(tc0, 3, protowire.BytesType) tc0 = protowire.AppendString(tc0, `{"title": "Triage issue 5802"}`) - tc0 = protowire.AppendTag(tc0, 4, protowire.VarintType) - tc0 = protowire.AppendVarint(tc0, 0) var f1 []byte f1 = protowire.AppendTag(f1, 6, protowire.BytesType) @@ -1086,8 +1076,6 @@ func TestConsumeDevinFramesToInteractions_SequentialToolCallsSameIndexDifferentI tc1 = protowire.AppendString(tc1, "bash") tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) tc1 = protowire.AppendString(tc1, `{"command": "gh issue view 5802 2>&1 | head -100"}`) - tc1 = protowire.AppendTag(tc1, 4, protowire.VarintType) - tc1 = protowire.AppendVarint(tc1, 0) var f2 []byte f2 = protowire.AppendTag(f2, 6, protowire.BytesType) @@ -1129,7 +1117,7 @@ func TestConsumeDevinFramesToInteractions_SequentialToolCallsSameIndexDifferentI func TestConsumeDevinFramesToInteractions_ToolCallsLimit128(t *testing.T) { var buf bytes.Buffer - // Create 135 tool calls across sequential ID switches on index 0 + // Create 135 tool calls across sequential ID switches for i := 0; i < 135; i++ { var tc []byte tc = protowire.AppendTag(tc, 1, protowire.BytesType) @@ -1138,8 +1126,6 @@ func TestConsumeDevinFramesToInteractions_ToolCallsLimit128(t *testing.T) { tc = protowire.AppendString(tc, fmt.Sprintf("tool_%d", i)) tc = protowire.AppendTag(tc, 3, protowire.BytesType) tc = protowire.AppendString(tc, `{"param":1}`) - tc = protowire.AppendTag(tc, 4, protowire.VarintType) - tc = protowire.AppendVarint(tc, 0) var f []byte f = protowire.AppendTag(f, 6, protowire.BytesType) @@ -1166,8 +1152,6 @@ func TestStreamDevinFrames_SameIDDoesNotDuplicateStart(t *testing.T) { tc0 = protowire.AppendString(tc0, "tool_1") tc0 = protowire.AppendTag(tc0, 3, protowire.BytesType) tc0 = protowire.AppendString(tc0, `{"a":`) - tc0 = protowire.AppendTag(tc0, 4, protowire.VarintType) - tc0 = protowire.AppendVarint(tc0, 0) var f1 []byte f1 = protowire.AppendTag(f1, 6, protowire.BytesType) @@ -1181,8 +1165,6 @@ func TestStreamDevinFrames_SameIDDoesNotDuplicateStart(t *testing.T) { tc1 = protowire.AppendString(tc1, "tool_1") tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) tc1 = protowire.AppendString(tc1, `1}`) - tc1 = protowire.AppendTag(tc1, 4, protowire.VarintType) - tc1 = protowire.AppendVarint(tc1, 0) var f2 []byte f2 = protowire.AppendTag(f2, 6, protowire.BytesType) @@ -2836,3 +2818,402 @@ func TestDevinExecutor_SupplementImagesEdgeCases(t *testing.T) { } }) } + +func TestRegressionIssue5910_ToolCallAggregationByCallID(t *testing.T) { + // Frame 1: call_1 start + partial args + var tc1 []byte + tc1 = protowire.AppendTag(tc1, 1, protowire.BytesType) + tc1 = protowire.AppendString(tc1, "call_1") + tc1 = protowire.AppendTag(tc1, 2, protowire.BytesType) + tc1 = protowire.AppendString(tc1, "tool_1") + tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) + tc1 = protowire.AppendString(tc1, `{"a":`) + + var f1 []byte + f1 = protowire.AppendTag(f1, 6, protowire.BytesType) + f1 = protowire.AppendBytes(f1, tc1) + + // Frame 2: call_2 start + args + var tc2 []byte + tc2 = protowire.AppendTag(tc2, 1, protowire.BytesType) + tc2 = protowire.AppendString(tc2, "call_2") + tc2 = protowire.AppendTag(tc2, 2, protowire.BytesType) + tc2 = protowire.AppendString(tc2, "tool_2") + tc2 = protowire.AppendTag(tc2, 3, protowire.BytesType) + tc2 = protowire.AppendString(tc2, `{"b":2}`) + + var f2 []byte + f2 = protowire.AppendTag(f2, 6, protowire.BytesType) + f2 = protowire.AppendBytes(f2, tc2) + + // Frame 3: call_1 continuation args + var tc1Cont []byte + tc1Cont = protowire.AppendTag(tc1Cont, 1, protowire.BytesType) + tc1Cont = protowire.AppendString(tc1Cont, "call_1") + tc1Cont = protowire.AppendTag(tc1Cont, 3, protowire.BytesType) + tc1Cont = protowire.AppendString(tc1Cont, `1}`) + + var f3 []byte + f3 = protowire.AppendTag(f3, 6, protowire.BytesType) + f3 = protowire.AppendBytes(f3, tc1Cont) + + var buf bytes.Buffer + buf.Write(helps.WrapConnectEnvelope(f1)) + buf.Write(helps.WrapConnectEnvelope(f2)) + buf.Write(helps.WrapConnectEnvelope(f3)) + buf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + _, respLog, err := consumeDevinFramesToInteractions(&buf, "devin/swe-2", "chat-model-uid") + if err != nil { + t.Fatalf("consumeDevinFramesToInteractions failed: %v", err) + } + + if len(respLog.ToolCalls) != 2 { + t.Fatalf("expected 2 tool calls, got %d", len(respLog.ToolCalls)) + } + if respLog.ToolCalls[0].ID != "call_1" || respLog.ToolCalls[0].Arguments != `{"a":1}` { + t.Errorf("tool call 0 = %+v, want call_1 with args {\"a\":1}", respLog.ToolCalls[0]) + } + if respLog.ToolCalls[1].ID != "call_2" || respLog.ToolCalls[1].Arguments != `{"b":2}` { + t.Errorf("tool call 1 = %+v, want call_2 with args {\"b\":2}", respLog.ToolCalls[1]) + } +} + +func TestRegressionIssue5910_CustomToolCallInvalidJSONStr(t *testing.T) { + // Upstream Devin sends raw arguments for custom tool calls in field 4 (invalid_json_str), + // parse error in field 5 (invalid_json_err), and custom flag in field 6 (is_custom_tool_call). + var tc []byte + tc = protowire.AppendTag(tc, 1, protowire.BytesType) + tc = protowire.AppendString(tc, "call_custom_1") + tc = protowire.AppendTag(tc, 2, protowire.BytesType) + tc = protowire.AppendString(tc, "bash") + // Field 4: invalid_json_str = "ls -la" + tc = protowire.AppendTag(tc, 4, protowire.BytesType) + tc = protowire.AppendString(tc, "ls -la") + // Field 5: invalid_json_err = "not valid json" + tc = protowire.AppendTag(tc, 5, protowire.BytesType) + tc = protowire.AppendString(tc, "not valid json") + // Field 6: is_custom_tool_call = true + tc = protowire.AppendTag(tc, 6, protowire.VarintType) + tc = protowire.AppendVarint(tc, 1) + + var f1 []byte + f1 = protowire.AppendTag(f1, 6, protowire.BytesType) + f1 = protowire.AppendBytes(f1, tc) + + var buf bytes.Buffer + buf.Write(helps.WrapConnectEnvelope(f1)) + buf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + interactionsJSON, respLog, err := consumeDevinFramesToInteractions(&buf, "devin/swe-2", "chat-model-uid") + if err != nil { + t.Fatalf("consumeDevinFramesToInteractions failed: %v", err) + } + + if len(respLog.ToolCalls) != 1 { + t.Fatalf("expected 1 tool call, got %d", len(respLog.ToolCalls)) + } + if respLog.ToolCalls[0].Arguments != "ls -la" { + t.Errorf("expected Arguments %q from invalid_json_str, got %q", "ls -la", respLog.ToolCalls[0].Arguments) + } + + steps := gjson.GetBytes(interactionsJSON, "steps").Array() + if len(steps) != 1 { + t.Fatalf("expected 1 step in interactions JSON, got %d", len(steps)) + } + if steps[0].Get("name").String() != "bash" || steps[0].Get("id").String() != "call_custom_1" { + t.Errorf("unexpected step 0: %s", steps[0].Raw) + } + if steps[0].Get("arguments").String() != "ls -la" { + t.Errorf("expected step 0 arguments %q, got %q", "ls -la", steps[0].Get("arguments").String()) + } + + // Also verify streaming receives raw arguments for custom tool call + var streamBuf bytes.Buffer + streamBuf.Write(helps.WrapConnectEnvelope(f1)) + streamBuf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + exec := NewDevinExecutor(&config.Config{}) + out := make(chan cliproxyexecutor.StreamChunk, 50) + opts := cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatInteractions, + } + + go func() { + defer close(out) + exec.streamDevinFrames( + context.Background(), + &streamBuf, + cliproxyexecutor.Request{Model: "devin/swe-2"}, + opts, + "chat-model-uid", + sdktranslator.FormatInteractions, + nil, + out, + ) + }() + + var streamArgs strings.Builder + for chunk := range out { + if chunk.Err != nil { + t.Fatalf("unexpected stream error: %v", chunk.Err) + } + lines := strings.Split(string(chunk.Payload), "\n") + for _, line := range lines { + if strings.HasPrefix(line, "data: ") { + data := strings.TrimPrefix(line, "data: ") + if gjson.Get(data, "event_type").String() == "step.delta" && gjson.Get(data, "delta.type").String() == "arguments_delta" { + streamArgs.WriteString(gjson.Get(data, "delta.arguments").String()) + } + } + } + } + + if streamArgs.String() != "ls -la" { + t.Errorf("expected stream delta arguments %q, got %q", "ls -la", streamArgs.String()) + } +} + +func TestRegressionIssue5910_StreamInterleavedToolCallsByCallID(t *testing.T) { + // Frame 1: call_1 start + partial args + var tc1 []byte + tc1 = protowire.AppendTag(tc1, 1, protowire.BytesType) + tc1 = protowire.AppendString(tc1, "call_1") + tc1 = protowire.AppendTag(tc1, 2, protowire.BytesType) + tc1 = protowire.AppendString(tc1, "tool_1") + tc1 = protowire.AppendTag(tc1, 3, protowire.BytesType) + tc1 = protowire.AppendString(tc1, `{"a":`) + + var f1 []byte + f1 = protowire.AppendTag(f1, 6, protowire.BytesType) + f1 = protowire.AppendBytes(f1, tc1) + + // Frame 2: call_2 start + full args + var tc2 []byte + tc2 = protowire.AppendTag(tc2, 1, protowire.BytesType) + tc2 = protowire.AppendString(tc2, "call_2") + tc2 = protowire.AppendTag(tc2, 2, protowire.BytesType) + tc2 = protowire.AppendString(tc2, "tool_2") + tc2 = protowire.AppendTag(tc2, 3, protowire.BytesType) + tc2 = protowire.AppendString(tc2, `{"b":2}`) + + var f2 []byte + f2 = protowire.AppendTag(f2, 6, protowire.BytesType) + f2 = protowire.AppendBytes(f2, tc2) + + // Frame 3: call_1 continuation args + var tc1Cont []byte + tc1Cont = protowire.AppendTag(tc1Cont, 1, protowire.BytesType) + tc1Cont = protowire.AppendString(tc1Cont, "call_1") + tc1Cont = protowire.AppendTag(tc1Cont, 3, protowire.BytesType) + tc1Cont = protowire.AppendString(tc1Cont, `1}`) + + var f3 []byte + f3 = protowire.AppendTag(f3, 6, protowire.BytesType) + f3 = protowire.AppendBytes(f3, tc1Cont) + + var buf bytes.Buffer + buf.Write(helps.WrapConnectEnvelope(f1)) + buf.Write(helps.WrapConnectEnvelope(f2)) + buf.Write(helps.WrapConnectEnvelope(f3)) + buf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + exec := NewDevinExecutor(&config.Config{}) + out := make(chan cliproxyexecutor.StreamChunk, 50) + opts := cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatInteractions, + } + + go func() { + defer close(out) + exec.streamDevinFrames( + context.Background(), + &buf, + cliproxyexecutor.Request{Model: "devin/swe-2"}, + opts, + "chat-model-uid", + sdktranslator.FormatInteractions, + nil, + out, + ) + }() + + var chunks []cliproxyexecutor.StreamChunk + for chunk := range out { + if chunk.Err != nil { + t.Fatalf("unexpected chunk error: %v", chunk.Err) + } + chunks = append(chunks, chunk) + } + + var events []gjson.Result + for _, chunk := range chunks { + lines := strings.Split(string(chunk.Payload), "\n") + for _, line := range lines { + if strings.HasPrefix(line, "data: ") { + data := strings.TrimPrefix(line, "data: ") + if strings.TrimSpace(data) != "[DONE]" { + events = append(events, gjson.Parse(data)) + } + } + } + } + + var toolCallsStarted []string + var toolCallsStopped []int64 + stepArgs := make(map[int64]*strings.Builder) + for _, ev := range events { + eventType := ev.Get("event_type").String() + if eventType == "step.start" && ev.Get("step.type").String() == "function_call" { + idx := ev.Get("index").Int() + toolCallsStarted = append(toolCallsStarted, fmt.Sprintf("index:%d,id:%s,name:%s", idx, ev.Get("step.id").String(), ev.Get("step.name").String())) + stepArgs[idx] = &strings.Builder{} + } else if eventType == "step.delta" && ev.Get("delta.type").String() == "arguments_delta" { + idx := ev.Get("index").Int() + if b, ok := stepArgs[idx]; ok { + b.WriteString(ev.Get("delta.arguments").String()) + } + } else if eventType == "step.stop" { + toolCallsStopped = append(toolCallsStopped, ev.Get("index").Int()) + } + } + + if len(toolCallsStarted) != 2 { + t.Fatalf("expected exactly 2 tool calls started, got %d: %v", len(toolCallsStarted), toolCallsStarted) + } + if toolCallsStarted[0] != "index:0,id:call_1,name:tool_1" { + t.Errorf("tool call 0 = %q, want index:0,id:call_1,name:tool_1", toolCallsStarted[0]) + } + if toolCallsStarted[1] != "index:1,id:call_2,name:tool_2" { + t.Errorf("tool call 1 = %q, want index:1,id:call_2,name:tool_2", toolCallsStarted[1]) + } + + if len(toolCallsStopped) != 2 { + t.Fatalf("expected exactly 2 tool calls stopped, got %d: %v", len(toolCallsStopped), toolCallsStopped) + } + if toolCallsStopped[0] != 0 || toolCallsStopped[1] != 1 { + t.Errorf("tool calls stopped indices = %v, want [0, 1]", toolCallsStopped) + } + + if stepArgs[0].String() != `{"a":1}` { + t.Errorf("stepArgs[0] = %q, want {\"a\":1}", stepArgs[0].String()) + } + if stepArgs[1].String() != `{"b":2}` { + t.Errorf("stepArgs[1] = %q, want {\"b\":2}", stepArgs[1].String()) + } +} + +func TestRegressionIssue5910_UsageStatsCacheWriteTokensInResponses(t *testing.T) { + // Frame 1 with Usage Field 7 + var f7Bytes []byte + // Field 2: prompt tokens = 3 + f7Bytes = protowire.AppendTag(f7Bytes, 2, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 3) + // Field 3: completion tokens = 10 + f7Bytes = protowire.AppendTag(f7Bytes, 3, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 10) + // Field 4: cache_write_tokens = 14361 + f7Bytes = protowire.AppendTag(f7Bytes, 4, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 14361) + // Field 5: cached tokens = 50 + f7Bytes = protowire.AppendTag(f7Bytes, 5, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 50) + + var frame []byte + frame = protowire.AppendTag(frame, 7, protowire.BytesType) + frame = protowire.AppendBytes(frame, f7Bytes) + + var buf bytes.Buffer + buf.Write(helps.WrapConnectEnvelope(frame)) + buf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + // 1. Non-streaming test + interactionsJSON, respLog, err := consumeDevinFramesToInteractions(&buf, "devin/swe-2", "chat-model-uid") + if err != nil { + t.Fatalf("consumeDevinFramesToInteractions failed: %v", err) + } + + if respLog.Usage == nil { + t.Fatal("expected non-nil respLog.Usage") + } + if respLog.Usage.PromptTokens != 3 { + t.Errorf("respLog.Usage.PromptTokens = %d, want 3", respLog.Usage.PromptTokens) + } + if respLog.Usage.CacheWriteTokens != 14361 { + t.Errorf("respLog.Usage.CacheWriteTokens = %d, want 14361", respLog.Usage.CacheWriteTokens) + } + + root := gjson.ParseBytes(interactionsJSON) + if root.Get("usage.total_input_tokens").Int() != 53 { + t.Errorf("usage.total_input_tokens = %d, want 53", root.Get("usage.total_input_tokens").Int()) + } + if root.Get("usage.cache_write_tokens").Int() != 14361 { + t.Errorf("usage.cache_write_tokens = %d, want 14361", root.Get("usage.cache_write_tokens").Int()) + } + detail := helps.ParseInteractionsUsage(interactionsJSON) + if detail.CacheCreationTokens != 14361 { + t.Errorf("ParseInteractionsUsage CacheCreationTokens = %d, want 14361", detail.CacheCreationTokens) + } + if detail.InputTokens != 53 { + t.Errorf("ParseInteractionsUsage InputTokens = %d, want 53", detail.InputTokens) + } + + // 2. Streaming test + var streamBuf bytes.Buffer + streamBuf.Write(helps.WrapConnectEnvelope(frame)) + streamBuf.Write(helps.WrapConnectEnvelopeWithFlag(helps.ConnectFlagEndStream, []byte(`{}`))) + + exec := NewDevinExecutor(&config.Config{}) + out := make(chan cliproxyexecutor.StreamChunk, 50) + opts := cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatInteractions, + } + + go func() { + defer close(out) + exec.streamDevinFrames( + context.Background(), + &streamBuf, + cliproxyexecutor.Request{Model: "devin/swe-2"}, + opts, + "chat-model-uid", + sdktranslator.FormatInteractions, + nil, + out, + ) + }() + + var completedEvent []byte + for chunk := range out { + if chunk.Err != nil { + t.Fatalf("unexpected chunk error: %v", chunk.Err) + } + lines := strings.Split(string(chunk.Payload), "\n") + for _, line := range lines { + if strings.HasPrefix(line, "data: ") { + data := []byte(strings.TrimPrefix(line, "data: ")) + if gjson.GetBytes(data, "event_type").String() == "interaction.completed" { + completedEvent = data + } + } + } + } + + if len(completedEvent) == 0 { + t.Fatal("expected interaction.completed event in stream") + } + cRoot := gjson.ParseBytes(completedEvent) + if cRoot.Get("interaction.usage.cache_write_tokens").Int() != 14361 { + t.Errorf("interaction.usage.cache_write_tokens = %d, want 14361", cRoot.Get("interaction.usage.cache_write_tokens").Int()) + } + sDetail, ok := helps.ParseInteractionsStreamUsage(completedEvent) + if !ok { + t.Fatal("ParseInteractionsStreamUsage returned false") + } + if sDetail.CacheCreationTokens != 14361 { + t.Errorf("ParseInteractionsStreamUsage CacheCreationTokens = %d, want 14361", sDetail.CacheCreationTokens) + } + if sDetail.InputTokens != 53 { + t.Errorf("ParseInteractionsStreamUsage InputTokens = %d, want 53", sDetail.InputTokens) + } +} diff --git a/internal/runtime/executor/helps/devin_wire.go b/internal/runtime/executor/helps/devin_wire.go index fe6b4dbac..3a6356fb5 100644 --- a/internal/runtime/executor/helps/devin_wire.go +++ b/internal/runtime/executor/helps/devin_wire.go @@ -68,10 +68,12 @@ type DevinToolCall struct { // DevinToolCallDelta represents a streaming tool call chunk from response Field 6. type DevinToolCallDelta struct { - ID string - Name string - Arguments string - Index int + ID string + Name string + Arguments string + InvalidJSONStr string + InvalidJSONErr string + IsCustomToolCall bool } // DevinImage represents an image attachment in a DevinPrompt (Prompt #10). @@ -98,6 +100,7 @@ type DevinUsage struct { PromptTokens int64 `json:"prompt_tokens"` CompletionTokens int64 `json:"completion_tokens"` CachedTokens int64 `json:"cached_tokens"` + CacheWriteTokens int64 `json:"cache_write_tokens,omitempty"` StatusCode uint64 `json:"status_code,omitempty"` RequestID string `json:"request_id,omitempty"` ModelName string `json:"model_name,omitempty"` @@ -253,7 +256,7 @@ func BuildDevinClientMetadataBytes(sessionToken, deviceSeed, osName string) []by var f1Bytes []byte f1Bytes = protowire.AppendTag(f1Bytes, 1, protowire.BytesType) - f1Bytes = protowire.AppendString(f1Bytes, "devin-cli") + f1Bytes = protowire.AppendString(f1Bytes, DevinDefaultClientName) f1Bytes = protowire.AppendTag(f1Bytes, 2, protowire.BytesType) f1Bytes = protowire.AppendString(f1Bytes, DevinDefaultClientVersion) @@ -273,9 +276,6 @@ func BuildDevinClientMetadataBytes(sessionToken, deviceSeed, osName string) []by f1Bytes = protowire.AppendTag(f1Bytes, 12, protowire.BytesType) f1Bytes = protowire.AppendString(f1Bytes, DevinDefaultClientName) - f1Bytes = protowire.AppendTag(f1Bytes, 28, protowire.BytesType) - f1Bytes = protowire.AppendString(f1Bytes, DevinDefaultClientName) - f1Bytes = protowire.AppendTag(f1Bytes, 31, protowire.BytesType) f1Bytes = protowire.AppendString(f1Bytes, deviceFingerprint) return f1Bytes @@ -674,8 +674,8 @@ func parseDevinToolCallDelta(data []byte) (DevinToolCallDelta, error) { return tc, protowire.ParseError(vn) } pos += vn - if num == 4 { - tc.Index = int(v) + if num == 6 { + tc.IsCustomToolCall = (v != 0) } case protowire.BytesType: val, bn := protowire.ConsumeBytes(data[pos:]) @@ -690,6 +690,10 @@ func parseDevinToolCallDelta(data []byte) (DevinToolCallDelta, error) { tc.Name = string(val) case 3: tc.Arguments = string(val) + case 4: + tc.InvalidJSONStr = string(val) + case 5: + tc.InvalidJSONErr = string(val) } default: nSkip := protowire.ConsumeFieldValue(num, typ, data[pos:]) @@ -785,8 +789,8 @@ func parseDevinUsageField(data []byte) *DevinUsage { u.PromptTokens += int64(v) case 3: // Output tokens u.CompletionTokens = int64(v) - case 4: // Additional context/system prompt tokens in OpenAI-family models (total prompt = 2 + 4) - u.PromptTokens += int64(v) + case 4: // Cache write tokens + u.CacheWriteTokens += int64(v) case 5: // Cache read tokens u.CachedTokens = int64(v) case 6: // Status code diff --git a/internal/runtime/executor/helps/devin_wire_test.go b/internal/runtime/executor/helps/devin_wire_test.go index 3ff42007b..5d06ad8bb 100644 --- a/internal/runtime/executor/helps/devin_wire_test.go +++ b/internal/runtime/executor/helps/devin_wire_test.go @@ -679,9 +679,11 @@ func TestParseDevinUsageField_HeadersAndField4(t *testing.T) { t.Fatal("expected non-nil usage") } - // 3 + 58 = 61 - if usage.PromptTokens != 61 { - t.Errorf("PromptTokens = %d, want 61 (3 turn + 58 context)", usage.PromptTokens) + if usage.PromptTokens != 3 { + t.Errorf("PromptTokens = %d, want 3", usage.PromptTokens) + } + if usage.CacheWriteTokens != 58 { + t.Errorf("CacheWriteTokens = %d, want 58", usage.CacheWriteTokens) } if usage.CompletionTokens != 39 { t.Errorf("CompletionTokens = %d, want 39", usage.CompletionTokens) @@ -875,3 +877,116 @@ func TestBuildDevinGetChatMessageRequest_FiltersAutomationUpdateAndObfuscatesDes t.Fatalf("wire bytes should contain 'to a existing unified'") } } + +func TestRegressionIssue5910_ClientMetadata(t *testing.T) { + b := BuildDevinClientMetadataBytes("test-session-token", "device-seed", "linux") + pos := 0 + var ideName string + hasTag28 := false + + for pos < len(b) { + num, typ, n := protowire.ConsumeTag(b[pos:]) + if n <= 0 { + t.Fatalf("corrupt tag at %d", pos) + } + pos += n + + if num == 1 && typ == protowire.BytesType { + val, bn := protowire.ConsumeBytes(b[pos:]) + if bn <= 0 { + t.Fatalf("corrupt bytes at %d", pos) + } + pos += bn + ideName = string(val) + } else if num == 28 { + hasTag28 = true + nSkip := protowire.ConsumeFieldValue(num, typ, b[pos:]) + if nSkip <= 0 { + t.Fatalf("corrupt field at %d", pos) + } + pos += nSkip + } else { + nSkip := protowire.ConsumeFieldValue(num, typ, b[pos:]) + if nSkip <= 0 { + t.Fatalf("corrupt field at %d", pos) + } + pos += nSkip + } + } + + if ideName != DevinDefaultClientName { + t.Errorf("BuildDevinClientMetadataBytes field 1 = %q, want %q", ideName, DevinDefaultClientName) + } + if hasTag28 { + t.Errorf("BuildDevinClientMetadataBytes should not emit field 28") + } +} + +func TestRegressionIssue5910_UsageStatsCacheWriteTokens(t *testing.T) { + var f7Bytes []byte + // Field 2: input_tokens = 3 + f7Bytes = protowire.AppendTag(f7Bytes, 2, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 3) + + // Field 4: cache_write_tokens = 14361 + f7Bytes = protowire.AppendTag(f7Bytes, 4, protowire.VarintType) + f7Bytes = protowire.AppendVarint(f7Bytes, 14361) + + usage := parseDevinUsageField(f7Bytes) + if usage == nil { + t.Fatal("expected non-nil usage") + } + + if usage.PromptTokens != 3 { + t.Errorf("PromptTokens = %d, want 3 (cache_write_tokens must not inflate prompt_tokens)", usage.PromptTokens) + } + if usage.CacheWriteTokens != 14361 { + t.Errorf("CacheWriteTokens = %d, want 14361", usage.CacheWriteTokens) + } +} + +func TestRegressionIssue5910_ToolCallDeltaFields(t *testing.T) { + var tcBytes []byte + // Field 1: id + tcBytes = protowire.AppendTag(tcBytes, 1, protowire.BytesType) + tcBytes = protowire.AppendString(tcBytes, "call_999") + // Field 2: name + tcBytes = protowire.AppendTag(tcBytes, 2, protowire.BytesType) + tcBytes = protowire.AppendString(tcBytes, "custom_bash") + // Field 3: arguments + tcBytes = protowire.AppendTag(tcBytes, 3, protowire.BytesType) + tcBytes = protowire.AppendString(tcBytes, `{"cmd":"pwd"}`) + // Field 4: invalid_json_str + tcBytes = protowire.AppendTag(tcBytes, 4, protowire.BytesType) + tcBytes = protowire.AppendString(tcBytes, `pwd && ls`) + // Field 5: invalid_json_err + tcBytes = protowire.AppendTag(tcBytes, 5, protowire.BytesType) + tcBytes = protowire.AppendString(tcBytes, "syntax error near unexpected token") + // Field 6: is_custom_tool_call + tcBytes = protowire.AppendTag(tcBytes, 6, protowire.VarintType) + tcBytes = protowire.AppendVarint(tcBytes, 1) + + tc, err := parseDevinToolCallDelta(tcBytes) + if err != nil { + t.Fatalf("parseDevinToolCallDelta failed: %v", err) + } + + if tc.ID != "call_999" { + t.Errorf("tc.ID = %q, want call_999", tc.ID) + } + if tc.Name != "custom_bash" { + t.Errorf("tc.Name = %q, want custom_bash", tc.Name) + } + if tc.Arguments != `{"cmd":"pwd"}` { + t.Errorf("tc.Arguments = %q, want {\"cmd\":\"pwd\"}", tc.Arguments) + } + if tc.InvalidJSONStr != "pwd && ls" { + t.Errorf("tc.InvalidJSONStr = %q, want 'pwd && ls'", tc.InvalidJSONStr) + } + if tc.InvalidJSONErr != "syntax error near unexpected token" { + t.Errorf("tc.InvalidJSONErr = %q, want 'syntax error near unexpected token'", tc.InvalidJSONErr) + } + if !tc.IsCustomToolCall { + t.Errorf("tc.IsCustomToolCall = %v, want true", tc.IsCustomToolCall) + } +}