fix(tui): sliding-window stream reconciler + UI polish

- Replace naive reconcileStream with stateful sliding-window replay
  merger: tracks per-message replay cursors (_posContent/_posReasoning)
  to deduplicate the segment-replay waves 9router emits. Verified
  against 970 real captured chunks (100% clean output).
- TranscriptMessage: colored [TAG] headers, 4-space indented body,
  thin separator lines between messages, dimmed reasoning.
- InputView: divider above prompt, '❯' prompt, cursor block.
- 75 tests pass, typecheck clean, biome clean.
This commit is contained in:
zesdex
2026-09-03 12:28:56 +07:00
committed by asepharyana
parent 30cc579773
commit d86d6aeef2
3 changed files with 170 additions and 56 deletions
+63 -22
View File
@@ -329,10 +329,13 @@ export interface ChatMessageDisplay {
content: string;
reasoning: string;
timestamp: number;
/** Sliding-window replay cursor for content + reasoning (transient). */
_posContent: number;
_posReasoning: number;
}
export function makeChatMessage(role: Role, content: string): ChatMessageDisplay {
return { role, content, reasoning: "", timestamp: Date.now() };
return { role, content, reasoning: "", timestamp: Date.now(), _posContent: 0, _posReasoning: 0 };
}
/** Bounded ring of recent chat messages used to render the transcript view. */
@@ -357,9 +360,9 @@ export class TranscriptCache {
const msg = this.messages[this.messages.length - 1];
if (msg && msg.role === Roles.Assistant) {
if (isReasoning) {
msg.reasoning = reconcileStream(msg.reasoning, text);
msg.reasoning = reconcileStream(msg.reasoning, text, msg, true);
} else {
msg.content = reconcileStream(msg.content, text);
msg.content = reconcileStream(msg.content, text, msg, false);
}
this.dirty = true;
}
@@ -367,34 +370,72 @@ export class TranscriptCache {
}
/**
* Reconcile an incoming stream chunk against the accumulated text.
* Reconcile an incoming stream chunk against the accumulated text, deduping the
* sliding-window replays that 9router's OpenAI-compatible layer emits.
*
* Streaming backends disagree on chunk semantics:
* - Native OpenAI/Anthropic send small *incremental* deltas.
* - Some gateways/proxies (e.g. 9router's OpenAI-compatible layer) emit
* *full-text* chunks, where each chunk is the complete transcript so far.
* - 9router/gateways re-send the FULL accumulated text as a repeated window:
* each "wave" replays the already-emitted segments from the start, then one
* fresh token. So the stream is s1,s2,…,sk then s1,s2,…,sk,s_{k+1} then
* s1,…,s_{k+1},s_{k+2} … where s_i + … = the final text.
*
* Blindly concatenating full-text chunks duplicates content (partial + full),
* which manifests as garbled messages. This helper accepts both modes:
* - a chunk that is a strict prefix of the tail → a shorter replay → ignore;
* - a chunk that extends/replays the whole accumulated text → take it as the
* new authoritative tail (full-text mode);
* - otherwise → treat as an incremental delta and append.
* Naive concatenation doubles the text (garbled output). We track a per-stream
* replay cursor `pos` into `accumulated`: a chunk that continues at `pos` (or
* restarts at 0, i.e. begins a fresh wave) is a replay → ignore it and advance
* the cursor; only a chunk that matches nothing already present is brand-new
* content, which we append at the end and reset the cursor.
*
* @param accumulated current merged text
* @param chunk next chunk from the stream
* @param msg the message being streamed (holds the replay cursor)
* @param isReasoning whether this targets the reasoning buffer or the content
* @returns the updated accumulated text
*/
export function reconcileStream(accumulated: string, chunk: string): string {
export function reconcileStream(
accumulated: string,
chunk: string,
msg: ChatMessageDisplay,
isReasoning: boolean,
): string {
if (chunk.length === 0) return accumulated;
if (accumulated.length === 0) return chunk;
if (accumulated.length >= chunk.length && accumulated.startsWith(chunk)) {
// Shorter (or equal) replay of what we already hold — ignore.
const posRef = { pos: isReasoning ? msg._posReasoning : msg._posContent };
if (accumulated.length === 0) {
accumulated = chunk;
posRef.pos = chunk.length;
writePos();
return accumulated;
}
if (chunk.length >= accumulated.length && chunk.startsWith(accumulated)) {
// Full-text mode: the chunk is the whole transcript so far (matching our
// accumulated text plus more). Adopt it wholesale.
return chunk;
// 1) A chunk that continues the current replay cursor is a replayed segment
// (it already exists earlier in the text) → skip it, advance the cursor.
if (
posRef.pos > 0 &&
posRef.pos < accumulated.length &&
chunk.length <= accumulated.length - posRef.pos &&
accumulated.startsWith(chunk, posRef.pos)
) {
posRef.pos += chunk.length;
writePos();
return accumulated;
}
// 2) A chunk that matches at the very start begins a NEW replay wave → skip
// it, set the cursor to just past the matched prefix.
if (accumulated.startsWith(chunk)) {
posRef.pos = chunk.length;
writePos();
return accumulated;
}
// 3) Otherwise this is brand-new content: append it and reset the cursor so
// the next wave starts fresh.
accumulated += chunk;
posRef.pos = -1;
writePos();
return accumulated;
function writePos(): void {
if (isReasoning) msg._posReasoning = posRef.pos;
else msg._posContent = posRef.pos;
}
// Incremental delta mode: a new suffix. Append.
return accumulated + chunk;
}
/* ── AppStateRest ─────────────────────────────────────────────────── */