fix(devin): aggregate tool calls by id and track cache write tokens

- Track and aggregate tool calls by call ID instead of slot index in streaming and buffered execution.
- Support raw arguments from invalid JSON fields for custom tool calls.
- Parse usage field 4 as cache write tokens instead of adding to prompt tokens.
- Align client metadata with the default client name and drop deprecated tag 28.

Closes: #5910
This commit is contained in:
Luis Pater
2026-09-18 01:04:43 +08:00
parent 0b55053944
commit 9e10db53ad
4 changed files with 619 additions and 72 deletions

View File

@@ -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)
}

View File

@@ -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)
}
}

View File

@@ -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

View File

@@ -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)
}
}