fix(response): coalesce tiny reasoning deltas into small blocks

[jatmn]

Problem

When using models like kimi-k2.7-code that stream reasoning via the OpenAI chat-completions shape, codex-warp was forwarding every tiny provider fragment as its own response.reasoning_summary_text.delta SSE event. Codex clients then rendered the thinking one line/token at a time, instead of the small blocks normally seen with GPT-style Responses API streams.

Root cause

ChatAccum::apply_chat_chunk emitted a reasoning delta event immediately for every upstream chunk. There was no buffering or coalescing step, so the granularity of the provider's chunks directly determined the granularity of the UI update.

Fix

Buffer streaming reasoning in both response conversion paths and emit it in small blocks:

• Chat completions (ChatAccum): new reasoning_pending buffer. A response.reasoning_summary_text.delta is emitted only when the buffered text reaches REASONING_DELTA_FLUSH_CHARS (160) or contains a paragraph break (\n\n). Remaining text is flushed at finish().
• Native Responses (native_stream_to_responses): new NativeReasoningBuffer coalesces response.reasoning_summary_text.delta events using the same threshold, flushing before any non-reasoning event or stream termination. The buffer resets on a different item_id / summary_index.

Tests

• Updated chat_stream_reasoning_content_emits_reasoning_deltas and chat_stream_reasoning_field_emits_reasoning_deltas for the new buffered behavior.
• Updated chat_stream_debug_summary_counts_reasoning_without_text delta-event count.
• Added chat_stream_reasoning_coalesces_line_sized_deltas_until_a_small_block.
• Added chat_stream_reasoning_flushes_a_small_block_without_waiting_for_completion.
• Added chat_stream_reasoning_flushes_at_paragraph_boundaries.
• Added native_stream_reasoning_summary_deltas_are_coalesced.
• Added native_stream_reasoning_summary_deltas_flush_at_paragraphs.

Validation

• PATH=${RUSTUP_TOOLCHAIN}/bin:$PATH cargo test --locked response_codec::tests → 187 passed
• PATH=${RUSTUP_TOOLCHAIN}/bin:$PATH SOURCE_CHECKS_SKIP_TYPOS=1 bash scripts/source-checks.sh → ok (fmt, clippy, JS harness)
• PATH=${RUSTUP_TOOLCHAIN}/bin:$PATH cargo build --locked → ok
• Full cargo test --locked has 25 pre-existing sandbox-only failures from tests that bind local listeners (PermissionDenied). They are unrelated to this change.

Limitations / notes

• Sub-agent based fresh-context blind review searches could not be run because the environment does not expose a working spawn_agent tool. The primary review was performed manually.
• The threshold (160 chars) and paragraph-boundary flush are heuristic; they match the small blocks behavior seen with well-sized upstream responses.

---

Fixes the streaming thinking display issue for kimi-k2.7-code and other providers that emit tiny reasoning deltas.

Summary by Sourcery

Coalesce streamed reasoning into client-friendly blocks and make replayed tool-call history compatible with strict Chat Completions providers.

Bug Fixes:

• Coalesce tiny streaming reasoning fragments into small response blocks, flushing at size, paragraph, content, stream termination, and error boundaries.
• Preserve buffered reasoning before forwarding upstream failures or incomplete-stream events.
• Normalize replayed function-call arguments to valid JSON objects for Chat Completions providers.

Enhancements:

• Apply consistent reasoning buffering to both Chat Completions and native Responses stream conversion while preserving item and summary identities.

Tests:

• Add coverage for reasoning coalescing, paragraph and threshold flushing, keepalive handling, identity changes, stale metadata, and error preservation.
• Add transform coverage for truncated and non-object function-call history arguments.

[jatmn]

@sourcery-ai review

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• native_reasoning_summary_delta() uses expect() on the parsed JSON object, which will panic on malformed or unexpected upstream frames; consider handling this more defensively so a bad provider event doesn’t take down the stream.
• NativeReasoningBuffer::append() and take_flush() currently operate only on item_id and summary_index; if other identifying fields are added to reasoning events in the future, the buffer may coalesce incompatible deltas, so it may be worth centralising the identity check in a helper or documenting the assumptions more explicitly.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- native_reasoning_summary_delta() uses expect() on the parsed JSON object, which will panic on malformed or unexpected upstream frames; consider handling this more defensively so a bad provider event doesn’t take down the stream.
- NativeReasoningBuffer::append() and take_flush() currently operate only on item_id and summary_index; if other identifying fields are added to reasoning events in the future, the buffer may coalesce incompatible deltas, so it may be worth centralising the identity check in a helper or documenting the assumptions more explicitly.

## Individual Comments

