Skip to content

Commit e2a1404

Browse files
luoxuanzaoQoder-AI
andauthored
fix(chat): buffer successor chunks across the steer splice window (#41)
* fix(chat): buffer successor chunks across the steer splice window The steer splice finalizes the interrupted bubble asynchronously before swapping the render target; successor chunks arriving in that window were rendered into the old bubble, dropping/misplacing the steered reply's opening. Buffer them in the send loop while the splice runs and replay them into the fresh bubble in order afterwards. Co-authored-by: QoderAI (Qwen 3.8 Max) <qoder_ai@qoder.com> * fix(chat): keep steer successor response streaming Co-authored-by: QoderAI (Qwen 3.8 Max) <qoder_ai@qoder.com> --------- Co-authored-by: QoderAI (Qwen 3.8 Max) <qoder_ai@qoder.com>
1 parent 685d05c commit e2a1404

4 files changed

Lines changed: 87 additions & 12 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,12 @@ version with its date and start a fresh empty `[Unreleased]` above it.
6666
failure) now fully clean up the streaming state and queue indicator,
6767
so the composer no longer stays stuck as "streaming" after an early
6868
send error.
69+
- Steering a queued message no longer drops or misplaces the opening of
70+
the steered reply: chunks that arrive while the chat swaps streaming
71+
bubbles are buffered and replayed into the fresh bubble in order.
72+
- The steered reply streams incrementally again: the interrupted turn's
73+
receipt no longer triggers a context-usage request that stalled the
74+
response consumer, which made the whole reply appear only at the end.
6975

7076
## [1.0.5] - 2026-08-18
7177

