diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 629667cee93..41a13f5d628 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,83 @@ def run_conversation( except Exception: pass break + + # Pre-API pressure check. The turn-prologue preflight only saw the + # incoming user message; a single turn can then grow by many large + # tool results and leave no output budget before the NEXT call (the + # live 271k/272k Codex failure). The post-response should_compress + # gate at the tool-loop tail uses API-reported last_prompt_tokens, + # which LAGS a just-appended huge tool result — so it misses this + # case. Re-check here against the current request estimate. + # + # Mirror the turn-prologue preflight's guard chain exactly (see + # turn_context.py): (1) defer when the rough estimate is known-noisy + # relative to a recent real provider prompt that fit under threshold + # (schema overhead / post-compaction over-count, #36718); (2) skip + # while a same-session compression-failure cooldown is active; (3) then + # should_compress() — reusing the canonical threshold_tokens (output + # room already reserved by _compute_threshold_tokens) and its summary- + # LLM cooldown + anti-thrash guards (#11529). compression_attempts is a + # hard per-turn backstop shared with the overflow error handlers. + _compressor = agent.context_compressor + _defer_preflight = getattr( + _compressor, "should_defer_preflight_to_real_usage", lambda _t: False + ) + _compression_cooldown = getattr( + _compressor, "get_active_compression_failure_cooldown", lambda: None + )() + if ( + agent.compression_enabled + and len(messages) > 1 + and compression_attempts < 3 + and not _defer_preflight(request_pressure_tokens) + and not _compression_cooldown + and _compressor.should_compress(request_pressure_tokens) + ): + compression_attempts += 1 + logger.info( + "Pre-API compression: ~%s request tokens >= %s threshold " + "(context=%s, attempt=%s/3)", + f"{request_pressure_tokens:,}", + f"{int(getattr(_compressor, 'threshold_tokens', 0) or 0):,}", + f"{int(getattr(_compressor, 'context_length', 0) or 0):,}" + if getattr(_compressor, "context_length", 0) 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 + # Re-baseline the flush cursor for the compaction mode that just + # ran. Legacy session-rotation returns None (the child session has + # not seen the compacted transcript, so the next flush writes it + # whole); in-place compaction returns list(messages) because the + # compacted rows are already persisted under the same session id — + # leaving None there would re-append them, doubling the active + # context and retriggering compression. Mirrors the post-response + # and preflight compaction sites; see + # conversation_history_after_compression(). + conversation_history = conversation_history_after_compression( + agent, messages + ) + 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 +1590,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..a1a46e2f44f 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,160 @@ 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_mid_turn_compaction_does_not_double_persist_in_place_rows(monkeypatch, tmp_path): + """Mid-turn pre-API compaction must re-baseline the flush cursor. + + In-place compaction (``compression.in_place: True``, the default) inserts + the compacted rows into the session DB itself via ``archive_and_compact`` + WITHOUT stamping them with the intrinsic persisted-marker. The loop must + therefore set ``conversation_history`` to those compacted dicts so the next + flush skips them by identity. Setting ``conversation_history = None`` here + (as the original PR did) makes the flush treat the already-persisted + compacted dicts as new and append them a second time — doubling the active + context and retriggering compression. This guards that regression with a + REAL SessionDB and the REAL archive_and_compact path (no persist stubs). + """ + from hermes_state import SessionDB + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + agent = _build_agent(monkeypatch) + # _build_agent stubs _persist_session; restore the real one so the flush + # cursor / double-write behaviour is exercised end to end. + agent._persist_session = run_agent.AIAgent._persist_session.__get__(agent) + agent._cleanup_task_resources = lambda task_id: None + + agent.context_compressor.context_length = 20_000 + agent.context_compressor.threshold_tokens = 20_000 + + agent._session_db = SessionDB() + agent._ensure_db_session() + + responses = [ + _codex_tool_call_response(), + _codex_message_response("Summary after compaction."), + ] + monkeypatch.setattr( + agent, "_interruptible_api_call", lambda api_kwargs: 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} + ) + + def _fake_compress_context(messages, system_message, *, approx_tokens=None, task_id="default", focus_topic=None): + # Emulate the real in-place compaction DB side effect: soft-archive the + # prior rows and insert the compacted set under the SAME session id, + # then reset the flush identity seed — exactly as archive_and_compact + + # the in_place branch in conversation_compression.py do. + agent._last_compaction_in_place = True + compacted = [{"role": "user", "content": "[summary of prior tool-heavy work]"}] + agent._session_db.archive_and_compact(agent.session_id, compacted) + agent._flushed_db_message_ids = set() + return compacted, "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 + + # The compacted summary row must appear exactly once in the active + # transcript that a resume would reload. + active = agent._session_db.get_messages(agent.session_id) + summary_rows = [ + m for m in active + if isinstance(m.get("content"), str) + and "summary of prior tool-heavy work" in m["content"] + ] + assert len(summary_rows) == 1, ( + f"compacted summary row double-persisted: {len(summary_rows)} copies " + "(conversation_history flush cursor not re-baselined for in-place compaction)" + ) + + 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