Skip to content

Commit e906e15

Browse files
claudeclaude
committed
fix(core): avoid duplicate content in OpenAI streaming when both delta and message have content
This fix addresses an issue where MiniMax and similar models send both delta content in choice.delta and full content in choice.message.content, causing the full content to be sent twice (once via delta, once via message). Changes: - Added skip_content flag when text_content already has content from delta processing - Changed message.content handling to skip when skip_content is true - Changed delta handling to use else if to properly handle mutual exclusivity - Added merge_stream_text fix to handle case where incoming contains text_content as prefix Test files added for verification: - test_fix_verification.ts - Main verification script - test_fix_minimax.ts - MiniMax-specific tests - test_fix_verification_kimi.ts - Kimi model tests - test_duplicate_simple.ts - Simple duplicate detection - debug_sse.ts - SSE debugging utility - test_minimax_stream.ts - Streaming tests - test_minimax.hcl - MiniMax config
1 parent 3ee7897 commit e906e15

9 files changed

Lines changed: 726 additions & 77 deletions

File tree

‎core/src/llm/openai.rs‎

Lines changed: 62 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,16 @@ impl OpenAiClient {
6060
if incoming == text_content.as_str() || text_content.ends_with(incoming) {
6161
return None;
6262
}
63+
// If incoming contains text_content as a prefix (incoming is the full content),
64+
// replace text_content instead of appending (avoids duplicate full content)
65+
if incoming.starts_with(text_content.as_str()) && incoming.len() > text_content.len() {
66+
let suffix = &incoming[text_content.len()..];
67+
if !suffix.is_empty() {
68+
*text_content = incoming.to_string();
69+
return Some(suffix.to_string());
70+
}
71+
return None;
72+
}
6373
if let Some(suffix) = incoming.strip_prefix(text_content.as_str()) {
6474
if suffix.is_empty() {
6575
return None;
@@ -647,39 +657,28 @@ impl LlmClient for OpenAiClient {
647657
finish_reason = Some(reason);
648658
}
649659

650-
let delta_content = choice
651-
.delta
652-
.as_ref()
653-
.and_then(|delta| delta.content.clone());
654-
655660
if let Some(message) = choice.message {
661+
// If text_content already has content (from delta processing),
662+
// skip message.content to avoid sending duplicate full content
663+
let skip_content = !text_content.is_empty();
656664
if let Some(reasoning) = message.reasoning_content {
657665
reasoning_content_accum.push_str(&reasoning);
658-
// For models like kimi that use reasoning_content for actual text output,
659-
// emit it as TextDelta when delta_content is empty (None or "")
660-
let delta_content_empty = delta_content
661-
.as_deref()
662-
.is_none_or(str::is_empty);
663-
if delta_content_empty {
664-
if first_token_ms.is_none() {
665-
first_token_ms = Some(
666-
request_started_at.elapsed().as_millis()
667-
as u64,
668-
);
669-
}
670-
if let Some(delta) = Self::merge_stream_text(
671-
&mut text_content,
672-
&reasoning,
673-
) {
674-
let _ = tx
675-
.send(StreamEvent::ReasoningDelta(
676-
delta,
677-
))
678-
.await;
679-
}
666+
if first_token_ms.is_none() {
667+
first_token_ms = Some(
668+
request_started_at.elapsed().as_millis()
669+
as u64,
670+
);
671+
}
672+
if let Some(delta) = Self::merge_stream_text(
673+
&mut text_content,
674+
&reasoning,
675+
) {
676+
let _ = tx
677+
.send(StreamEvent::ReasoningDelta(delta))
678+
.await;
680679
}
681680
}
682-
if delta_content.as_deref().is_none_or(str::is_empty) {
681+
if !skip_content {
683682
if let Some(content) = message
684683
.content
685684
.filter(|value| !value.is_empty())
@@ -712,34 +711,21 @@ impl LlmClient for OpenAiClient {
712711
);
713712
}
714713
}
715-
}
716-
717-
if let Some(delta) = choice.delta {
714+
} else if let Some(delta) = choice.delta {
718715
if let Some(ref rc) = delta.reasoning_content {
719716
reasoning_content_accum.push_str(rc);
720-
// For models like kimi that use reasoning_content for actual text output,
721-
// emit it as TextDelta when content is empty (None or "")
722-
let content_empty = delta
723-
.content
724-
.as_deref()
725-
.is_none_or(|s| s.is_empty());
726-
if content_empty {
727-
if first_token_ms.is_none() {
728-
first_token_ms = Some(
729-
request_started_at.elapsed().as_millis()
730-
as u64,
731-
);
732-
}
733-
if let Some(delta) = Self::merge_stream_text(
734-
&mut text_content,
735-
rc,
736-
) {
737-
let _ = tx
738-
.send(StreamEvent::ReasoningDelta(
739-
delta,
740-
))
741-
.await;
742-
}
717+
if first_token_ms.is_none() {
718+
first_token_ms = Some(
719+
request_started_at.elapsed().as_millis()
720+
as u64,
721+
);
722+
}
723+
if let Some(delta) =
724+
Self::merge_stream_text(&mut text_content, rc)
725+
{
726+
let _ = tx
727+
.send(StreamEvent::ReasoningDelta(delta))
728+
.await;
743729
}
744730
}
745731

@@ -844,15 +830,23 @@ impl LlmClient for OpenAiClient {
844830
if let Some(reason) = choice.finish_reason {
845831
finish_reason = Some(reason);
846832
}
847-
let delta_content = choice
848-
.delta
849-
.as_ref()
850-
.and_then(|delta| delta.content.clone());
833+
// If text_content already has content (from delta processing),
834+
// skip message.content to avoid sending duplicate full content
835+
let skip_content = !text_content.is_empty();
851836
if let Some(message) = choice.message {
852837
if let Some(reasoning) = message.reasoning_content {
853838
reasoning_content_accum.push_str(&reasoning);
839+
if first_token_ms.is_none() {
840+
first_token_ms =
841+
Some(request_started_at.elapsed().as_millis() as u64);
842+
}
843+
if let Some(delta) =
844+
Self::merge_stream_text(&mut text_content, &reasoning)
845+
{
846+
let _ = tx.send(StreamEvent::ReasoningDelta(delta)).await;
847+
}
854848
}
855-
if delta_content.as_deref().is_none_or(str::is_empty) {
849+
if !skip_content {
856850
if let Some(content) =
857851
message.content.filter(|value| !value.is_empty())
858852
{
@@ -876,26 +870,17 @@ impl LlmClient for OpenAiClient {
876870
);
877871
}
878872
}
879-
}
880-
if let Some(delta) = choice.delta {
873+
} else if let Some(delta) = choice.delta {
881874
if let Some(ref rc) = delta.reasoning_content {
882875
reasoning_content_accum.push_str(rc);
883-
// For models like kimi that use reasoning_content for actual text output,
884-
// emit it as TextDelta when content is empty (None or "")
885-
let content_empty =
886-
delta.content.as_deref().is_none_or(|s| s.is_empty());
887-
if content_empty {
888-
if first_token_ms.is_none() {
889-
first_token_ms = Some(
890-
request_started_at.elapsed().as_millis() as u64,
891-
);
892-
}
893-
if let Some(delta) =
894-
Self::merge_stream_text(&mut text_content, rc)
895-
{
896-
let _ =
897-
tx.send(StreamEvent::ReasoningDelta(delta)).await;
898-
}
876+
if first_token_ms.is_none() {
877+
first_token_ms =
878+
Some(request_started_at.elapsed().as_millis() as u64);
879+
}
880+
if let Some(delta) =
881+
Self::merge_stream_text(&mut text_content, rc)
882+
{
883+
let _ = tx.send(StreamEvent::ReasoningDelta(delta)).await;
899884
}
900885
}
901886
if let Some(content) = delta.content {

‎sdk/node/examples/streaming/debug_sse.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
#!/usr/bin/env npx tsx
2+
/**
3+
* Debug SSE events to see raw chunk structure.
4+
*/
5+
6+
import { Agent } from '../../index.js';
7+
8+
async function main(): Promise<void> {
9+
const configPath = '/Users/roylin/Desktop/code/a3s/crates/code/sdk/node/examples/streaming/test_minimax.hcl';
10+
11+
const agent = await Agent.create(configPath);
12+
const session = agent.session('.', {
13+
permissive: true,
14+
max_tool_rounds: 0,
15+
});
16+
17+
const prompt = 'Say hello in 3 words.';
18+
console.log(`Prompt: "${prompt}"`);
19+
20+
const stream = await session.stream(prompt);
21+
22+
let eventCount = 0;
23+
24+
while (true) {
25+
const result = await stream.next();
26+
if (!result.value || result.done) break;
27+
28+
const event = result.value;
29+
eventCount++;
30+
31+
// Log all events
32+
console.log(`[${eventCount}] type="${event.type}"` +
33+
(event.text ? ` text="${event.text.slice(0, 30)}..."` : '') +
34+
(event.data ? ` data="${JSON.stringify(event.data)?.slice(0, 60)}..."` : '')
35+
);
36+
37+
// Stop after 30 events to avoid flooding
38+
if (eventCount >= 30) {
39+
console.log('\n... (stopping after 30 events)');
40+
break;
41+
}
42+
}
43+
}
44+
45+
main().catch(err => {
46+
console.error('Error:', err);
47+
process.exit(1);
48+
});

‎sdk/node/examples/streaming/test_duplicate_simple.ts‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
#!/usr/bin/env npx tsx
2+
/**
3+
* Ultra-simple test to check for duplicate text_delta.
4+
*/
5+
6+
import { Agent } from '../../index.js';
7+
8+
async function main(): Promise<void> {
9+
const configPath = '/Users/roylin/Desktop/code/a3s/crates/code/sdk/node/examples/streaming/test_minimax.hcl';
10+
11+
const agent = await Agent.create(configPath);
12+
const session = agent.session('.', {
13+
permissive: true,
14+
max_tool_rounds: 0,
15+
});
16+
17+
// Use a very simple prompt
18+
const prompt = 'Hi';
19+
console.log(`Prompt: "${prompt}"`);
20+
21+
const stream = await session.stream(prompt);
22+
23+
const textDeltas: string[] = [];
24+
const seen = new Set<string>();
25+
let duplicateFound = false;
26+
let eventCount = 0;
27+
28+
while (true) {
29+
const result = await stream.next();
30+
if (!result.value || result.done) break;
31+
32+
const event = result.value;
33+
eventCount++;
34+
35+
if (event.type === 'text_delta' && event.text) {
36+
const text = event.text;
37+
textDeltas.push(text);
38+
39+
if (seen.has(text)) {
40+
console.log(`❌ DUPLICATE FOUND: "${text}" (event ${eventCount})`);
41+
duplicateFound = true;
42+
} else {
43+
seen.add(text);
44+
console.log(`[${eventCount}] text_delta: "${text}"`);
45+
}
46+
}
47+
}
48+
49+
console.log('\n--- Summary ---');
50+
console.log(`Total text_delta events: ${textDeltas.length}`);
51+
console.log(`Unique text_delta events: ${seen.size}`);
52+
53+
if (duplicateFound) {
54+
console.log('\n❌ DUPLICATES DETECTED');
55+
process.exit(1);
56+
} else {
57+
console.log('\n✅ No duplicates');
58+
}
59+
}
60+
61+
main().catch(err => {
62+
console.error('Error:', err);
63+
process.exit(1);
64+
});

‎sdk/node/examples/streaming/test_fix_minimax.ts‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
#!/usr/bin/env npx tsx
2+
/**
3+
* Quick test to verify no duplicate text_delta/reasoning_delta events.
4+
* Uses MiniMax model from config file.
5+
*
6+
* Run with:
7+
* npx tsx sdk/node/examples/streaming/test_fix_minimax.ts
8+
*/
9+
10+
import { Agent } from '../../index.js';
11+
12+
async function main(): Promise<void> {
13+
const configPath = '/Users/roylin/Desktop/code/a3s/crates/code/sdk/node/examples/streaming/test_minimax.hcl';
14+
console.log(`Using config: ${configPath}\n`);
15+
16+
const agent = await Agent.create(configPath);
17+
const session = agent.session('.', {
18+
permissive: true,
19+
max_tool_rounds: 0,
20+
});
21+
22+
const prompt = 'Say hello in 3 Chinese characters.';
23+
console.log(`Streaming with prompt: "${prompt}"\n`);
24+
25+
const stream = await session.stream(prompt);
26+
27+
const textDeltas: string[] = [];
28+
const reasoningDeltas: string[] = [];
29+
let eventCount = 0;
30+
31+
while (true) {
32+
const result = await stream.next();
33+
if (!result.value || result.done) break;
34+
35+
const event = result.value;
36+
eventCount++;
37+
const type = event.type || 'unknown';
38+
39+
if (type === 'text_delta' && event.text) {
40+
textDeltas.push(event.text);
41+
console.log(`[${eventCount}] text_delta: "${event.text}"`);
42+
} else if (type === 'reasoning_delta' && event.text) {
43+
reasoningDeltas.push(event.text);
44+
if (eventCount <= 3) {
45+
console.log(`[${eventCount}] reasoning_delta: "${event.text.slice(0, 40)}..."`);
46+
}
47+
}
48+
}
49+
50+
console.log('\n' + '='.repeat(80));
51+
console.log('Summary:');
52+
console.log(`Total events: ${eventCount}`);
53+
console.log(`TextDelta count: ${textDeltas.length}`);
54+
console.log(`ReasoningDelta count: ${reasoningDeltas.length}`);
55+
56+
const combinedText = textDeltas.join('');
57+
console.log(`\nFinal text: "${combinedText}"`);
58+
59+
// Check for duplicates
60+
const uniqueChunks = new Set(textDeltas);
61+
console.log(`Unique chunks: ${uniqueChunks.size} / ${textDeltas.length}`);
62+
63+
if (textDeltas.length >= 2) {
64+
const halfLen = Math.floor(combinedText.length / 2);
65+
const firstHalf = combinedText.slice(0, halfLen);
66+
const secondHalf = combinedText.slice(halfLen);
67+
68+
if (firstHalf === secondHalf && combinedText.length > 4) {
69+
console.log('\n❌ DUPLICATE DETECTED: Full text appears twice!');
70+
process.exit(1);
71+
}
72+
}
73+
74+
const chunkCounts = new Map<string, number>();
75+
for (const chunk of textDeltas) {
76+
if (chunk.length > 2) {
77+
chunkCounts.set(chunk, (chunkCounts.get(chunk) || 0) + 1);
78+
}
79+
}
80+
81+
for (const [chunk, count] of chunkCounts) {
82+
if (count > 1) {
83+
console.log(`\n❌ DUPLICATE CHUNK: "${chunk}" appears ${count} times`);
84+
process.exit(1);
85+
}
86+
}
87+
88+
console.log('\n✅ No duplicates detected!');
89+
}
90+
91+
main().catch((err: unknown) => {
92+
console.error('Test failed:', err);
93+
process.exit(1);
94+
});

0 commit comments

Comments
 (0)