### Comment 1
<location path="src/response_codec.rs" line_range="76-85" />
<code_context>
+        if should_flush { self.take_all() } else { None }
+    }
+
+    fn take_all(&mut self) -> Option<String> {
+        if self.pending.is_empty() {
+            return None;
+        }
+        let mut value = self.template.clone()?;
+        if let Some(delta) = value.get_mut("delta") {
+            *delta = Value::String(self.pending.clone());
+        }
+        self.pending.clear();
+        Some(sse("response.reasoning_summary_text.delta", value))
+    }
+}
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Retaining `template` after `take_all` can mis-detect identity changes on the next stream.

After `take_all`, `pending` is cleared but `template`, `item_id`, and `summary_index` remain set. On the next reasoning event, `append` will see `template.is_some()` and perform identity comparison against stale metadata, potentially triggering `identity_changed` and a flush even though `pending` is empty. Consider clearing `template` (and `item_id`/`summary_index` if applicable) in `take_all` so identity detection is based only on actual buffered content.

Suggested implementation:

```rust
        let text_only = self
            .pending
            .strip_prefix(REASONING_DISPLAY_HEADER)
            .unwrap_or(&self.pending);
        let should_flush = !text_only.is_empty()
            && (text_only.len() >= REASONING_DELTA_FLUSH_CHARS
                || text_only.contains(REASONING_DELTA_FLUSH_PARAGRAPH));
        if should_flush { self.take_all() } else { None }
    }

    fn take_all(&mut self) -> Option<String> {
        // When we finish a buffered stream, clear identity metadata so
        // future identity comparisons don't use stale template/item_id/summary_index.
        if self.pending.is_empty() {
            self.template = None;
            self.item_id = None;
            self.summary_index = None;
            return None;
        }

        // Take ownership of the template (and clear it) so it can't be reused
        // for subsequent streams.
        let mut value = self.template.take()?;
        if let Some(delta) = value.get_mut("delta") {
            *delta = Value::String(self.pending.clone());
        }

        // Clear all buffered state for the completed stream.
        self.pending.clear();
        self.item_id = None;
        self.summary_index = None;

        Some(sse("response.reasoning_summary_text.delta", value))
    }
}

impl NativeReasoningBuffer {

```

```rust
    fn append(&mut self, delta: &NativeReasoningSummaryDelta) -> Option<String> {
        let identity_changed = self.template.is_some()
            && (self.item_id.as_deref() != Some(delta.item_id.as_str())
                || self.summary_index != Some(delta.summary_index));
        let flushed = if identity_changed {
            // take_all now also clears template/item_id/summary_index so
            // identity detection for the next stream is based solely on new content.
            self.take_all()
        } else {
            None

```

If `item_id` or `summary_index` are not `Option` types, adjust the `None` assignments to the appropriate "cleared" value for your types (e.g. `String::new()` or a sentinel index). The core idea is that `take_all` must reset all identity-related fields (template, item_id, summary_index) whenever it completes or decides there's nothing to flush, so subsequent calls to `append` don't compare against stale metadata.
</issue_to_address>

### Comment 2
<location path="src/response_codec_tests.rs" line_range="311" />
<code_context>
+#[tokio::test]
+async fn native_stream_reasoning_summary_deltas_reset_on_different_item_id() {
</code_context>
<issue_to_address>
**suggestion (testing):** Add a companion native-stream test that verifies buffer reset on summary_index changes.

The current test only covers buffer reset when `item_id` changes. Please add a complementary test (e.g., `native_stream_reasoning_summary_deltas_reset_on_different_summary_index`) where `item_id` is constant but `summary_index` changes, and assert that you get two separate deltas (one per index) instead of a single coalesced block. This will protect the identity-based reset logic from regressions along the `summary_index` path.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[jatmn]

Review follow-up (c2756ed)

Addressed the Sourcery threads and one additional live failure found in local warp/session logs.

Sourcery

1. Native reasoning buffer no longer retains stale identity after take_all (clears template/item_id/summary_index).
2. Added summary_index identity-reset test.
3. Also made native reasoning template parsing non-panicking.

Additional live bug (not previously in PR scope)

• Error: messages.N.tool_calls.0.function.arguments: arguments must be a valid JSON object string
• Root cause: session history contained a truncated function call (exec_command args = {"cmd": "gh), and warp replayed it into Chat Completions history for Kimi/Concentrate. Providers that validate history reject the whole request.
• Fix: on responses→chat history conversion, sanitize function.arguments to a JSON object string (invalid/truncated -> {}, non-object JSON wrapped as {"value": ...}).

Validation: focused response_codec + transform tests green; source-checks/clippy green.

[jatmn]

@sourcery-ai review

[sourcery-ai]

Sorry @jatmn, you have reached your weekly rate limit of 500000 diff characters.

Please try again later or upgrade to continue using Sourcery