fix: recover Codex max-output truncation + re-baseline mid-turn compaction flush

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 _perform_api_call(next_api_kwargs):
                     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":

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 _fake_execute_tool_calls(assistant_message, messages, effective_task_id):
     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