valknar/llmx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

hide CoT by default; show headers in status indicator (#2316)

Browse Source 
Plan is for full CoT summaries to be visible in a "transcript view" when
we implement that, but for now they're hidden.


https://github.com/user-attachments/assets/e8a1b0ef-8f2a-48ff-9625-9c3c67d92cdb
This commit is contained in:
Jeremy Rose
2025-08-20 16:58:56 -07:00
committed by GitHub
parent 2ec5a28528
commit e95cad1946
13 changed files with 231 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
codex-rs/tui/src/streaming/controller.rs
View File

@@ -2,7 +2,6 @@ use codex_core::config::Config;
use ratatui::text::Line;
use super::HeaderEmitter;
use super::StreamKind;
use super::StreamState;
/// Sink for history insertions and animation control.
@@ -36,8 +35,8 @@ type Lines = Vec<Line<'static>>;
pub(crate) struct StreamController {
config: Config,
header: HeaderEmitter,
states: [StreamState; 2],
current_stream: Option<StreamKind>,
state: StreamState,
active: bool,
finishing_after_drain: bool,
}
@@ -46,8 +45,8 @@ impl StreamController {
Self {
config,
header: HeaderEmitter::new(),
states: [StreamState::new(), StreamState::new()],
current_stream: None,
state: StreamState::new(),
active: false,
finishing_after_drain: false,
}
}
@@ -57,29 +56,18 @@ impl StreamController {
}
pub(crate) fn is_write_cycle_active(&self) -> bool {
self.current_stream.is_some()
self.active
}
pub(crate) fn clear_all(&mut self) {
self.states.iter_mut().for_each(|s| s.clear());
self.current_stream = None;
self.state.clear();
self.active = false;
self.finishing_after_drain = false;
// leave header state unchanged; caller decides when to reset
}
#[inline]
fn idx(kind: StreamKind) -> usize {
kind as usize
}
fn state(&self, kind: StreamKind) -> &StreamState {
&self.states[Self::idx(kind)]
}
fn state_mut(&mut self, kind: StreamKind) -> &mut StreamState {
&mut self.states[Self::idx(kind)]
}
fn emit_header_if_needed(&mut self, kind: StreamKind, out_lines: &mut Lines) -> bool {
self.header.maybe_emit(kind, out_lines)
fn emit_header_if_needed(&mut self, out_lines: &mut Lines) -> bool {
self.header.maybe_emit(out_lines)
}
#[inline]
@@ -93,56 +81,23 @@ impl StreamController {
}
}
/// Begin a stream, flushing previously completed lines from any other
/// active stream to maintain ordering.
pub(crate) fn begin(&mut self, kind: StreamKind, sink: &impl HistorySink) {
if let Some(current) = self.current_stream
&& current != kind
{
// Synchronously flush completed lines from previous stream.
let cfg = self.config.clone();
let prev_state = self.state_mut(current);
let newly_completed = prev_state.collector.commit_complete_lines(&cfg);
if !newly_completed.is_empty() {
prev_state.enqueue(newly_completed);
}
let step = prev_state.drain_all();
if !step.history.is_empty() {
let mut lines: Lines = Vec::new();
self.emit_header_if_needed(current, &mut lines);
lines.extend(step.history);
// Ensure at most one trailing blank after the flushed block.
Self::ensure_single_trailing_blank(&mut lines);
sink.insert_history(lines);
}
self.current_stream = None;
}
if self.current_stream != Some(kind) {
let prev = self.current_stream;
self.current_stream = Some(kind);
// Starting a new stream cancels any pending finish-from-previous-stream animation.
self.finishing_after_drain = false;
if prev.is_some() {
self.header.reset_for_stream(kind);
}
// Emit header immediately for reasoning; for answers, defer to first commit.
if matches!(kind, StreamKind::Reasoning) {
let mut header_lines = Vec::new();
if self.emit_header_if_needed(kind, &mut header_lines) {
sink.insert_history(header_lines);
}
}
/// Begin an answer stream. Does not emit header yet; it is emitted on first commit.
pub(crate) fn begin(&mut self, _sink: &impl HistorySink) {
// Starting a new stream cancels any pending finish-from-previous-stream animation.
if !self.active {
self.header.reset_for_stream();
}
self.finishing_after_drain = false;
self.active = true;
}
/// Push a delta; if it contains a newline, commit completed lines and start animation.
pub(crate) fn push_and_maybe_commit(&mut self, delta: &str, sink: &impl HistorySink) {
let Some(kind) = self.current_stream else {
if !self.active {
return;
};
}
let cfg = self.config.clone();
let state = self.state_mut(kind);
let state = &mut self.state;
// Record that at least one delta was received for this stream
if !delta.is_empty() {
state.has_seen_delta = true;
@@ -157,42 +112,22 @@ impl StreamController {
}
}
/// Insert a reasoning section break and commit any newly completed lines.
pub(crate) fn insert_reasoning_section_break(&mut self, sink: &impl HistorySink) {
if self.current_stream != Some(StreamKind::Reasoning) {
self.begin(StreamKind::Reasoning, sink);
}
let cfg = self.config.clone();
let state = self.state_mut(StreamKind::Reasoning);
state.collector.insert_section_break();
let newly_completed = state.collector.commit_complete_lines(&cfg);
if !newly_completed.is_empty() {
state.enqueue(newly_completed);
sink.start_commit_animation();
}
}
/// Finalize the active stream. If `flush_immediately` is true, drain and emit now.
pub(crate) fn finalize(
&mut self,
kind: StreamKind,
flush_immediately: bool,
sink: &impl HistorySink,
) -> bool {
if self.current_stream != Some(kind) {
pub(crate) fn finalize(&mut self, flush_immediately: bool, sink: &impl HistorySink) -> bool {
if !self.active {
return false;
}
let cfg = self.config.clone();
// Finalize collector first.
let remaining = {
let state = self.state_mut(kind);
let state = &mut self.state;
state.collector.finalize_and_drain(&cfg)
};
if flush_immediately {
// Collect all output first to avoid emitting headers when there is no content.
let mut out_lines: Lines = Vec::new();
{
let state = self.state_mut(kind);
let state = &mut self.state;
if !remaining.is_empty() {
state.enqueue(remaining);
}
@@ -201,28 +136,28 @@ impl StreamController {
}
if !out_lines.is_empty() {
let mut lines_with_header: Lines = Vec::new();
self.emit_header_if_needed(kind, &mut lines_with_header);
self.emit_header_if_needed(&mut lines_with_header);
lines_with_header.extend(out_lines);
Self::ensure_single_trailing_blank(&mut lines_with_header);
sink.insert_history(lines_with_header);
}
// Cleanup
self.state_mut(kind).clear();
// Allow a subsequent block of the same kind in this turn to emit its header.
self.header.allow_reemit_for_same_kind_in_turn(kind);
self.state.clear();
// Allow a subsequent block in this turn to emit its header.
self.header.allow_reemit_in_turn();
// Also clear the per-stream emitted flag so the header can render again.
self.header.reset_for_stream(kind);
self.current_stream = None;
self.header.reset_for_stream();
self.active = false;
self.finishing_after_drain = false;
true
} else {
if !remaining.is_empty() {
let state = self.state_mut(kind);
let state = &mut self.state;
state.enqueue(remaining);
}
// Spacer animated out
self.state_mut(kind).enqueue(vec![Line::from("")]);
self.state.enqueue(vec![Line::from("")]);
self.finishing_after_drain = true;
sink.start_commit_animation();
false
@@ -231,32 +166,29 @@ impl StreamController {
/// Step animation: commit at most one queued line and handle end-of-drain cleanup.
pub(crate) fn on_commit_tick(&mut self, sink: &impl HistorySink) -> bool {
let Some(kind) = self.current_stream else {
if !self.active {
return false;
};
let step = {
let state = self.state_mut(kind);
state.step()
};
}
let step = { self.state.step() };
if !step.history.is_empty() {
let mut lines: Lines = Vec::new();
self.emit_header_if_needed(kind, &mut lines);
self.emit_header_if_needed(&mut lines);
let mut out = lines;
out.extend(step.history);
sink.insert_history(out);
}
let is_idle = self.state(kind).is_idle();
let is_idle = self.state.is_idle();
if is_idle {
sink.stop_commit_animation();
if self.finishing_after_drain {
// Reset and notify
self.state_mut(kind).clear();
// Allow a subsequent block of the same kind in this turn to emit its header.
self.header.allow_reemit_for_same_kind_in_turn(kind);
self.state.clear();
// Allow a subsequent block in this turn to emit its header.
self.header.allow_reemit_in_turn();
// Also clear the per-stream emitted flag so the header can render again.
self.header.reset_for_stream(kind);
self.current_stream = None;
self.header.reset_for_stream();
self.active = false;
self.finishing_after_drain = false;
return true;
}
@@ -267,24 +199,14 @@ impl StreamController {
/// Apply a full final answer: replace queued content with only the remaining tail,
/// then finalize immediately and notify completion.
pub(crate) fn apply_final_answer(&mut self, message: &str, sink: &impl HistorySink) -> bool {
self.apply_full_final(StreamKind::Answer, message, true, sink)
self.apply_full_final(message, sink)
}
pub(crate) fn apply_final_reasoning(&mut self, message: &str, sink: &impl HistorySink) -> bool {
self.apply_full_final(StreamKind::Reasoning, message, false, sink)
}
fn apply_full_final(
&mut self,
kind: StreamKind,
message: &str,
immediate: bool,
sink: &impl HistorySink,
) -> bool {
self.begin(kind, sink);
fn apply_full_final(&mut self, message: &str, sink: &impl HistorySink) -> bool {
self.begin(sink);
{
let state = self.state_mut(kind);
let state = &mut self.state;
// Only inject the final full message if we have not seen any deltas for this stream.
// If deltas were received, rely on the collector's existing buffer to avoid duplication.
if !state.has_seen_delta && !message.is_empty() {
@@ -301,7 +223,6 @@ impl StreamController {
.replace_with_and_mark_committed(&msg, committed);
}
}
self.finalize(kind, immediate, sink)
self.finalize(true, sink)
}
}