‎src/features/chat/controllers/input-controller.ts‎

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,13 @@ export class InputController {
7979
private readonly inputCommands: InputCommandController;
8080
private readonly queuedMessages: QueuedMessageController;
8181
private activeStreamingAssistantMessage: ChatMessage | null = null;
82+
// While a steer splice is swapping the render target (finalizing the old
83+
// bubble is async), the send loop buffers successor chunks instead of
84+
// rendering them into the old bubble; the splice replays them into the
85+
// fresh bubble afterwards. Without this, the steered reply's opening
86+
// tokens land in the interrupted bubble and look lost/misplaced.
87+
private steerSpliceActive = false;
88+
private steerSpliceBuffer: StreamChunk[] = [];
8289
// Contract: `user_message_start` / `assistant_message_start` boundary
8390
// chunks currently have no producer in the runtime — they are a reserved
8491
// mechanism, so the handlers below are dormant. Steering relies on
@@ -393,6 +400,11 @@ export class InputController {
393400
continue;
394401
}
395402

403+
if (this.steerSpliceActive) {
404+
this.steerSpliceBuffer.push(chunk);
405+
continue;
406+
}
407+
396408
await streamController.handleStreamChunk(
397409
chunk,
398410
this.activeStreamingAssistantMessage ?? assistantMsg,
@@ -569,13 +581,53 @@ export class InputController {
569581
if (!text) return false;
570582
const accepted = this.getAgentService()?.steerTurn?.(text) ?? false;
571583
if (!accepted) return false;
572-
void this.spliceRuntimeUserMessage({
584+
this.beginSteerSplice({
573585
displayContent: turn.displayContent,
574586
persistedContent: text,
575587
});
576588
return true;
577589
}
578590

591+
/**
592+
* Runs the steer splice with chunk buffering active: successor chunks that
593+
* arrive while the splice awaits the old bubble's finalization are held
594+
* back and replayed into the fresh bubble once the splice completes.
595+
*/
596+
private beginSteerSplice(details: { displayContent: string; persistedContent?: string }): void {
597+
if (!this.steerSpliceActive) {
598+
this.steerSpliceActive = true;
599+
this.steerSpliceBuffer = [];
600+
}
601+
void this.spliceRuntimeUserMessage(details)
602+
.catch(() => undefined)
603+
.then(() => this.drainSteerSpliceBuffer())
604+
.catch(() => {
605+
this.steerSpliceActive = false;
606+
this.steerSpliceBuffer = [];
607+
});
608+
}
609+
610+
/** Replays chunks buffered during the splice, then reopens direct rendering. */
611+
private async drainSteerSpliceBuffer(): Promise<void> {
612+
try {
613+
while (this.steerSpliceBuffer.length > 0) {
614+
const chunk = this.steerSpliceBuffer.shift()!;
615+
const target = this.activeStreamingAssistantMessage;
616+
if (target) await this.deps.streamController.handleStreamChunk(chunk, target);
617+
}
618+
} catch (error) {
619+
const target = this.activeStreamingAssistantMessage;
620+
if (target) {
621+
const message = error instanceof Error ? error.message : 'Unknown error';
622+
await this.deps.streamController
623+
.handleStreamChunk({ type: 'error', content: message }, target)
624+
.catch(() => undefined);
625+
}
626+
} finally {
627+
this.steerSpliceActive = false;
628+
}
629+
}
630+
579631
private async buildTurnSubmission(options: {
580632
content: string;
581633
images?: ChatMessage['images'];

‎src/qoder/runtime/qoder-response-router.ts‎

Lines changed: 14 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,19 @@ export class QoderResponseRouter {
145145

146146
if (!isTurnCompleteMessage(message)) return;
147147

148+
// A priority-'now' steer emits an aborted result for the interrupted
149+
// turn, but the persistent Query is already running its successor turn.
150+
// Do not ask the Query for context usage here: that control request waits
151+
// for the successor to become idle, which blocks this sole response
152+
// consumer and batches every successor delta until the turn finishes.
153+
// The message channel must also remain active so another steer can be
154+
// accepted while the successor is running.
155+
if (isAbortedResult(message)) {
156+
handler?.resetStreamText();
157+
handler?.resetStreamThinking();
158+
return;
159+
}
160+
148161
const contextUsageChunk = await this.deps.turnTracker.fetchContextUsage({
149162
query: this.deps.getCurrentQuery(),
150163
isCurrentQuery: query => this.deps.getCurrentQuery() === query,
@@ -160,13 +173,7 @@ export class QoderResponseRouter {
160173
if (handler) {
161174
handler.resetStreamText();
162175
handler.resetStreamThinking();
163-
// A deliberately aborted turn (Esc or priority-'now' steer) is followed
164-
// by its successor turn or by consumer teardown — never by session
165-
// idle with the handler still waiting. Keep the handler alive so the
166-
// successor's chunks keep streaming into the same turn generator.
167-
if (!isAbortedResult(message)) {
168-
handler.onDone();
169-
}
176+
handler.onDone();
170177
} else {
171178
await this.flushAutoTurnBuffer();
172179
}

‎tests/unit/qoder/runtime/qoder-response-router.test.ts‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,11 @@ import { createResponseHandler } from '@/qoder/runtime/types';
77

88
function createRouterHarness() {
99
const onTurnComplete = jest.fn();
10+
const fetchContextUsage = jest.fn().mockResolvedValue(null);
1011
const deps = {
1112
turnTracker: {
1213
getTransformOptions: () => ({}),
13-
fetchContextUsage: async () => null,
14+
fetchContextUsage,
1415
bufferUsage: (chunk: StreamChunk) => chunk,
1516
updateContextWindow: () => null,
1617
record: jest.fn(),
@@ -35,7 +36,7 @@ function createRouterHarness() {
3536
});
3637
router.register(handler);
3738

38-
return { router, handler, chunks, events, onTurnComplete };
39+
return { router, handler, chunks, events, fetchContextUsage, onTurnComplete };
3940
}
4041

4142
function buildAbortResult(): SDKResultError {
@@ -54,18 +55,27 @@ describe('QoderResponseRouter', () => {
5455

5556
await router.route(buildAbortResult());
5657

57-
expect(onTurnComplete).toHaveBeenCalledTimes(1);
58+
expect(onTurnComplete).not.toHaveBeenCalled();
5859
expect(events.onDone).not.toHaveBeenCalled();
5960
});
6061

62+
it('does not block successor streaming on a context-usage request', async () => {
63+
const { router, fetchContextUsage } = createRouterHarness();
64+
65+
await router.route(buildAbortResult());
66+
67+
expect(fetchContextUsage).not.toHaveBeenCalled();
68+
});
69+
6170
it('completes the handler on the successor turn result', async () => {
62-
const { router, events } = createRouterHarness();
71+
const { router, events, onTurnComplete } = createRouterHarness();
6372

6473
await router.route(buildAbortResult());
6574
expect(events.onDone).not.toHaveBeenCalled();
6675

6776
await router.route(buildSDKMessage({ type: 'result', subtype: 'success' }));
6877
expect(events.onDone).toHaveBeenCalledTimes(1);
78+
expect(onTurnComplete).toHaveBeenCalledTimes(1);
6979
});
7080

7181
it('routes nothing to the handler for an abort receipt', async () => {

0 commit comments

Comments
 (0)