diff --git a/.github/workflows/contributor-check.yml b/.github/workflows/contributor-check.yml index f8d65a3ea4b6c..e4e0cb0e8a765 100644 --- a/.github/workflows/contributor-check.yml +++ b/.github/workflows/contributor-check.yml @@ -35,7 +35,7 @@ jobs: while IFS= read -r email; do # Skip teknium and bot emails case "$email" in - *teknium*|*noreply@github.com*|*dependabot*|*github-actions*|*anthropic.com*|*cursor.com*) + *teknium*|*noreply@github.com*|*@users.noreply.github.com|*dependabot*|*github-actions*|*anthropic.com*|*cursor.com*) continue ;; esac diff --git a/run_agent.py b/run_agent.py index 5922534646c10..6bb1796efb9d1 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2736,6 +2736,168 @@ def _usage_summary_for_api_request_hook(self, response: Any) -> Optional[Dict[st summary["total_tokens"] = cu.total_tokens return summary + def _record_response_usage(self, response: Any, *, api_duration: float) -> Optional[Any]: + """Update compressor/session accounting from a successful API response.""" + raw_usage = getattr(response, "usage", None) + if not raw_usage: + return None + + canonical_usage = normalize_usage( + raw_usage, + provider=self.provider, + api_mode=self.api_mode, + ) + prompt_tokens = canonical_usage.prompt_tokens + completion_tokens = canonical_usage.output_tokens + total_tokens = canonical_usage.total_tokens + usage_dict = { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": total_tokens, + } + self.context_compressor.update_from_response(usage_dict) + + # Cache discovered context length after successful call. + # Only persist limits confirmed by the provider (parsed + # from the error message), not guessed probe tiers. + if getattr(self.context_compressor, "_context_probed", False): + ctx = self.context_compressor.context_length + if getattr(self.context_compressor, "_context_probe_persistable", False): + save_context_length(self.model, self.base_url, ctx) + self._safe_print( + f"{self.log_prefix}💾 Cached context length: {ctx:,} tokens for {self.model}" + ) + self.context_compressor._context_probed = False + self.context_compressor._context_probe_persistable = False + + self.session_prompt_tokens += prompt_tokens + self.session_completion_tokens += completion_tokens + self.session_total_tokens += total_tokens + self.session_api_calls += 1 + self.session_input_tokens += canonical_usage.input_tokens + self.session_output_tokens += canonical_usage.output_tokens + self.session_cache_read_tokens += canonical_usage.cache_read_tokens + self.session_cache_write_tokens += canonical_usage.cache_write_tokens + self.session_reasoning_tokens += canonical_usage.reasoning_tokens + + # Log API call details for debugging/observability + _cache_pct = "" + if canonical_usage.cache_read_tokens and prompt_tokens: + _cache_pct = ( + f" cache={canonical_usage.cache_read_tokens}/{prompt_tokens} " + f"({100 * canonical_usage.cache_read_tokens / prompt_tokens:.0f}%)" + ) + logger.info( + "API call #%d: model=%s provider=%s in=%d out=%d total=%d latency=%.1fs%s", + self.session_api_calls, + self.model, + self.provider or "unknown", + prompt_tokens, + completion_tokens, + total_tokens, + api_duration, + _cache_pct, + ) + + cost_result = estimate_usage_cost( + self.model, + canonical_usage, + provider=self.provider, + base_url=self.base_url, + api_key=getattr(self, "api_key", ""), + ) + if cost_result.amount_usd is not None: + self.session_estimated_cost_usd += float(cost_result.amount_usd) + self.session_cost_status = cost_result.status + self.session_cost_source = cost_result.source + + # Persist token counts to session DB for /insights. + # Do this for every platform with a session_id so non-CLI + # sessions (gateway, cron, delegated runs) cannot lose + # token/accounting data if a higher-level persistence path + # is skipped or fails. Gateway/session-store writes use + # absolute totals, so they safely overwrite these per-call + # deltas instead of double-counting them. + if self._session_db and self.session_id: + try: + self._session_db.update_token_counts( + self.session_id, + input_tokens=canonical_usage.input_tokens, + output_tokens=canonical_usage.output_tokens, + cache_read_tokens=canonical_usage.cache_read_tokens, + cache_write_tokens=canonical_usage.cache_write_tokens, + reasoning_tokens=canonical_usage.reasoning_tokens, + estimated_cost_usd=float(cost_result.amount_usd) + if cost_result.amount_usd is not None + else None, + cost_status=cost_result.status, + cost_source=cost_result.source, + billing_provider=self.provider, + billing_base_url=self.base_url, + billing_mode="subscription_included" + if cost_result.status == "included" + else None, + model=self.model, + ) + except Exception: + pass # never block the agent loop + + if self.verbose_logging: + logging.debug( + "Token usage: prompt=%s, completion=%s, total=%s", + f"{usage_dict['prompt_tokens']:,}", + f"{usage_dict['completion_tokens']:,}", + f"{usage_dict['total_tokens']:,}", + ) + + # Log cache hit stats when prompt caching is active + if self._use_prompt_caching: + if self.api_mode == "anthropic_messages": + # Anthropic uses cache_read_input_tokens / cache_creation_input_tokens + cached = getattr(response.usage, "cache_read_input_tokens", 0) or 0 + written = getattr(response.usage, "cache_creation_input_tokens", 0) or 0 + else: + # OpenRouter uses prompt_tokens_details.cached_tokens + details = getattr(response.usage, "prompt_tokens_details", None) + cached = getattr(details, "cached_tokens", 0) or 0 if details else 0 + written = getattr(details, "cache_write_tokens", 0) or 0 if details else 0 + prompt = usage_dict["prompt_tokens"] + hit_pct = (cached / prompt * 100) if prompt > 0 else 0 + if not self.quiet_mode: + self._vprint( + f"{self.log_prefix} 💾 Cache: {cached:,}/{prompt:,} tokens " + f"({hit_pct:.0f}% hit, {written:,} written)" + ) + + return canonical_usage + + def _maybe_compress_after_empty_reasoning_response( + self, + messages: List[Dict[str, Any]], + system_message: str, + *, + real_tokens: int, + task_id: str, + status_message: str, + ) -> tuple[List[Dict[str, Any]], str, bool]: + """Try a one-pass context compaction before retrying an empty reasoning turn.""" + if not self.compression_enabled or real_tokens <= 0: + return messages, system_message, False + + compressor = getattr(self, "context_compressor", None) + if compressor is None or not compressor.should_compress(real_tokens): + return messages, system_message, False + + original_len = len(messages) + self._emit_status(status_message) + compressed_messages, new_system_prompt = self._compress_context( + messages, + system_message, + approx_tokens=compressor.last_prompt_tokens or real_tokens, + task_id=task_id, + ) + return compressed_messages, new_system_prompt, len(compressed_messages) < original_len + def _dump_api_request_debug( self, api_kwargs: Dict[str, Any], @@ -8279,6 +8441,7 @@ def run_conversation( thinking_sig_retry_attempted = False has_retried_429 = False restart_with_compressed_messages = False + compression_restart_consumed_call = False restart_with_length_continuation = False finish_reason = "stop" @@ -8588,6 +8751,11 @@ def _stop_spinner(): else: finish_reason = response.choices[0].finish_reason + canonical_usage = self._record_response_usage( + response, + api_duration=api_duration, + ) + if finish_reason == "length": self._vprint(f"{self.log_prefix}⚠️ Response truncated (finish_reason='length') - model hit max output tokens", force=True) @@ -8602,6 +8770,14 @@ def _stop_spinner(): _trunc_msg = response.choices[0].message if (hasattr(response, "choices") and response.choices) else None _trunc_content = getattr(_trunc_msg, "content", None) if _trunc_msg else None _trunc_has_tool_calls = bool(getattr(_trunc_msg, "tool_calls", None)) if _trunc_msg else False + _has_structured_reasoning = bool( + _trunc_msg + and ( + getattr(_trunc_msg, "reasoning", None) + or getattr(_trunc_msg, "reasoning_content", None) + or getattr(_trunc_msg, "reasoning_details", None) + ) + ) elif self.api_mode == "anthropic_messages": # Anthropic response.content is a list of blocks _text_parts = [] @@ -8609,6 +8785,9 @@ def _stop_spinner(): if getattr(_blk, "type", None) == "text": _text_parts.append(getattr(_blk, "text", "")) _trunc_content = "\n".join(_text_parts) if _text_parts else None + _has_structured_reasoning = False + else: + _has_structured_reasoning = False # A response is "thinking exhausted" only when the model # actually produced reasoning blocks but no visible text after @@ -8626,7 +8805,7 @@ def _stop_spinner(): ) _thinking_exhausted = ( not _trunc_has_tool_calls - and _has_think_tags + and (_has_think_tags or _has_structured_reasoning) and ( (_trunc_content is not None and not self._has_content_after_think_block(_trunc_content)) or _trunc_content is None @@ -8634,6 +8813,39 @@ def _stop_spinner(): ) if _thinking_exhausted: + _real_tokens = ( + canonical_usage.total_tokens + if canonical_usage is not None + else ( + (self.context_compressor.last_prompt_tokens or 0) + + (self.context_compressor.last_completion_tokens or 0) + ) + ) + if ( + compression_attempts < max_compression_attempts + and self.compression_enabled + and _real_tokens > 0 + and self.context_compressor.should_compress(_real_tokens) + ): + compression_attempts += 1 + messages, active_system_prompt, compressed = ( + self._maybe_compress_after_empty_reasoning_response( + messages, + system_message, + real_tokens=_real_tokens, + task_id=effective_task_id, + status_message=( + "🗜️ Thinking-only truncated response hit context pressure — " + "compacting context and retrying..." + ), + ) + ) + if compressed: + conversation_history = None + compression_restart_consumed_call = True + restart_with_compressed_messages = True + break + _exhaust_error = ( "Model used all output tokens on reasoning with none left " "for the response. Try lowering reasoning effort or " @@ -8764,115 +8976,6 @@ def _stop_spinner(): "error": "First response truncated due to output length limit" } - # Track actual token usage from response for context management - if hasattr(response, 'usage') and response.usage: - canonical_usage = normalize_usage( - response.usage, - provider=self.provider, - api_mode=self.api_mode, - ) - prompt_tokens = canonical_usage.prompt_tokens - completion_tokens = canonical_usage.output_tokens - total_tokens = canonical_usage.total_tokens - usage_dict = { - "prompt_tokens": prompt_tokens, - "completion_tokens": completion_tokens, - "total_tokens": total_tokens, - } - self.context_compressor.update_from_response(usage_dict) - - # Cache discovered context length after successful call. - # Only persist limits confirmed by the provider (parsed - # from the error message), not guessed probe tiers. - if getattr(self.context_compressor, "_context_probed", False): - ctx = self.context_compressor.context_length - if getattr(self.context_compressor, "_context_probe_persistable", False): - save_context_length(self.model, self.base_url, ctx) - self._safe_print(f"{self.log_prefix}💾 Cached context length: {ctx:,} tokens for {self.model}") - self.context_compressor._context_probed = False - self.context_compressor._context_probe_persistable = False - - self.session_prompt_tokens += prompt_tokens - self.session_completion_tokens += completion_tokens - self.session_total_tokens += total_tokens - self.session_api_calls += 1 - self.session_input_tokens += canonical_usage.input_tokens - self.session_output_tokens += canonical_usage.output_tokens - self.session_cache_read_tokens += canonical_usage.cache_read_tokens - self.session_cache_write_tokens += canonical_usage.cache_write_tokens - self.session_reasoning_tokens += canonical_usage.reasoning_tokens - - # Log API call details for debugging/observability - _cache_pct = "" - if canonical_usage.cache_read_tokens and prompt_tokens: - _cache_pct = f" cache={canonical_usage.cache_read_tokens}/{prompt_tokens} ({100*canonical_usage.cache_read_tokens/prompt_tokens:.0f}%)" - logger.info( - "API call #%d: model=%s provider=%s in=%d out=%d total=%d latency=%.1fs%s", - self.session_api_calls, self.model, self.provider or "unknown", - prompt_tokens, completion_tokens, total_tokens, - api_duration, _cache_pct, - ) - - cost_result = estimate_usage_cost( - self.model, - canonical_usage, - provider=self.provider, - base_url=self.base_url, - api_key=getattr(self, "api_key", ""), - ) - if cost_result.amount_usd is not None: - self.session_estimated_cost_usd += float(cost_result.amount_usd) - self.session_cost_status = cost_result.status - self.session_cost_source = cost_result.source - - # Persist token counts to session DB for /insights. - # Do this for every platform with a session_id so non-CLI - # sessions (gateway, cron, delegated runs) cannot lose - # token/accounting data if a higher-level persistence path - # is skipped or fails. Gateway/session-store writes use - # absolute totals, so they safely overwrite these per-call - # deltas instead of double-counting them. - if self._session_db and self.session_id: - try: - self._session_db.update_token_counts( - self.session_id, - input_tokens=canonical_usage.input_tokens, - output_tokens=canonical_usage.output_tokens, - cache_read_tokens=canonical_usage.cache_read_tokens, - cache_write_tokens=canonical_usage.cache_write_tokens, - reasoning_tokens=canonical_usage.reasoning_tokens, - estimated_cost_usd=float(cost_result.amount_usd) - if cost_result.amount_usd is not None else None, - cost_status=cost_result.status, - cost_source=cost_result.source, - billing_provider=self.provider, - billing_base_url=self.base_url, - billing_mode="subscription_included" - if cost_result.status == "included" else None, - model=self.model, - ) - except Exception: - pass # never block the agent loop - - if self.verbose_logging: - logging.debug(f"Token usage: prompt={usage_dict['prompt_tokens']:,}, completion={usage_dict['completion_tokens']:,}, total={usage_dict['total_tokens']:,}") - - # Log cache hit stats when prompt caching is active - if self._use_prompt_caching: - if self.api_mode == "anthropic_messages": - # Anthropic uses cache_read_input_tokens / cache_creation_input_tokens - cached = getattr(response.usage, 'cache_read_input_tokens', 0) or 0 - written = getattr(response.usage, 'cache_creation_input_tokens', 0) or 0 - else: - # OpenRouter uses prompt_tokens_details.cached_tokens - details = getattr(response.usage, 'prompt_tokens_details', None) - cached = getattr(details, 'cached_tokens', 0) or 0 if details else 0 - written = getattr(details, 'cache_write_tokens', 0) or 0 if details else 0 - prompt = usage_dict["prompt_tokens"] - hit_pct = (cached / prompt * 100) if prompt > 0 else 0 - if not self.quiet_mode: - self._vprint(f"{self.log_prefix} 💾 Cache: {cached:,}/{prompt:,} tokens ({hit_pct:.0f}% hit, {written:,} written)") - has_retried_429 = False # Reset on success self._touch_activity(f"API call #{api_call_count} completed") break # Success, exit retry loop @@ -9635,13 +9738,15 @@ def _stop_spinner(): break if restart_with_compressed_messages: - api_call_count -= 1 - self.iteration_budget.refund() + if not compression_restart_consumed_call: + api_call_count -= 1 + self.iteration_budget.refund() # Count compression restarts toward the retry limit to prevent # infinite loops when compression reduces messages but not enough # to fit the context window. retry_count += 1 restart_with_compressed_messages = False + compression_restart_consumed_call = False continue if restart_with_length_continuation: @@ -10211,6 +10316,28 @@ def _stop_spinner(): or getattr(assistant_message, "reasoning_content", None) or getattr(assistant_message, "reasoning_details", None) ) + _real_tokens = ( + (self.context_compressor.last_prompt_tokens or 0) + + (self.context_compressor.last_completion_tokens or 0) + ) + if _has_structured and compression_attempts < max_compression_attempts: + messages, active_system_prompt, compressed = ( + self._maybe_compress_after_empty_reasoning_response( + messages, + system_message, + real_tokens=_real_tokens, + task_id=effective_task_id, + status_message=( + "🗜️ Thinking-only response hit context pressure — " + "compacting context before retry..." + ), + ) + ) + if compressed: + compression_attempts += 1 + conversation_history = None + continue + if _has_structured and self._thinking_prefill_retries < 2: self._thinking_prefill_retries += 1 logger.info( diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index d71e6a625542d..ed533db05f286 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -2228,6 +2228,71 @@ def test_length_thinking_exhausted_skips_continuation(self, agent): assert "Thinking Budget Exhausted" in result["final_response"] assert "/thinkon" in result["final_response"] + def test_length_structured_reasoning_exhausted_skips_continuation(self, agent): + """Structured reasoning fields should trigger thinking-budget recovery too.""" + self._setup_agent(agent) + resp = _mock_response( + content=None, + finish_reason="length", + reasoning_content="internal reasoning without visible text", + ) + agent.client.chat.completions.create.return_value = resp + + with ( + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("hello") + + assert result["completed"] is False + assert result["api_calls"] == 1 + assert "reasoning" in result["error"].lower() + assert "output tokens" in result["error"].lower() + assert result["final_response"] is not None + assert "Thinking Budget Exhausted" in result["final_response"] + + def test_length_structured_reasoning_compresses_before_retry(self, agent): + """When thinking exhausts the budget under high context pressure, compact before retrying.""" + self._setup_agent(agent) + agent.compression_enabled = True + agent.context_compressor.threshold_tokens = 1000 + history = [ + {"role": "user", "content": "old question"}, + {"role": "assistant", "content": "old answer"}, + {"role": "user", "content": "another question"}, + ] + resp1 = _mock_response( + content=None, + finish_reason="length", + reasoning_content="internal reasoning without visible text", + usage={"prompt_tokens": 900, "completion_tokens": 200}, + ) + resp2 = _mock_response( + content="Recovered after compression.", + finish_reason="stop", + usage={"prompt_tokens": 120, "completion_tokens": 40}, + ) + agent.client.chat.completions.create.side_effect = [resp1, resp2] + + compressed_messages = [ + {"role": "user", "content": "Compressed summary of earlier context."}, + {"role": "user", "content": "hello"}, + ] + + with ( + patch.object(agent, "_compress_context", return_value=(compressed_messages, agent._cached_system_prompt)) as mock_compress, + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("hello", conversation_history=history) + + mock_compress.assert_called_once() + assert result["completed"] is True + assert result["final_response"] == "Recovered after compression." + assert result["api_calls"] == 2 + def test_length_empty_content_without_think_tags_retries_normally(self, agent): """When finish_reason='length' and content is None but no think tags, fall through to normal continuation retry (not thinking-exhaustion).""" @@ -2247,6 +2312,47 @@ def test_length_empty_content_without_think_tags_retries_normally(self, agent): assert result["api_calls"] == 3 assert result["completed"] is False + def test_reasoning_only_response_compresses_before_prefill_when_context_pressure_is_high(self, agent): + """Reasoning-only stop responses should compact context before adding more prefill state.""" + self._setup_agent(agent) + agent.compression_enabled = True + agent.context_compressor.threshold_tokens = 1000 + history = [ + {"role": "user", "content": "old question"}, + {"role": "assistant", "content": "old answer"}, + {"role": "user", "content": "another question"}, + ] + resp1 = _mock_response( + content=None, + finish_reason="stop", + reasoning_content="internal reasoning without visible text", + usage={"prompt_tokens": 900, "completion_tokens": 200}, + ) + resp2 = _mock_response( + content="Recovered after compression.", + finish_reason="stop", + usage={"prompt_tokens": 120, "completion_tokens": 40}, + ) + agent.client.chat.completions.create.side_effect = [resp1, resp2] + + compressed_messages = [ + {"role": "user", "content": "Compressed summary of earlier context."}, + {"role": "user", "content": "hello"}, + ] + + with ( + patch.object(agent, "_compress_context", return_value=(compressed_messages, agent._cached_system_prompt)) as mock_compress, + patch.object(agent, "_persist_session"), + patch.object(agent, "_save_trajectory"), + patch.object(agent, "_cleanup_task_resources"), + ): + result = agent.run_conversation("hello", conversation_history=history) + + mock_compress.assert_called_once() + assert result["completed"] is True + assert result["final_response"] == "Recovered after compression." + assert result["api_calls"] == 2 + def test_length_with_tool_calls_returns_partial_without_executing_tools(self, agent): self._setup_agent(agent) bad_tc = _mock_tool_call(