#!/usr/bin/env python
import asyncio
import base64
import copy
import hashlib
import json
import locale
import math
import mimetypes
import os
import platform
import re
import sys
import threading
import time
import traceback
import weakref
from collections import defaultdict
from datetime import date, datetime
import xxhash
# Optional dependency: used to convert locale codes (eg ``en_US``)
# into human-readable language names (eg ``English``).
try:
from babel import Locale # type: ignore
except ImportError: # Babel not installed we will fall back to a small mapping
Locale = None
from json.decoder import JSONDecodeError
from pathlib import Path
from typing import List
from urllib.parse import urlparse
from uuid import uuid4 as generate_unique_id
import cecli.prompts.utils.system as prompts
from cecli import __version__, models, urls, utils
from cecli.commands import Commands, SwitchCoderSignal
from cecli.decoding import safe_open
from cecli.exceptions import LiteLLMExceptions
from cecli.helpers import command_parser, command_queue, coroutines, nested, responses
from cecli.helpers.conversation import ConversationService, MessageTag
from cecli.helpers.file_system import FileSystemService
from cecli.helpers.io_proxy import IOProxy
from cecli.helpers.loop_detect import LoopDetectedError, LoopDetector
from cecli.helpers.memory_control import trim_memory
from cecli.helpers.observations.service import ObservationService
from cecli.helpers.profiler import TokenProfiler
from cecli.helpers.sessions import SessionManager
from cecli.helpers.threading import ThreadSafeEvent
from cecli.history import ChatSummary
from cecli.hooks import HookIntegration
from cecli.io import ConfirmGroup, InputOutput
from cecli.linter import Linter
from cecli.llm import litellm
from cecli.mcp import LocalServer
from cecli.reasoning_tags import (
REASONING_TAG,
format_reasoning_content,
remove_reasoning_content,
replace_reasoning_tags,
)
from cecli.repo import ANY_GIT_ERROR, GitRepoProxy
from cecli.repomap import RepoMap
from cecli.report import update_error_prefix
from cecli.run_cmd import run_cmd_async
from cecli.tools.utils.output import print_tool_response
from cecli.tools.utils.registry import ToolRegistry
from cecli.utils import copy_tool_call, format_tokens, is_image_file, safe_relpath
from ..dump import dump # noqa: F401
from ..prompts.utils.registry import PromptObject, PromptRegistry
GLOBAL_DATE = date.today().isoformat()
# Default per-minute token budget used for rate limiting when not configured.
DEFAULT_TOKENS_PER_MINUTE = 1000000
class UnknownEditFormat(ValueError):
def __init__(self, edit_format, valid_formats):
self.edit_format = edit_format
self.valid_formats = valid_formats
super().__init__(
f"Unknown edit format {edit_format}. Valid formats are: {', '.join(valid_formats)}"
)
class MissingAPIKeyError(ValueError):
pass
class FinishReasonLength(Exception):
pass
class EmptyResponseError(Exception):
pass
def wrap_fence(name):
return f"", f""
all_fences = [
("`" * 3, "`" * 3),
("`" * 4, "`" * 4), # LLMs ignore and revert to triple-backtick, causing #2879
wrap_fence("source"),
wrap_fence("code"),
wrap_fence("pre"),
wrap_fence("codeblock"),
wrap_fence("sourcecode"),
]
class UsageMeta(type):
"""Metaclass that provides shared accumulator properties across all Coder subclasses.
Every instance shares the same unified total token and cost amounts."""
_total_cost = 0
_total_tokens_sent = 0
_total_tokens_received = 0
_total_cached_tokens = 0
_token_usage_buffer = {}
_token_usage_window = 60.0
@property
def total_cost(cls):
return UsageMeta._total_cost
@total_cost.setter
def total_cost(cls, value):
UsageMeta._total_cost = value
@property
def total_tokens_sent(cls):
return UsageMeta._total_tokens_sent
@total_tokens_sent.setter
def total_tokens_sent(cls, value):
UsageMeta._total_tokens_sent = value
@property
def total_tokens_received(cls):
return UsageMeta._total_tokens_received
@total_tokens_received.setter
def total_tokens_received(cls, value):
UsageMeta._total_tokens_received = value
@property
def total_cached_tokens(cls):
return UsageMeta._total_cached_tokens
@total_cached_tokens.setter
def total_cached_tokens(cls, value):
UsageMeta._total_cached_tokens = value
@classmethod
def _purge_token_usage(cls, model, now=None):
"""Drop a model's token-usage entries older than the rolling window."""
if now is None:
now = time.time()
cutoff = now - UsageMeta._token_usage_window
entries = UsageMeta._token_usage_buffer.get(model, [])
UsageMeta._token_usage_buffer[model] = [
(tokens, ts) for tokens, ts in entries if ts >= cutoff
]
@classmethod
def _record_token_usage(cls, model, delta, now=None):
"""Record a model's request prompt-token usage as (tokens, timestamp)."""
if delta self.cost_multiplier * (
nested.getter(self.args, "cost_limit", float("inf")) or float("inf")
):
if await self.io.confirm_ask(
"You have reached your configured cost limit. Continue?",
group_response="Cost Limit",
explicit_yes_required=True,
):
Coder.cost_multiplier += 1
else:
return
async for _ in self.send_message(message):
pass
await self.hot_reload()
if not self.empty_response:
if not self.reflected_message:
await self.auto_save_session(force=True)
break
if self.num_reflections >= self.max_reflections:
self.io.tool_warning(
f"Only {self.max_reflections} reflections allowed, stopping."
)
break
self.num_reflections += 1
if self.tool_reflection:
self.num_reflections -= 1
if self.reflected_message is True:
message = None
else:
message = self.reflected_message
elif self.stop_on_empty:
await self.auto_save_session(force=True)
break
if self.enable_context_compaction:
await self.compact_context_if_needed()
if nested.getter(self, "agent_finished", False):
await self.auto_save_session(force=True)
break
await self.auto_save_session(force=True)
# Drain any queued prompts (CLI-33) only after the current message has
# fully completed, so the queue is processed within run_one() instead of
# being watched by the generation loops.
await self.process_prompt_queue(preproc)
if not await HookIntegration.call_end_hooks(self):
self.io.tool_warning("Execution stopped by end hook")
return
def _is_url_allowed(self, url):
allowed_domains = self.security_config.get("allowed-domains")
if not allowed_domains:
return True
parsed_url = urlparse(url)
domain = parsed_url.netloc.lower()
if not domain:
return False
for allowed in allowed_domains:
allowed = allowed.lower()
if domain == allowed or domain.endswith("." + allowed):
return True
return False
async def check_and_open_urls(self, exc, friendly_msg=None):
"""Check exception for URLs, offer to open in a browser, with user-friendly error msgs."""
text = str(exc)
if friendly_msg:
self.io.tool_warning(text)
self.io.tool_error(f"{friendly_msg}")
else:
self.io.tool_error(text)
# Exclude double quotes from the matched URL characters
url_pattern = re.compile(r'(https?://[^\s/$.?#].[^\s"]*)')
# Use set to remove duplicates
urls = list(set(url_pattern.findall(text)))
for url in urls:
url = url.rstrip(".',\"}") # Added } to the characters to strip
if self._is_url_allowed(url):
await self.io.offer_url(url)
return urls
async def check_for_urls(self, inp: str) -> List[str]:
"""Check input for URLs and offer to add them to the chat."""
if not self.detect_urls or (self.args and self.args.disable_scraping):
return inp
# Exclude double quotes from the matched URL characters
url_pattern = re.compile(r'(https?://[^\s/$.?#].[^\s"]*[^\s,.])')
# Use set to remove duplicates
urls = list(set(url_pattern.findall(inp)))
group = ConfirmGroup(urls)
for url in urls:
if url not in self.rejected_urls and self._is_url_allowed(url):
url = url.rstrip(".',\"")
if await self.io.confirm_ask(
"Add URL to the chat?",
subject=url,
group=group,
allow_never=True,
explicit_yes_required=not self.args.yes_always_commands,
):
inp += "\n\n"
inp += await self.commands.execute("web", url, return_content=True)
else:
self.rejected_urls.add(url)
return inp
def keyboard_interrupt(self):
# Ensure cursor is visible on exit
if not self.tui:
from rich.console import Console
Console().show_cursor(True)
self.io.tool_warning("^C KeyboardInterrupt")
self.interrupt_event.set()
self.last_keyboard_interrupt = time.time()
# Old summarization system removed - using context compaction logic instead
async def compact_context_if_needed(self, force=False, message=""):
if not self.enable_context_compaction:
return
# Trigger background observation/reflection check
await ObservationService.get_instance(self).check_and_trigger()
manager = ConversationService.get_manager(self)
done_messages = manager.get_messages_dict(MessageTag.DONE)
cur_messages = manager.get_messages_dict(MessageTag.CUR)
diff_messages = manager.get_messages_dict(MessageTag.DIFFS)
file_context_messages = manager.get_messages_dict(MessageTag.FILE_CONTEXTS)
all_messages = manager.get_messages_dict()
# Exclude first cur_message since that's the user's initial input
done_tokens = self.summarizer.count_tokens(done_messages)
cur_tokens = self.summarizer.count_tokens(cur_messages[1:] if len(cur_messages) > 1 else [])
diff_tokens = self.summarizer.count_tokens(diff_messages)
file_context_tokens = self.summarizer.count_tokens(file_context_messages)
all_tokens = self.summarizer.count_tokens(all_messages)
# Determine if compaction is worthwhile
compactable_tokens = done_tokens + cur_tokens + diff_tokens + file_context_tokens
# Condition 1: Is min size
is_min_size = all_tokens >= self.context_compaction_max_tokens
# Condition 2: Percentage check (is chat history a significant part of the context?)
is_worth_by_percentage = all_tokens > 0 and (compactable_tokens / all_tokens) >= 0.20
# Condition 3: Absolute savings check (would compacting save enough tokens to fit?)
tokens_over_limit = all_tokens - (self.context_compaction_max_tokens or all_tokens)
potential_savings = compactable_tokens * 0.90
is_worth_by_absolute_savings = (
tokens_over_limit > 0 and potential_savings >= tokens_over_limit
)
if not force:
if not is_min_size:
return
if not (is_worth_by_percentage or is_worth_by_absolute_savings):
self.io.tool_output(
"Skipping compaction: Not enough chat history to make a difference. Use /drop to remove files."
)
return
# Compaction can only shrink the conversation buckets counted above.
# If the fixed context (files added to the chat, repo map, system
# prompt, rules, tool schemas) already meets the limit on its own,
# summarizing cannot bring the context under the limit. Bail out with
# guidance instead of churning every turn, which is what a large file
# added to the chat otherwise causes.
non_compactable_tokens = all_tokens - compactable_tokens
if (
self.context_compaction_max_tokens
and non_compactable_tokens >= self.context_compaction_max_tokens
):
if getattr(self, "_compaction_noop_warning_time", 0) + 300 < time.time():
self._compaction_noop_warning_time = time.time()
self.io.tool_warning(
"Compaction cannot reduce the context: the fixed context (files in the"
f" chat, repo map, prompts) is already ~{non_compactable_tokens:,}"
f" tokens, at or above the ~{self.context_compaction_max_tokens:,} token"
" compaction limit."
)
self.io.tool_output("- Use /drop to remove unneeded files from the chat")
self.io.tool_output("- Use /clear to clear the chat history")
return
message_tokens = done_tokens + cur_tokens
file_tokens = diff_tokens + file_context_tokens
combined_tokens = done_tokens + cur_tokens + diff_tokens + file_context_tokens
self.context_compaction_current_ratio = all_tokens / self.context_compaction_max_tokens
if force or (
all_tokens >= self.context_compaction_max_tokens * 0.9
and ConversationService.get_chunks(self).last_clear_count > 20
and file_tokens / max(message_tokens, 1) > 2
):
manager.clear_tag(MessageTag.LINT, ratio=0.33)
manager.clear_tag(MessageTag.DIFFS, ratio=0.33)
manager.clear_tag(MessageTag.FILE_CONTEXTS, ratio=0.33)
ConversationService.get_files(self).clear_file_cache()
ConversationService.get_chunks(self).flush_removals()
ConversationService.get_chunks(self).reset_clear_count()
ObservationService.get_instance(self).reset_index()
if not force and combined_tokens < self.context_compaction_max_tokens:
return
if force:
self.io.tool_output("Forcing compaction of chat history...")
else:
self.io.tool_output("Compacting chat history to make room for new messages...")
self.io.update_spinner("Compacting...")
try:
compaction_prompt = self.gpt_prompts.compaction_prompt
if message:
compaction_prompt = f"{compaction_prompt}\n\n{message}"
async def summarize_and_update(messages, tag):
text = await self.summarizer.summarize_all_as_text(
messages,
compaction_prompt,
self.context_compaction_summary_tokens,
coder=self,
)
if not text:
raise ValueError(f"Summarization of {tag} messages returned empty.")
if ObservationService.get_instance(self).observations:
obs_text = "\n".join(ObservationService.get_instance(self).observations)
text = f"HISTORICAL OBSERVATIONS:\n{obs_text}\n\n{text}"
manager.clear_tag(tag)
if tag == MessageTag.DONE:
manager.queue_message(message_dict={"role": "user", "content": text}, tag=tag)
manager.queue_message(
message_dict={
"role": "assistant",
"content": (
"Ok, I will use this summary and the observations as context for"
" our conversation going forward."
),
},
tag=tag,
)
else:
if self.last_user_message:
manager.queue_message(
message_dict={"role": "user", "content": self.last_user_message},
tag=tag,
)
manager.queue_message(
message_dict={
"role": "assistant",
"content": "Ok. I am awaiting your summary of our goals to proceed.",
},
tag=tag,
force=True,
)
manager.queue_message(
message_dict={
"role": "user",
"content": (
"Here is a summary of our current goals and historical"
f" context:\n{text}"
),
},
tag=tag,
)
manager.queue_message(
message_dict={
"role": "assistant",
"content": (
"Ok, I will use this summary and proceed with our task. I will"
" first apply any changes in the summary and then continue"
" exploration as necessary."
),
},
tag=tag,
force=True,
)
latest_messages = []
for msg in reversed(messages):
latest_messages.append(msg)
if msg["role"] == "assistant":
break
for msg in reversed(latest_messages):
manager.add_message(msg, tag=tag)
# Fire memorizer after successful compaction
if self.auto_memory and self.edit_format not in ["subagent"]:
from cecli.helpers.memory.utils import invoke_memorizer
asyncio.create_task(invoke_memorizer(self, additional_context=text))
await self._rate_limit_sleep()
if done_tokens > self.context_compaction_max_tokens or done_tokens > cur_tokens:
await summarize_and_update(done_messages, MessageTag.DONE)
if cur_tokens > self.context_compaction_max_tokens or cur_tokens > done_tokens:
await summarize_and_update(cur_messages, MessageTag.CUR)
manager.clear_tag(MessageTag.DIFFS)
manager.clear_tag(MessageTag.FILE_CONTEXTS)
ConversationService.get_files(self).clear_file_cache()
ConversationService.get_chunks(self).flush_removals()
ConversationService.get_chunks(self).reset_clear_count()
ObservationService.get_instance(self).reset_index()
self.format_chat_chunks()
# Post-compaction token floor check
# Recalculate tokens after compaction to prevent infinite loops
# on already-minimal context
all_messages = manager.get_messages_dict()
post_compaction_tokens = self.summarizer.count_tokens(all_messages)
token_floor = (
self.context_compaction_max_tokens * 0.25
if self.context_compaction_max_tokens
else 0
)
if post_compaction_tokens < token_floor:
self.io.tool_output(
"...context is already at minimum size, cannot compact further."
)
else:
self.io.tool_output("...chat history compacted.")
self.io.update_spinner(self.io.last_spinner_text)
except Exception as e:
self.io.tool_warning(f"Context compaction failed: {e}")
self.io.tool_warning("Proceeding with full history for now.")
return
def normalize_language(self, lang_code):
"""
Convert a locale code such as ``en_US`` or ``fr`` into a readable
language name (e.g. ``English`` or ``French``). If Babel is
available it is used for reliable conversion; otherwise a small
built-in fallback map handles common languages.
"""
if not lang_code:
return None
if lang_code.upper() in ("C", "POSIX"):
return None
# Probably already a language name
if (
len(lang_code) > 3
and "_" not in lang_code
and "-" not in lang_code
and lang_code[0].isupper()
):
return lang_code
# Preferred: Babel
if Locale is not None:
try:
loc = Locale.parse(lang_code.replace("-", "_"))
return loc.get_display_name("en").capitalize()
except Exception:
pass # Fall back to manual mapping
# Simple fallback for common languages
fallback = {
"en": "English",
"fr": "French",
"es": "Spanish",
"de": "German",
"it": "Italian",
"pt": "Portuguese",
"zh": "Chinese",
"ja": "Japanese",
"ko": "Korean",
"ru": "Russian",
}
primary_lang_code = lang_code.replace("-", "_").split("_")[0].lower()
return fallback.get(primary_lang_code, lang_code)
def get_user_language(self):
"""
Detect the user's language preference and return a human-readable
language name such as ``English``. Detection order:
1. ``self.chat_language`` if explicitly set
2. ``locale.getlocale()``
3. ``LANG`` / ``LANGUAGE`` / ``LC_ALL`` / ``LC_MESSAGES`` environment variables
"""
# Explicit override
if self.chat_language:
return self.normalize_language(self.chat_language)
# System locale
try:
lang = locale.getlocale()[0]
if lang:
lang = self.normalize_language(lang)
if lang:
return lang
except Exception:
pass
# Environment variables
for env_var in ("LANG", "LANGUAGE", "LC_ALL", "LC_MESSAGES"):
lang = os.environ.get(env_var)
if lang:
lang = lang.split(".")[0] # Strip encoding if present
return self.normalize_language(lang)
return None
def get_platform_info(self):
platform_text = ""
try:
platform_text = f"- Platform: {platform.platform()}\n"
except KeyError:
# Skip platform info if it can't be retrieved
platform_text = "- Platform information unavailable\n"
shell_var = "COMSPEC" if os.name == "nt" else "SHELL"
shell_val = os.getenv(shell_var)
platform_text += f"- Shell: {shell_var}={shell_val}\n"
user_lang = self.get_user_language()
if user_lang:
platform_text += f"- Language: {user_lang}\n"
dt = datetime.now().astimezone().strftime("%Y-%m-%d")
platform_text += f"- Current date: {dt}\n"
if self.repo:
platform_text += "- The user is operating inside a git repository\n"
if self.lint_cmds:
if self.auto_lint:
platform_text += (
"- The user's pre-commit runs these lint commands, don't suggest running"
" them:\n"
)
else:
platform_text += "- The user prefers these lint commands:\n"
for lang, cmd in self.lint_cmds.items():
if lang is None:
platform_text += f" - {cmd}\n"
else:
platform_text += f" - {lang}: {cmd}\n"
if self.test_cmd:
if self.auto_test:
platform_text += (
"- The user's pre-commit runs this test command, don't suggest running them: "
)
else:
platform_text += "- The user prefers this test command: "
platform_text += self.test_cmd + "\n"
return platform_text
def fmt_system_prompt(self, prompt):
final_reminders = []
lazy_prompt = ""
if self.get_active_model().lazy:
lazy_prompt = self.gpt_prompts.lazy_prompt
final_reminders.append(lazy_prompt)
overeager_prompt = ""
if self.get_active_model().overeager:
overeager_prompt = self.gpt_prompts.overeager_prompt
final_reminders.append(overeager_prompt)
user_lang = self.get_user_language()
if user_lang:
final_reminders.append(f"Reply in {user_lang}.\n")
platform_text = self.get_platform_info()
if self.suggest_shell_commands:
shell_cmd_prompt = self.gpt_prompts.shell_cmd_prompt.format(platform=platform_text)
shell_cmd_reminder = self.gpt_prompts.shell_cmd_reminder.format(platform=platform_text)
rename_with_shell = self.gpt_prompts.rename_with_shell
else:
shell_cmd_prompt = self.gpt_prompts.no_shell_cmd_prompt.format(platform=platform_text)
shell_cmd_reminder = self.gpt_prompts.no_shell_cmd_reminder.format(
platform=platform_text
)
rename_with_shell = ""
if user_lang: # user_lang is the result of self.get_user_language()
language = user_lang
else:
# Default if no specific lang detected
language = "the same language they are using"
if self.fence[0] == "`" * 4:
quad_backtick_reminder = (
"\nIMPORTANT: Use *quadruple* backticks ```` as fences, not triple backticks!\n"
)
else:
quad_backtick_reminder = ""
if self.mcp_tools and len(self.mcp_tools) > 0:
final_reminders.append(self.gpt_prompts.tool_prompt)
final_reminders = "\n\n".join(final_reminders)
prompt = prompt.format(
fence=self.fence,
quad_backtick_reminder=quad_backtick_reminder,
final_reminders=final_reminders,
platform=platform_text,
shell_cmd_prompt=shell_cmd_prompt,
rename_with_shell=rename_with_shell,
shell_cmd_reminder=shell_cmd_reminder,
go_ahead_tip=self.gpt_prompts.go_ahead_tip,
language=language,
lazy_prompt=lazy_prompt,
overeager_prompt=overeager_prompt,
)
return prompt
def format_chat_chunks(self):
# Choose appropriate fence based on file content
self.choose_fence()
ConversationService.get_chunks(self).initialize_conversation_system()
# Clean up ConversationFiles and remove corresponding messages
ConversationService.get_chunks(self).cleanup_files()
# Add reminder message with list of readonly and editable files
ConversationService.get_chunks(self).add_file_list_reminder()
# Add system messages (system prompt, examples, reminder)
ConversationService.get_chunks(self).add_system_messages()
# Add rules messages
ConversationService.get_chunks(self).add_rules_messages()
# Add repository map messages
ConversationService.get_chunks(self).add_repo_map_messages()
# Add read-only file messages
ConversationService.get_chunks(self).add_readonly_files_messages()
# Add chat and edit file messages
ConversationService.get_chunks(self).add_chat_files_messages()
ConversationService.get_manager(self).flush_queue()
# Return formatted messages for LLM
return ConversationService.get_manager(self).get_messages_dict()
def format_messages(self):
chunks = self.format_chat_chunks()
return chunks
def warm_cache(self, chunks):
if not self.add_cache_headers:
return
if not self.num_cache_warming_pings:
return
if not self.ok_to_warm_cache:
return
delay = 5 * 60 - 5
delay = float(os.environ.get("CECLI_CACHE_KEEPALIVE_DELAY", delay))
self.next_cache_warm = time.time() + delay
self.warming_pings_left = self.num_cache_warming_pings
self.cache_warming_chunks = chunks
if self.cache_warming_thread:
return
def warm_cache_worker():
while self.ok_to_warm_cache:
time.sleep(1)
if self.warming_pings_left = max_input_tokens:
if (
self.enable_context_compaction
and input_tokens >= self.context_compaction_max_tokens * 0.95
):
self.io.tool_output(
f"Estimated chat context of {input_tokens:,} tokens exceeds the"
f" {max_input_tokens:,} token limit. Attempting to compact..."
)
await self.compact_context_if_needed()
# After compaction, re-format messages and re-check tokens
messages = self.format_messages()
input_tokens = self.get_active_model().token_count(messages)
if max_input_tokens and input_tokens >= max_input_tokens:
if not hasattr(self, "_last_compaction_warning_time"):
self._last_compaction_warning_time = time.time()
if getattr(self, "_last_compaction_warning_time", 0) + 300 < time.time():
self._last_compaction_warning_time = time.time()
self.io.tool_error(
f"Your estimated chat context of {input_tokens:,} tokens still exceeds the"
f" {max_input_tokens:,} token limit for {self.get_active_model().name}!"
)
self.io.tool_output("To reduce the chat context:")
self.io.tool_output("- Use /drop to remove unneeded files from the chat")
self.io.tool_output("- Use /clear to clear the chat history")
self.io.tool_output("- Break your code into smaller files")
self.io.tool_output(
"It's probably safe to try and send the request, most providers won't charge if"
" the context limit is exceeded."
)
if not await self.io.confirm_ask("Try to proceed anyway?"):
self._last_compaction_warning_time = 0
return None
else:
self._last_compaction_warning_time = 0
return messages
def get_active_model(self):
return self.main_model
def empty_llm_tool_warning(self) -> str:
"""Ollama-friendly copy for local models; cloud hint otherwise."""
name = str(getattr(getattr(self, "main_model", None), "name", "") or "")
if "ollama" in name.lower():
return (
"Empty response from the local model (Ollama). "
"The model may have timed out, unloaded, or hit context limits."
)
return "Empty response received from LLM. Check API keys, quota, or provider status."
async def send_message(self, inp):
# Notify IO that LLM processing is starting
self.io.llm_started()
ConversationService.get_manager(self).flush_queue()
# Clear any stale interrupt state before starting formatting
# to avoid immediately re-catching a previous interrupt
self.interrupt_event.clear()
if inp:
# Make sure current coder actually has control of conversation system
ConversationService.get_chunks(self).initialize_conversation_system()
self.format_chat_chunks()
# Always add user message to conversation manager
ConversationService.get_manager(self).queue_message(
message_dict=dict(role="user", content=inp),
tag=MessageTag.CUR,
hash_key=(
"user_message",
xxhash.xxh3_128_hexdigest(inp.encode("utf-8", errors="replace")),
str(time.monotonic_ns()),
),
)
ConversationService.get_manager(self).decrement_message_markers()
import asyncio
loop = asyncio.get_running_loop()
async def format_in_executor():
return await loop.run_in_executor(None, self.format_messages)
result, interrupted = await self.coroutines.interruptible(
format_in_executor(), self.interrupt_event
)
if interrupted:
# Use CancelledError instead of KeyboardInterrupt to avoid
# propagating through the asyncio event loop during cleanup.
# KeyboardInterrupt is re-raised by Task.__step and bypasses
# asyncio.gather(return_exceptions=True), causing crashes
# when tasks are gathered during _cleanup_loop.
raise asyncio.CancelledError("Interrupted during message formatting")
messages = result
messages = await self.check_tokens(messages)
if not messages:
return
if self.verbose:
utils.show_messages(messages, functions=self.functions)
self.multi_response_content = ""
if self.show_pretty():
spinner_text = f"Waiting for {self.get_active_model().name}"
if not self.tui:
spinner_text += f" ${self.format_cost(self.total_cost)} session"
if self.io.spinner_active:
self.io.start_spinner(spinner_text, coder_uuid=getattr(self, "uuid", None))
else:
self.message_cost_deferred = spinner_text
if self.stream:
self.mdstream = True
else:
self.mdstream = None
else:
self.mdstream = None
retry_delay = 0.125
litellm_ex = LiteLLMExceptions()
self.usage_report = None
exhausted = False
interrupted = False
try:
while True:
try:
async for chunk in self.send(messages, tools=self.get_tool_list()):
yield chunk
break
except EmptyResponseError:
self.io.tool_warning(self.empty_llm_tool_warning())
retry_config = models.parse_retry_config(self.get_active_model().retries)
retry_on_empty = retry_config["retry_on_empty"]
if not retry_on_empty:
break
retry_delay *= retry_config["retry_backoff_factor"]
if retry_delay > retry_config["retry_timeout"]:
self.io.tool_error("Retry timeout exceeded on empty response.")
break
self.io.tool_output(f"Retrying in {retry_delay:.1f} seconds...")
_res, interrupted_sleep = await coroutines.interruptible(
asyncio.sleep(retry_delay), self.interrupt_event
)
if interrupted_sleep:
interrupted = True
break
continue
except litellm_ex.exceptions_tuple() as err:
ex_info = litellm_ex.get_ex_info(err)
if ex_info.name == "ContextWindowExceededError":
exhausted = True
break
retry_config = models.parse_retry_config(self.get_active_model().retries)
should_retry, _ = models.parse_model_error(retry_config, err)
if should_retry:
retry_delay *= retry_config["retry_backoff_factor"]
if retry_delay > retry_config["retry_timeout"]:
should_retry = False
if not should_retry:
self.mdstream = None
await self.check_and_open_urls(err, ex_info.description)
break
err_msg = str(err)
if ex_info.description:
self.io.tool_warning(err_msg)
self.io.tool_error(ex_info.description)
else:
self.io.tool_error(err_msg)
self.io.tool_output(f"Retrying in {retry_delay:.1f} seconds...")
_res, interrupted_sleep = await coroutines.interruptible(
asyncio.sleep(retry_delay), self.interrupt_event
)
if interrupted_sleep:
interrupted = True
break
continue
except (KeyboardInterrupt, asyncio.CancelledError):
interrupted = True
break
except FinishReasonLength:
# We hit the output limit!
if not self.get_active_model().info.get("supports_assistant_prefill"):
exhausted = True
break
self.multi_response_content = self.get_multi_response_content_in_progress()
if messages[-1]["role"] == "assistant":
messages[-1]["content"] = self.multi_response_content
else:
messages.append(
dict(role="assistant", content=self.multi_response_content, prefix=True)
)
except Exception as err:
self.error_code = 1
self.mdstream = None
lines = traceback.format_exception(type(err), err, err.__traceback__)
self.io.tool_warning("".join(lines))
self.io.tool_error(str(err))
return
finally:
if self.mdstream:
content_to_show = (
"" if self.tui and self.tui() else self.live_incremental_response(True)
)
self.stream_wrapper(content_to_show, final=True)
self.mdstream = None
# Ensure any waiting spinner is stopped
self.io.start_spinner("Processing Answer...", coder_uuid=getattr(self, "uuid", None))
if not self.io.spinner_active:
self.partial_response_content = self.get_multi_response_content_in_progress(True)
self.remove_reasoning_content()
self.multi_response_content = ""
self.io.tool_output()
self.show_usage_report()
await self.add_assistant_reply_to_cur_messages()
if exhausted:
cur_messages = ConversationService.get_manager(self).get_messages_dict(MessageTag.CUR)
if cur_messages and cur_messages[-1]["role"] == "user":
# Always add to conversation manager
ConversationService.get_manager(self).add_message(
message_dict=dict(
role="assistant",
content="FinishReasonLength exception: you sent too many tokens",
),
tag=MessageTag.CUR,
force=True,
promotion=ConversationService.get_manager(self).DEFAULT_TAG_PROMOTION_VALUE,
mark_for_demotion=1,
)
await self.show_exhausted_error()
self.num_exhausted_context_windows += 1
self._release_response_buffers()
return
if self.partial_response_function_call:
args = self.parse_partial_args()
if args:
content = args.get("explanation") or ""
else:
content = ""
elif self.partial_response_content:
content = self.partial_response_content
else:
content = ""
if interrupted:
# Always add to conversation manager
ConversationService.get_manager(self).add_message(
message_dict=dict(role="user", content="^C KeyboardInterrupt"),
tag=MessageTag.CUR,
force=True,
promotion=ConversationService.get_manager(self).DEFAULT_TAG_PROMOTION_VALUE,
mark_for_demotion=1,
)
# Always add assistant response to conversation manager
ConversationService.get_manager(self).add_message(
message_dict=dict(
role="assistant", content="I see that you interrupted my previous reply."
),
tag=MessageTag.CUR,
force=True,
promotion=ConversationService.get_manager(self).DEFAULT_TAG_PROMOTION_VALUE,
mark_for_demotion=1,
)
# The reply was interrupted mid-stream; drop the partial chunk buffers
# rather than holding them until the next send().
self._release_response_buffers()
return
edited = await self.apply_updates()
# Run tests before committing so failing tests abort the commit and
# reflect the errors back for the model to fix first.
if edited and self.auto_test and self.test_cmd:
test_errors = await self.commands.execute("test", self.test_cmd)
self.test_outcome = not test_errors
if test_errors:
ok = await self.io.confirm_ask("Attempt to fix test errors?")
if ok:
self.reflected_message = test_errors
return
if edited:
self.coder_edited_files.update(edited)
saved_message = await self.auto_commit(edited)
if not saved_message and hasattr(self.gpt_prompts, "files_content_gpt_edits_no_repo"):
saved_message = self.gpt_prompts.files_content_gpt_edits_no_repo
if not interrupted:
add_rel_files_message = await self.check_for_file_mentions(content)
if add_rel_files_message:
if self.reflected_message:
self.reflected_message += "\n\n" + add_rel_files_message
else:
self.reflected_message = add_rel_files_message
return
# Process any tools using MCP servers
try:
if self.partial_response_tool_calls:
tool_call_response, a, b = self.consolidate_chunks()
if await self.process_tool_calls(tool_call_response):
self.num_tool_calls += 1
self.reflected_message = self.reflected_message or True
return
except Exception as e:
self.io.tool_error(f"Error processing tool calls: {str(e)}")
self.io.tool_error(traceback.format_exc())
self.reflected_message = True
return
# Continue without tool processing
self.num_tool_calls = 0
try:
if await self.reply_completed():
return
except KeyboardInterrupt:
interrupted = True
if self.reflected_message:
return
if edited and self.auto_lint:
lint_errors = await self.lint_edited(edited)
if lint_errors is None: # Interrupted
return
await self.auto_commit(edited, context="Ran the linter")
self.lint_outcome = not lint_errors
if lint_errors:
ok = await self.io.confirm_ask("Attempt to fix lint errors?")
if ok:
self.reflected_message = lint_errors
return
shared_output = await self.run_shell_commands()
if shared_output:
ConversationService.get_manager(self).add_message(
message_dict=dict(role="user", content=shared_output),
tag=MessageTag.CUR,
force=True, # Force update existing message
promotion=ConversationService.get_manager(self).DEFAULT_TAG_PROMOTION_VALUE,
mark_for_demotion=1,
)
ConversationService.get_manager(self).add_message(
message_dict=dict(role="assistant", content="Ok"),
tag=MessageTag.CUR,
force=True, # Force update existing message
promotion=ConversationService.get_manager(self).DEFAULT_TAG_PROMOTION_VALUE,
mark_for_demotion=1,
)
# Turn complete: drop the per-turn LLM stream buffers. They are reset at
# the start of the next send(), so holding on to them while idle only
# wastes memory (chunks can be large for long streaming responses).
self._release_response_buffers()
def _extract_and_prepare_tool_calls(self, tool_call_response):
"""
Unified extraction and preparation of tool calls.
Returns: list of prepared tool calls
"""
# 1. Use partial_response_tool_calls if available
if self.partial_response_tool_calls:
tool_calls = self.partial_response_tool_calls
# 2. Extract from tool_call_response
elif tool_call_response is not None:
tool_calls = self._extract_from_response(tool_call_response)
else:
return []
# 3. Expand concatenated JSON
return self._expand_concatenated_json(tool_calls)
def _extract_from_response(self, response):
"""Extract tool calls from various response formats."""
original_tool_calls = []
try:
if hasattr(response, "choices") and response.choices:
message = response.choices[0].message
if hasattr(message, "tool_calls") and message.tool_calls:
original_tool_calls = message.tool_calls
except (AttributeError, IndexError):
pass
return original_tool_calls
def _expand_concatenated_json(self, tool_calls):
"""Expand concatenated JSON arguments."""
expanded_tool_calls = []
for tool_call in tool_calls:
args_string = tool_call.function.arguments.strip()
# If there are no arguments, or it's not a string that looks like it could
# be concatenated JSON, just add it and continue.
if not args_string or not (args_string.startswith("{") or args_string.startswith("[")):
expanded_tool_calls.append(tool_call)
continue
json_chunks = utils.split_concatenated_json(args_string)
# If it's just a single JSON object, there's nothing to expand.
if len(json_chunks) = max_compaction_retries:
self.io.tool_error(
f"Context window exceeded after {max_compaction_retries}"
" compaction attempt(s)."
)
raise err
compaction_retry_count += 1
self.io.tool_error(
f"Compacting context... retry {compaction_retry_count}"
f"/{max_compaction_retries}"
)
try:
await self.compact_context_if_needed(
force=True,
message="Context window exceeded, please summarize to retry.",
)
except Exception:
self.io.tool_error(
"Context compaction failed." " Please use /clear or /compact manually."
)
raise err
# Re-format messages after compaction and retry
messages = self.format_messages()
continue
self.chat_completion_call_hashes.append(hash_object.hexdigest())
if not isinstance(completion, ModelResponse):
async for chunk in self.show_send_output_stream(completion):
yield chunk
else:
await self.show_send_output(completion)
if self.empty_response:
raise EmptyResponseError
response, func_err, content_err = self.consolidate_chunks()
if response:
completion = response
# Calculate costs for successful responses
self.calculate_and_show_tokens_and_cost(messages, completion, model=model)
except litellm_ex.exceptions_tuple() as err:
self.error_code = 1
ex_info = litellm_ex.get_ex_info(err)
if ex_info.name == "ContextWindowExceededError":
# Still calculate costs for context window errors
self.token_profiler.on_error()
self.calculate_and_show_tokens_and_cost(messages, completion, model=model)
raise
except (KeyboardInterrupt, asyncio.CancelledError) as kbi:
self.error_code = 130 # apparently standard?
self.keyboard_interrupt()
raise kbi
finally:
self.preprocess_response()
if self.partial_response_content:
self.io.ai_output(self.partial_response_content)
elif self.partial_response_function_call:
# TODO: push this into subclasses
args = self.parse_partial_args()
if args:
self.io.ai_output(json.dumps(args, indent=4))
async def show_send_output(self, completion):
ModelResponse = litellm.types.utils.ModelResponse
if self.verbose:
print(completion)
if not isinstance(completion, ModelResponse):
self.io.tool_error(str(completion))
return
if not completion.choices:
self.io.tool_error(str(completion))
return
self.partial_response_chunks.append(completion)
response, func_err, content_err = self.consolidate_chunks()
if not await HookIntegration.call_on_message_hooks(self, self.partial_response_content):
self.io.tool_warning("Execution stopped by on message hook")
return
resp_hash = dict(
function_call=str(self.partial_response_function_call),
content=self.partial_response_content,
)
resp_hash = hashlib.sha1(json.dumps(resp_hash, sort_keys=True).encode())
self.chat_completion_response_hashes.append(resp_hash.hexdigest())
if func_err and content_err:
self.io.tool_error(func_err)
self.io.tool_error(content_err)
raise Exception("No data found in LLM response!")
show_resp = self.render_incremental_response(True)
if self.partial_response_reasoning_content:
if nested.getter(self, "args.show_thinking"):
formatted_reasoning = format_reasoning_content(
self.partial_response_reasoning_content, self.reasoning_tag_name
)
show_resp = formatted_reasoning + show_resp
if len(self.partial_response_tool_calls):
self.tool_reflection = True
if nested.getter(self, "args.show_thinking"):
show_resp = replace_reasoning_tags(show_resp, self.reasoning_tag_name)
if (
not len(self.partial_response_content)
and not len(self.partial_response_tool_calls)
and not _is_meaningful_reasoning(self.partial_response_reasoning_content)
):
self.empty_response = True
return
self.io.assistant_output(show_resp, pretty=self.show_pretty())
if (
self.edit_format == "agent"
and self.stream
and not show_resp
and nested.getter(self, "_has_empty_reflected")
):
await asyncio.sleep(4)
self._has_empty_reflected = True
self.reflected_message = True
self.empty_response = True
else:
self._has_empty_reflected = False
if (
hasattr(completion.choices[0], "finish_reason")
and completion.choices[0].finish_reason == "length"
):
raise FinishReasonLength()
async def show_send_output_stream(self, completion):
received_content = False
chunk_index = 0
content_detector = LoopDetector()
tool_detector = LoopDetector()
loop_detected = False
stream = coroutines.interruptible_async_generator(completion, self.interrupt_event)
try:
async for chunk in stream:
if self.args.debug:
with safe_open(".cecli/logs/chunks.log", "a") as f:
print(chunk, file=f)
# Check if confirmation is in progress and wait if needed
if not self.io.confirmation_in_progress_event.is_set():
await self.io.confirmation_in_progress_event.wait()
if isinstance(chunk, str):
self.io.tool_error(chunk)
continue
else:
if len(chunk.choices) == 0:
continue
if (
hasattr(chunk.choices[0], "finish_reason")
and chunk.choices[0].finish_reason == "length"
):
raise FinishReasonLength()
try:
if chunk.choices[0].delta.tool_calls:
received_content = True
self.token_profiler.on_token()
for tool_call_chunk in chunk.choices[0].delta.tool_calls:
self.tool_reflection = True
if tool_call_chunk.type:
self.io.update_spinner_suffix(tool_call_chunk.type)
if tool_call_chunk.function:
if tool_call_chunk.function.name:
self.io.update_spinner_suffix(tool_call_chunk.function.name)
if tool_call_chunk.function.arguments:
self.io.update_spinner_suffix(
tool_call_chunk.function.arguments
)
tool_detector.push(tool_call_chunk.function.arguments)
except (AttributeError, IndexError):
# Handle cases where the response structure doesn't match expectations
pass
try:
func = chunk.choices[0].delta.function_call
# dump(func)
if func:
for k, v in func.items():
self.tool_reflection = True
self.io.update_spinner_suffix(v)
received_content = True
self.token_profiler.on_token()
except AttributeError:
pass
text = ""
try:
reasoning_content = chunk.choices[0].delta.reasoning_content
except AttributeError:
try:
reasoning_content = chunk.choices[0].delta.reasoning
except AttributeError:
reasoning_content = None
try:
content = chunk.choices[0].delta.content
if content:
if self.got_reasoning_content and not self.ended_reasoning_content:
text += f"\n\n\n\n"
self.ended_reasoning_content = True
text += content
received_content = True
self.token_profiler.on_token()
self.io.update_spinner_suffix(content)
except AttributeError:
pass
if reasoning_content:
if (
nested.getter(self.args, "show_thinking")
and not self.ended_reasoning_content
):
if not self.got_reasoning_content:
text += f"\n\n"
text += reasoning_content
self.got_reasoning_content = True
if _is_meaningful_reasoning(reasoning_content):
received_content = True
self.token_profiler.on_token()
self.io.update_spinner_suffix(reasoning_content)
self.partial_response_reasoning_content += reasoning_content
self.partial_response_content += text
chunk_index += 1
chunk._hidden_params["created_at"] = chunk_index
self.partial_response_chunks.append(chunk)
if text:
content_detector.push(text)
if self.show_pretty():
# Use simplified streaming - just call the method with full content
content_to_show = self.live_incremental_response(False)
self.stream_wrapper(content_to_show, final=False)
elif text:
# Apply reasoning tag formatting for non-pretty output
if nested.getter(self.args, "show_thinking"):
text = replace_reasoning_tags(text, self.reasoning_tag_name)
try:
self.stream_wrapper(text, final=False)
except UnicodeEncodeError:
# Safely encode and decode the text
safe_text = text.encode(
sys.stdout.encoding, errors="backslashreplace"
).decode(sys.stdout.encoding)
self.stream_wrapper(safe_text, final=False)
yield text
except (asyncio.CancelledError, KeyboardInterrupt):
raise KeyboardInterrupt
except LoopDetectedError as e:
self._output_loop_detected = True
self._output_loop_message = str(e)
loop_detected = True
if loop_detected:
self.io.tool_warning(
f"Output loop detected while streaming: {self._output_loop_message}"
)
# Explicitly close the async generators so the wrapper's interrupt
# task and the underlying provider generator are cleaned up instead
# of being left suspended after we stop consuming them.
if hasattr(stream, "aclose"):
try:
await stream.aclose()
except Exception:
pass
if hasattr(completion, "aclose"):
try:
await completion.aclose()
except Exception:
pass
return
if (
self.show_pretty()
and nested.getter(self.args, "show_thinking")
and self.got_reasoning_content
and not self.ended_reasoning_content
):
self.partial_response_content += f"\n\n\n\n"
content_to_show = self.live_incremental_response(False)
self.stream_wrapper(content_to_show, final=False)
# The Part Doing the Heavy Lifting Now
self.consolidate_chunks()
if not await HookIntegration.call_on_message_hooks(self, self.partial_response_content):
self.io.tool_warning("Execution stopped by on message hook")
return
# Treat the response as empty when nothing was received, or when the
# only thing received was reasoning made entirely of non-alphanumeric
# characters (e.g. moonshotai/kimi-k3 returning "!!!!").
if (
not received_content
and len(self.partial_response_tool_calls) == 0
and not _is_meaningful_reasoning(self.partial_response_reasoning_content)
):
self.empty_response = True
return
def consolidate_chunks(self):
if self.partial_response_consolidated:
return self.partial_response_consolidated
response = (
self.partial_response_chunks[0]
if not self.stream
else litellm.stream_chunk_builder(self.partial_response_chunks)
)
func_err = None
content_err = None
if len(self.partial_response_chunks):
last_chunk = self.partial_response_chunks[len(self.partial_response_chunks) - 1]
if last_chunk:
if getattr(last_chunk, "usage", None):
response.usage = last_chunk.usage
# Collect message-level provider-specific fields (e.g. `reasoning_items`
# for reasoning models) from ALL chunks. litellm's stream_chunk_builder()
# merges these with last-wins semantics for list fields, silently dropping
# every reasoning item except the final one. Reasoning models depend on the
# full ordered item list being present in the assistant message so that
# exact-prefix prompt caching keeps working across turns, so we collect the
# fields ourselves and concatenate list-valued entries.
message_provider_specific_fields = {}
for chunk in self.partial_response_chunks:
try:
if chunk.choices and chunk.choices[0].delta:
psf = getattr(chunk.choices[0].delta, "provider_specific_fields", None)
if psf and isinstance(psf, dict):
for key, value in psf.items():
if isinstance(value, list):
message_provider_specific_fields.setdefault(key, []).extend(value)
elif (
key in message_provider_specific_fields
and isinstance(message_provider_specific_fields[key], dict)
and isinstance(value, dict)
):
# Merge dict-valued metadata (e.g. gemini per-call
# function_call_signatures) so parallel tool calls
# each keep their own signature across chunks.
message_provider_specific_fields[key].update(value)
elif value is not None:
message_provider_specific_fields[key] = value
except (AttributeError, IndexError):
continue
if message_provider_specific_fields:
message_psf = getattr(response.choices[0].message, "provider_specific_fields", None)
if not isinstance(message_psf, dict):
message_psf = {}
message_psf.update(message_provider_specific_fields)
response.choices[0].message.provider_specific_fields = message_psf
try:
message_tool_calls = response.choices[0].message.tool_calls
if message_tool_calls and len(message_tool_calls):
if self.stream:
built_tool_calls = self._build_tool_calls_from_chunks()
if built_tool_calls:
response.choices[0].message.tool_calls = built_tool_calls
self.partial_response_tool_calls = built_tool_calls
else:
# Fall back to litellm's merged list, keeping every call
self.partial_response_tool_calls = list(message_tool_calls)
else:
# Non-streaming: the single response chunk already carries the
# full tool_calls list
self.partial_response_tool_calls = list(message_tool_calls)
self.partial_response_function_call = self.partial_response_tool_calls[0].function
except AttributeError as e:
func_err = e
try:
reasoning_content = response.choices[0].message.reasoning_content
except AttributeError:
try:
reasoning_content = response.choices[0].message.reasoning
except AttributeError:
reasoning_content = None
self.partial_response_reasoning_content = reasoning_content or ""
try:
content = response.choices[0].message.content
if isinstance(content, list):
# OpenAI-compatible APIs sometimes return content as a list
# of blocks; join the textual pieces for display.
content = "".join(
block.get("text", "")
for block in content
if isinstance(block, dict) and block.get("type") == "output_text"
) or "".join(
block.get("text", "")
for block in content
if isinstance(block, dict) and block.get("type") == "text"
)
self.partial_response_content = content or ""
except AttributeError as e:
content_err = e
# If no native tool calls, check if the content contains JSON tool calls
# This handles models that write JSON in text instead of using native calling
if not self.partial_response_tool_calls and self.partial_response_content:
extracted_calls = responses.extract_tools_from_content_json(
self.partial_response_content
)
if not extracted_calls:
extracted_calls = responses.extract_tools_from_content_xml(
self.partial_response_content
)
if not extracted_calls:
extracted_calls = responses.extract_tools_from_pseudo_json(
self.partial_response_content
)
if extracted_calls:
self.tool_reflection = True
self.partial_response_tool_calls = extracted_calls
if self._output_loop_detected:
# A repeating output loop was caught while streaming; turn it into an
# assistant message so the model can adjust and drop any tool calls.
marker = "\n\n[SYSTEM CANCEL: OUTPUT LOOP DETECTED]\n"
self.partial_response_content += marker
self.partial_response_tool_calls = []
self.partial_response_function_call = dict()
# The assistant message stored in the conversation is built from the
# response object (via model_dump()), so the marker has to be written
# there too, otherwise it never reaches the model to react to.
message = response.choices[0].message
message.content = (message.content or "") + marker
message.tool_calls = []
if hasattr(message, "function_call"):
message.function_call = None
self.partial_response_consolidated = (response, func_err, content_err)
return response, func_err, content_err
def _build_tool_calls_from_chunks(self):
"""Rebuild tool calls from the raw streaming chunks.
Parallel tool calls arrive interleaved and may start at any index.
Most providers key fragments by a per-call ``index`` (openai / anthropic /
gemini), but some (e.g. deepseek) reuse index0 for every call and only
distinguish them by the ``id`` announced on the first fragment. Keying
by id when present -- and remembering the index -> key mapping so later
id-less fragments resolve to the right call -- preserves every parallel
call instead of collapsing them onto one, keeps them ordered by first
appearance, and retains provider-specific fields (e.g. thought
signatures) attached.
"""
ChatCompletionMessageToolCall = litellm.types.utils.ChatCompletionMessageToolCall
Function = litellm.types.utils.Function
tool_calls_dict = {}
index_lookup = {}
last_key = None
for chunk in self.partial_response_chunks:
try:
if not (chunk.choices and chunk.choices[0].delta):
continue
delta = chunk.choices[0].delta
for tool_call in delta.tool_calls or []:
if tool_call is None:
continue
if nested.getter(tool_call, "function") is None:
continue
tool_id = nested.getter(tool_call, "id") or ""
index = nested.getter(tool_call, "index")
if tool_id:
key = ("id", tool_id)
if index is not None:
index_lookup[index] = key
last_key = key
elif index is not None and index in index_lookup:
key = index_lookup[index]
elif index is not None:
key = ("index", index)
index_lookup[index] = key
elif last_key is not None:
key = last_key
else:
key = ("slot", len(tool_calls_dict))
entry = tool_calls_dict.setdefault(
key,
{
"id": None,
"name": None,
"type": "function",
"arguments": [],
"provider_specific_fields": {},
"_order": len(tool_calls_dict),
},
)
entry["id"] = tool_id or entry["id"]
entry["type"] = nested.getter(tool_call, "type") or entry["type"]
entry["name"] = nested.getter(tool_call, "function.name") or entry["name"]
arguments = nested.getter(tool_call, "function.arguments")
if arguments:
entry["arguments"].append(arguments)
psf = nested.getter(tool_call, "provider_specific_fields")
if not psf:
psf = nested.getter(tool_call, "function.provider_specific_fields")
if psf and isinstance(psf, dict):
entry["provider_specific_fields"].update(psf)
except (AttributeError, IndexError):
continue
tool_calls = []
for key in sorted(tool_calls_dict.keys(), key=lambda k: tool_calls_dict[k]["_order"]):
data = tool_calls_dict[key]
if not (data["id"] and data["name"]):
continue
function = Function(
arguments="".join(data["arguments"]) or "{}",
name=data["name"],
)
params = {
"id": data["id"],
"function": function,
"type": data["type"] or "function",
}
if data["provider_specific_fields"]:
params["provider_specific_fields"] = data["provider_specific_fields"]
tool_calls.append(ChatCompletionMessageToolCall(**params))
return tool_calls
def stream_wrapper(self, content, final):
if not hasattr(self, "_streaming_buffer_length"):
self._streaming_buffer_length = 0
if final:
content += "\n\n"
if isinstance(content, str):
self._streaming_buffer_length += len(content)
self.io.stream_output(content, final=final)
if final:
self._streaming_buffer_length = 0
def live_incremental_response(self, final):
show_resp = self.render_incremental_response(final)
# Apply any reasoning tag formatting
if nested.getter(self.args, "show_thinking"):
show_resp = replace_reasoning_tags(show_resp, self.reasoning_tag_name)
# Track streaming state to avoid repetitive output
if not hasattr(self, "_streaming_buffer_length"):
self._streaming_buffer_length = 0
# Only send new content that hasn't been streamed yet
if len(show_resp) >= self._streaming_buffer_length:
new_content = show_resp[self._streaming_buffer_length :]
return new_content
else:
self._streaming_buffer_length = 0
self.io.reset_streaming_response()
return show_resp
def render_incremental_response(self, final):
# Just return the current content - the streaming logic will handle incremental updates
return self.get_multi_response_content_in_progress()
def preprocess_response(self):
if len(self.partial_response_tool_calls):
tool_list = []
tool_id_set = set()
for tool_call in self.partial_response_tool_calls:
# Handle both dictionary and object tool calls
if isinstance(tool_call, dict):
tool_id = tool_call.get("id")
else:
tool_id = getattr(tool_call, "id", None)
# LLM APIs sometimes return duplicates and that's annoying part 2
if tool_id in tool_id_set:
continue
tool_id_set.add(tool_id)
tool_list.append(tool_call)
self.partial_response_tool_calls = tool_list
def remove_reasoning_content(self):
"""Remove reasoning content from the model's response."""
self.partial_response_content = remove_reasoning_content(
self.partial_response_content,
self.reasoning_tag_name,
)
def calculate_and_show_tokens_and_cost(self, messages, completion=None, model=None):
active_model = model or self.get_active_model()
prompt_tokens = 0
completion_tokens = 0
cache_hit_tokens = 0
cache_write_tokens = 0
usage = nested.getter(completion, "usage") if completion else None
if (
usage is not None
and nested.getter(usage, ["prompt_tokens", "input_tokens"]) is not None
and nested.getter(usage, ["completion_tokens", "output_tokens"]) is not None
):
prompt_tokens = (
nested.getter(usage, ["prompt_tokens", "input_tokens", "prompt_eval_count"], 0) or 0
)
completion_tokens = (
nested.getter(usage, ["completion_tokens", "output_tokens", "eval_count"], 0) or 0
)
cache_hit_tokens = _first_usage_tokens(
usage,
[
"prompt_cache_hit_tokens",
"cache_read_input_tokens",
"input_tokens_details.cached_tokens",
"prompt_tokens_details.cached_tokens",
],
0,
)
cache_write_tokens = nested.getter(usage, "cache_creation_input_tokens", 0) or 0
self.message_cached_tokens += cache_hit_tokens
self.message_tokens_sent += prompt_tokens
else:
prompt_tokens = active_model.token_count(messages)
completion_tokens = active_model.token_count(self.partial_response_content)
self.message_tokens_sent += prompt_tokens
self.message_tokens_received += completion_tokens
UsageMeta._record_token_usage(active_model.name, prompt_tokens)
if prompt_tokens > 0:
hit_rate = round(cache_hit_tokens / prompt_tokens * 100, 1) if cache_hit_tokens else 0.0
else:
hit_rate = 0.0
tokens_str = f"{format_tokens(prompt_tokens)} {hit_rate:.1f}%"
tokens_report = f"{tokens_str} {format_tokens(completion_tokens)} "
tokens_report = self.token_profiler.add_to_usage_report(
tokens_report, self.message_tokens_sent, self.message_tokens_received
)
total_combined_tokens = (
self.total_tokens_sent
+ self.total_tokens_received
+ self.message_tokens_sent
+ self.message_tokens_received
)
total_combined_cached = self.total_cached_tokens + self.message_cached_tokens
total_input_tokens = self.total_tokens_sent + self.message_tokens_sent
if total_input_tokens > 0:
total_hit_rate = (
round(total_combined_cached / total_input_tokens * 100, 1)
if total_combined_cached
else 0.0
)
else:
total_hit_rate = 0.0
total_stats = f"{format_tokens(total_combined_tokens)} {total_hit_rate:.1f}% "
if not active_model.info.get("input_cost_per_token"):
self.usage_report = tokens_report + " " + total_stats
return
try:
cost = litellm.completion_cost(completion_response=completion)
except Exception:
cost = 0
if not cost:
cost = self.compute_costs_from_tokens(
prompt_tokens,
completion_tokens,
cache_write_tokens,
cache_hit_tokens,
model=active_model,
)
self.total_cost += cost
self.message_cost += cost
cost_report = (
f"${self.format_cost(self.message_cost)} {total_stats}"
f" ${self.format_cost(self.total_cost)}"
)
self.usage_report = tokens_report + " " + cost_report
def format_cost(self, value):
if value == 0:
return "0.00"
magnitude = abs(value)
if magnitude >= 0.01:
return f"{value:.2f}"
else:
return f"{value:.{max(2, 2 - int(math.log10(magnitude)))}f}"
def compute_costs_from_tokens(
self, prompt_tokens, completion_tokens, cache_write_tokens, cache_hit_tokens, model=None
):
cost = 0
active_model = model or self.get_active_model()
info = getattr(active_model, "info", {}) or {}
input_cost_per_token = info.get("input_cost_per_token") or 0
output_cost_per_token = info.get("output_cost_per_token") or 0
input_cost_per_token_cache_hit = (
info.get("input_cost_per_token_cache_hit")
or info.get("cache_read_input_token_cost")
or 0
)
if input_cost_per_token_cache_hit:
cost += cache_hit_tokens * input_cost_per_token_cache_hit
cost += (prompt_tokens - cache_hit_tokens) * input_cost_per_token
else:
cost += cache_write_tokens * input_cost_per_token * 1.25
cost += cache_hit_tokens * input_cost_per_token * 0.10
cost += (prompt_tokens - cache_hit_tokens) * input_cost_per_token
cost += completion_tokens * output_cost_per_token
return cost
def record_background_usage_and_cost(self, messages, completion=None, model=None):
"""Account for token usage and cost of a background model call."""
active_model = model or self.get_active_model()
prompt_tokens = 0
completion_tokens = 0
cache_hit_tokens = 0
cache_write_tokens = 0
usage = nested.getter(completion, "usage") if completion else None
if usage is not None:
prompt_tokens = (
nested.getter(usage, ["prompt_tokens", "input_tokens", "prompt_eval_count"], 0) or 0
)
completion_tokens = (
nested.getter(usage, ["completion_tokens", "output_tokens", "eval_count"], 0) or 0
)
cache_hit_tokens = _first_usage_tokens(
usage,
[
"prompt_cache_hit_tokens",
"cache_read_input_tokens",
"input_tokens_details.cached_tokens",
"prompt_tokens_details.cached_tokens",
],
0,
)
cache_write_tokens = nested.getter(usage, "cache_creation_input_tokens", 0) or 0
elif active_model is not None:
prompt_tokens = active_model.token_count(messages) or 0
model_name = getattr(active_model, "name", None)
UsageMeta._record_token_usage(model_name, prompt_tokens)
self.total_tokens_sent += prompt_tokens
self.total_tokens_received += completion_tokens
self.total_cached_tokens += cache_hit_tokens
info = getattr(active_model, "info", {}) or {}
if not info.get("input_cost_per_token"):
return
try:
cost = litellm.completion_cost(completion_response=completion)
except Exception:
cost = 0
if not cost:
cost = self.compute_costs_from_tokens(
prompt_tokens,
completion_tokens,
cache_write_tokens,
cache_hit_tokens,
model=active_model,
)
self.total_cost += cost
def calculate_dynamic_sleep(self, model=None):
"""Compute how long to sleep before the next LLM API call to stay under the configured per-minute token limit.
Uses the rolling token-usage buffer (last 60s) to estimate:
* ``used_last_min`` - prompt tokens already consumed in the window
* ``max_request`` - largest single request observed in the window
* ``requests_per_min`` - request rate observed in the window
Returns a sleep duration (seconds) that is ``0`` when the current
trajectory stays within 90% of ``tokens_per_minute`` both over the next
15 seconds and, if sustained, over a full minute. Otherwise it returns a
pause rounded up to the nearest 0.25s interval.
"""
limit = nested.getter(
getattr(self, "args", None), "tokens_per_minute", DEFAULT_TOKENS_PER_MINUTE
)
if not isinstance(limit, (int, float)):
limit = DEFAULT_TOKENS_PER_MINUTE
if limit