diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 629667cee93..517eac3ec68 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -946,15 +946,20 @@ def run_conversation( # the OpenAI SDK. Sanitizing here prevents the 3-retry cycle. _sanitize_messages_surrogates(api_messages) - # Calculate approximate request size for logging + # Calculate approximate request size for logging and pressure checks. + # estimate_messages_tokens_rough(api_messages) includes the system + # prompt copy but not the tool schema payload, which is sent as a + # separate field. Add tools back for compression decisions so long + # tool-heavy turns do not creep up to the context ceiling and leave + # no room for the model's final answer. total_chars = sum(len(str(msg)) for msg in api_messages) approx_tokens = estimate_messages_tokens_rough(api_messages) - approx_request_tokens = estimate_request_tokens_rough( + request_pressure_tokens = estimate_request_tokens_rough( api_messages, tools=agent.tools or None ) _runtime_context_error = _ollama_context_limit_error( - agent, approx_request_tokens + agent, request_pressure_tokens ) if _runtime_context_error: final_response = _runtime_context_error @@ -969,6 +974,64 @@ def run_conversation( except Exception: pass break + + _ctx_len = int(getattr(agent.context_compressor, "context_length", 0) or 0) + _threshold_tokens = int(getattr(agent.context_compressor, "threshold_tokens", 0) or 0) + _reserve_base = agent.max_tokens if isinstance(agent.max_tokens, int) and agent.max_tokens > 0 else 8192 + _reserve_cap = max(2048, _ctx_len // 4) if _ctx_len else _reserve_base + _output_reserve_tokens = min(max(_reserve_base, 8192), _reserve_cap) + _output_pressure_limit = (_ctx_len - _output_reserve_tokens) if _ctx_len else 0 + _compression_pressure_limit = _threshold_tokens or _output_pressure_limit + if _output_pressure_limit > 0: + _compression_pressure_limit = ( + min(_compression_pressure_limit, _output_pressure_limit) + if _compression_pressure_limit > 0 else _output_pressure_limit + ) + + if ( + agent.compression_enabled + and _compression_pressure_limit > 0 + and request_pressure_tokens >= _compression_pressure_limit + and len(messages) > 1 + and compression_attempts < 3 + ): + compression_attempts += 1 + logger.info( + "Pre-API compression: ~%s request tokens >= %s pressure limit " + "(threshold=%s, context=%s, output_reserve=%s, attempt=%s/3)", + f"{request_pressure_tokens:,}", + f"{_compression_pressure_limit:,}", + f"{_threshold_tokens:,}" if _threshold_tokens else "unknown", + f"{_ctx_len:,}" if _ctx_len else "unknown", + f"{_output_reserve_tokens:,}" if _output_reserve_tokens else "unknown", + compression_attempts, + ) + agent._emit_status( + f"📦 Pre-API compression: ~{request_pressure_tokens:,} tokens " + f"near the context/output limit. Compacting before the next model call." + ) + messages, active_system_prompt = agent._compress_context( + messages, + system_message, + approx_tokens=request_pressure_tokens, + task_id=effective_task_id, + ) + # Reset retry/empty-response state so the compacted request + # gets a fresh chance instead of inheriting stale recovery + # counters from the pre-compaction history. + agent._empty_content_retries = 0 + agent._thinking_prefill_retries = 0 + agent._last_content_with_tools = None + agent._last_content_tools_all_housekeeping = False + agent._mute_post_response = False + # Compression creates a new durable session boundary; write all + # compacted messages on the next persistence flush and rebuild + # the API-message copy from the compressed history. + conversation_history = None + api_call_count -= 1 + agent._api_call_count = api_call_count + agent.iteration_budget.refund() + continue # Thinking spinner for quiet mode (animated during API call) thinking_spinner = None @@ -1508,7 +1571,14 @@ def run_conversation( else: incomplete_reason = getattr(incomplete_details, "reason", None) if status == "incomplete" and incomplete_reason in {"max_output_tokens", "length"}: - finish_reason = "length" + # Responses API max-output exhaustion is a normal + # Codex incomplete turn. Let the Codex-specific + # continuation path below append the incomplete + # assistant state and retry, instead of routing to + # the generic chat-completions length rollback that + # emits "Response truncated due to output length + # limit" and stops gateway turns. + finish_reason = "incomplete" else: finish_reason = "stop" elif agent.api_mode == "anthropic_messages": diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py index 1d355c65543..e2d816dc256 100644 --- a/tests/run_agent/test_run_agent_codex_responses.py +++ b/tests/run_agent/test_run_agent_codex_responses.py @@ -123,6 +123,25 @@ def _codex_incomplete_message_response(text: str): ) +def _codex_max_output_incomplete_response(text: str = ""): + content = [] + if text: + content.append(SimpleNamespace(type="output_text", text=text)) + return SimpleNamespace( + output=[ + SimpleNamespace( + type="message", + status="incomplete", + content=content, + ) + ], + usage=SimpleNamespace(input_tokens=270_000, output_tokens=1, total_tokens=270_001), + status="incomplete", + incomplete_details=SimpleNamespace(reason="max_output_tokens"), + model="gpt-5-codex", + ) + + def _codex_commentary_message_response(text: str): return SimpleNamespace( output=[ @@ -1388,6 +1407,87 @@ def test_run_conversation_codex_continues_after_incomplete_interim_message(monke assert any(msg.get("role") == "tool" and msg.get("tool_call_id") == "call_1" for msg in result["messages"]) +def test_run_conversation_codex_continues_after_max_output_incomplete(monkeypatch): + """Codex max_output_tokens terminal status is a resumable incomplete turn. + + It must not be routed through the generic chat-completions length handler, + which returns the user-facing "Response truncated due to output length + limit" warning and stops the gateway turn. + """ + agent = _build_agent(monkeypatch) + responses = [ + _codex_max_output_incomplete_response("Partial final answer"), + _codex_message_response(" after continuation."), + ] + monkeypatch.setattr(agent, "_interruptible_api_call", lambda api_kwargs: responses.pop(0)) + + result = agent.run_conversation("write a long final answer") + + assert result["completed"] is True + assert result["final_response"] == "after continuation." + assert "Response truncated due to output length limit" not in str(result) + assert any( + msg.get("role") == "assistant" + and msg.get("finish_reason") == "incomplete" + and "Partial final answer" in (msg.get("content") or "") + for msg in result["messages"] + ) + + +def test_run_conversation_compresses_mid_turn_before_output_budget_exhaustion(monkeypatch): + """Long tool-heavy turns should compact before the next API request. + + Initial preflight compression only sees the user's first message. A single + turn can then grow by many tool results and leave almost no output budget + (the live 271k/272k GPT-5.5 failure). The agent should re-check request + pressure before every API call and compact before asking the model to + produce the final answer. + """ + agent = _build_agent(monkeypatch) + agent.context_compressor.context_length = 20_000 + agent.context_compressor.threshold_tokens = 20_000 + + responses = [ + _codex_tool_call_response(), + _codex_message_response("Summary after compaction."), + ] + requests = [] + monkeypatch.setattr( + agent, + "_interruptible_api_call", + lambda api_kwargs: requests.append(api_kwargs) or responses.pop(0), + ) + + def _fake_execute_tool_calls(assistant_message, messages, effective_task_id, api_call_count=0): + for call in assistant_message.tool_calls: + messages.append( + { + "role": "tool", + "tool_call_id": call.id, + "content": "x" * 80_000, + } + ) + + compress_calls = [] + + def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None): + compress_calls.append(approx_tokens) + return [ + {"role": "user", "content": "[summary of prior tool-heavy work]"}, + ], "You are Hermes." + + monkeypatch.setattr(agent, "_execute_tool_calls", _fake_execute_tool_calls) + monkeypatch.setattr(agent, "_compress_context", _fake_compress_context) + + result = agent.run_conversation("do a tool-heavy task") + + assert result["completed"] is True + assert result["final_response"] == "Summary after compaction." + assert len(compress_calls) == 1 + assert compress_calls[0] >= 15_000 + assert len(requests) == 2 + + def test_normalize_codex_response_marks_commentary_only_message_as_incomplete(monkeypatch): agent = _build_agent(monkeypatch) from agent.codex_responses_adapter import _normalize_codex_response