diff --git a/agent/agent_init.py b/agent/agent_init.py
index 599927ba0c78..65c897c0c833 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -1781,6 +1781,34 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     compression_enabled = str(_compression_cfg.get("enabled", True)).lower() in {"true", "1", "yes"}
     compression_target_ratio = float(_compression_cfg.get("target_ratio", 0.20))
     compression_protect_last = int(_compression_cfg.get("protect_last_n", 20))
+    # Cap on compression retry rounds before a turn gives up with "max
+    # compression attempts reached" (compression.max_attempts).  Hardcoding 3
+    # strands sessions that legitimately need more rounds — e.g. a restart
+    # history reload whose incompressible tool schemas keep the request
+    # estimate above the threshold even though the messages compress fine
+    # (the #62605 failure class).  Default 3 preserves current behavior, so
+    # an unset key is behavior-neutral; validated >= 1, hard-capped at 10,
+    # and any non-int-like value falls back to 3.  Booleans are rejected
+    # (bool subclasses int, so int(True) would silently become 1) and
+    # fractional floats are rejected rather than truncated — "4.7 attempts"
+    # is a config mistake, not a request for 4.
+    _raw_max_attempts = _compression_cfg.get("max_attempts", 3)
+    if isinstance(_raw_max_attempts, bool):
+        compression_max_attempts = 3
+    elif isinstance(_raw_max_attempts, int):
+        compression_max_attempts = _raw_max_attempts
+    elif isinstance(_raw_max_attempts, float):
+        compression_max_attempts = (
+            int(_raw_max_attempts) if _raw_max_attempts.is_integer() else 3
+        )
+    else:
+        try:
+            compression_max_attempts = int(str(_raw_max_attempts).strip())
+        except (TypeError, ValueError):
+            compression_max_attempts = 3
+    if compression_max_attempts < 1:
+        compression_max_attempts = 3
+    compression_max_attempts = min(compression_max_attempts, 10)
     # protect_first_n is the number of non-system messages to protect at
     # the head, in addition to the system prompt (which is always
     # implicitly protected by the compressor).  Floor at 0 — a value of
@@ -2224,6 +2252,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     agent.compression_enabled = compression_enabled
     agent.compression_in_place = compression_in_place
     agent.codex_app_server_auto_compaction = codex_app_server_auto_compaction
+    agent.max_compression_attempts = compression_max_attempts
 
     # Reject models whose context window is below the minimum required
     # for reliable tool-calling workflows (64K tokens).
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 517c9d86ed7e..4cea0eb989a0 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -685,6 +685,13 @@ def run_conversation(
     truncated_tool_call_retries = 0
     truncated_response_parts: List[str] = []
     compression_attempts = 0
+    # One resolved per-turn compression attempt cap, shared by every site that
+    # consumes ``compression_attempts``: the pre-API pressure gate, the
+    # overflow/413 retry handlers, and the post-tool compaction gate.
+    # Config-driven via compression.max_attempts (parsed + validated in
+    # agent_init); default 3 preserves the prior hardcoded behavior for
+    # objects without the attribute (older pickles / minimal stubs).
+    max_compression_attempts = getattr(agent, "max_compression_attempts", 3)
     _last_preflight_pressure: Optional[int] = None
     _preflight_compression_blocked = _ctx.preflight_compression_blocked
     _turn_exit_reason = "unknown"  # Diagnostic: why the loop ended
@@ -1168,7 +1175,7 @@ def run_conversation(
         if (
             agent.compression_enabled
             and len(messages) > 1
-            and compression_attempts < 3
+            and compression_attempts < max_compression_attempts
             and not _preflight_compression_blocked
             and not _defer_preflight(request_pressure_tokens)
             and not _compression_cooldown
@@ -1177,12 +1184,13 @@ def run_conversation(
             compression_attempts += 1
             logger.info(
                 "Pre-API compression: ~%s request tokens >= %s threshold "
-                "(context=%s, attempt=%s/3)",
+                "(context=%s, attempt=%s/%s)",
                 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,
+                max_compression_attempts,
             )
             agent._emit_status(
                 f"📦 Pre-API compression: ~{request_pressure_tokens:,} tokens "
@@ -1252,7 +1260,6 @@ def run_conversation(
         retry_count = 0
         max_retries = agent._api_max_retries
         _retry = TurnRetryState()
-        max_compression_attempts = 3
 
         finish_reason = "stop"
         response = None  # Guard against UnboundLocalError if all retries fail
@@ -5175,7 +5182,12 @@ def _perform_api_call(next_api_kwargs):
                         messages, tools=agent.tools or None
                     )
 
-                if agent.compression_enabled and _compressor.should_compress(_real_tokens):
+                if (
+                    agent.compression_enabled
+                    and compression_attempts < max_compression_attempts
+                    and _compressor.should_compress(_real_tokens)
+                ):
+                    compression_attempts += 1
                     agent._safe_print("  ⟳ compacting context…")
                     messages, active_system_prompt = agent._compress_context(
                         messages, system_message,
diff --git a/agent/turn_context.py b/agent/turn_context.py
index 89952495b381..669d615847cf 100644
--- a/agent/turn_context.py
+++ b/agent/turn_context.py
@@ -665,7 +665,13 @@ def build_turn_context(
                 f">= {_compressor.threshold_tokens:,} threshold. "
                 "This may take a moment."
             )
-            for _pass in range(3):
+            # Preflight passes honor the same configured per-turn cap
+            # (compression.max_attempts) as the loop's compression sites;
+            # default 3 preserves the prior hardcoded behavior.
+            _max_preflight_passes = max(
+                1, int(getattr(agent, "max_compression_attempts", 3) or 3)
+            )
+            for _pass in range(_max_preflight_passes):
                 _orig_len = len(messages)
                 _orig_tokens = _preflight_tokens
                 messages, active_system_prompt = agent._compress_context(
diff --git a/cli-config.yaml.example b/cli-config.yaml.example
index d12d4ff63052..d04cde3ec528 100644
--- a/cli-config.yaml.example
+++ b/cli-config.yaml.example
@@ -429,6 +429,12 @@ compression:
   # compression of older turns.
   protect_last_n: 20
 
+  # Compression retry rounds before a turn gives up with "max compression
+  # attempts reached" (default: 3, same as the previous hardcoded value).
+  # Raise (e.g. 6) for tool-schema-heavy sessions where 3 rounds cannot bring
+  # the request estimate under the threshold. Validated >= 1, hard cap 10.
+  max_attempts: 3
+
   # Codex app-server (codex CLI runtime) thread-compaction mode. The codex
   # agent owns the real thread context on this runtime, so Hermes' summarizer
   # cannot shrink it — compaction goes through the app server instead.
diff --git a/contributors/emails/dominicbejar@gmail.com b/contributors/emails/dominicbejar@gmail.com
new file mode 100644
index 000000000000..f69e7425843e
--- /dev/null
+++ b/contributors/emails/dominicbejar@gmail.com
@@ -0,0 +1 @@
+dombejar
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index 7a527b75f53c..8290887230d7 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -1471,6 +1471,11 @@ def _ensure_hermes_home_managed(home: Path):
                                       # set this above 0.75 to override the floor.
         "target_ratio": 0.20,         # fraction of threshold to preserve as recent tail
         "protect_last_n": 20,         # minimum recent messages to keep uncompressed
+        "max_attempts": 3,            # compression retry rounds before a turn gives up
+                                      # with "max compression attempts reached". Raise
+                                      # (e.g. 6) for tool-schema-heavy sessions where 3
+                                      # rounds cannot clear the request estimate.
+                                      # Validated >= 1, hard-capped at 10.
         "hygiene_hard_message_limit": 5000,  # gateway session-hygiene force-compress threshold by message count
         "protect_first_n": 3,         # non-system head messages always preserved
                                       # verbatim, in ADDITION to the system prompt
diff --git a/tests/agent/test_compression_max_attempts_config.py b/tests/agent/test_compression_max_attempts_config.py
new file mode 100644
index 000000000000..8fb23f89d0a7
--- /dev/null
+++ b/tests/agent/test_compression_max_attempts_config.py
@@ -0,0 +1,119 @@
+"""compression.max_attempts — config-driven compression retry cap.
+
+The conversation loop's compression retry cap was hardcoded to 3, stranding
+sessions that legitimately need more rounds — e.g. a restart history reload
+whose incompressible tool schemas keep the request estimate above the
+threshold while the messages themselves compress fine (the #62605 failure
+class).  The cap is now parsed from ``compression.max_attempts`` in
+``agent_init`` and read by the loop via
+``getattr(agent, "max_compression_attempts", 3)``.
+
+These tests pin the parse/validate/attach seam: default preserved, custom
+value honored, floor and ceiling enforced, garbage tolerated.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+from pathlib import Path
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _config(max_attempts=None) -> dict:
+    compression = {
+        "enabled": True,
+        "threshold": 0.50,
+        "target_ratio": 0.20,
+        "protect_first_n": 3,
+        "protect_last_n": 20,
+    }
+    if max_attempts is not None:
+        compression["max_attempts"] = max_attempts
+    return {
+        "compression": compression,
+        "prompt_caching": {"cache_ttl": "5m"},
+        "sessions": {},
+        "bedrock": {},
+    }
+
+
+def _make_agent(monkeypatch, tmp_path: Path, *, max_attempts=None):
+    from hermes_cli import config as config_mod
+
+    monkeypatch.setattr(
+        config_mod, "load_config", lambda: _config(max_attempts=max_attempts)
+    )
+    db = SessionDB(db_path=tmp_path / "state.db")
+    with contextlib.redirect_stdout(io.StringIO()):
+        agent = AIAgent(
+            base_url="https://chatgpt.com/backend-api/codex",
+            api_key="test-key",
+            provider="openai-codex",
+            model="gpt-5.5",
+            enabled_toolsets=[],
+            disabled_toolsets=[],
+            quiet_mode=True,
+            skip_memory=True,
+            session_db=db,
+            session_id="max-attempts-test",
+        )
+    return agent
+
+
+class TestCompressionMaxAttemptsConfig:
+    def test_default_is_three_when_unset(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path)
+        assert agent.max_compression_attempts == 3
+
+    def test_custom_value_is_honored(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=6)
+        assert agent.max_compression_attempts == 6
+
+    def test_hard_capped_at_ten(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=25)
+        assert agent.max_compression_attempts == 10
+
+    def test_zero_and_negative_fall_back_to_default(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=0)
+        assert agent.max_compression_attempts == 3
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=-2)
+        assert agent.max_compression_attempts == 3
+
+    def test_non_integer_falls_back_to_default(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts="lots")
+        assert agent.max_compression_attempts == 3
+
+    def test_boolean_is_rejected_not_coerced(self, monkeypatch, tmp_path):
+        # bool subclasses int: int(True) == 1 would silently near-disable
+        # compression retries. YAML `max_attempts: true` must fall back to 3.
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=True)
+        assert agent.max_compression_attempts == 3
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=False)
+        assert agent.max_compression_attempts == 3
+
+    def test_fractional_float_is_rejected_not_truncated(self, monkeypatch, tmp_path):
+        # "4.7 attempts" is a config mistake, not a request for 4.
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=4.7)
+        assert agent.max_compression_attempts == 3
+
+    def test_integral_float_and_numeric_string_are_accepted(
+        self, monkeypatch, tmp_path
+    ):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=5.0)
+        assert agent.max_compression_attempts == 5
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts="6")
+        assert agent.max_compression_attempts == 6
+
+    def test_loop_pickup_degrades_to_default_when_attribute_missing(
+        self, monkeypatch, tmp_path
+    ):
+        # The loop reads getattr(agent, "max_compression_attempts", 3): a
+        # configured agent exposes its value, and an object without the
+        # attribute (older pickle / minimal stub) degrades to the prior
+        # hardcoded behavior.
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=7)
+        assert getattr(agent, "max_compression_attempts", 3) == 7
+        assert getattr(object(), "max_compression_attempts", 3) == 3
diff --git a/tests/run_agent/test_post_tool_compression_attempt_cap.py b/tests/run_agent/test_post_tool_compression_attempt_cap.py
new file mode 100644
index 000000000000..a15c13eb4c5a
--- /dev/null
+++ b/tests/run_agent/test_post_tool_compression_attempt_cap.py
@@ -0,0 +1,206 @@
+"""Behavioral regression tests for the post-tool compression attempt cap.
+
+The pre-API pressure gate, the overflow/413 error handlers, and the post-tool
+compaction gate all share ``compression_attempts`` as a per-turn backstop,
+bounded by the resolved ``compression.max_attempts`` cap (default 3).  Before
+the fix the post-tool path neither checked nor incremented the counter, so a
+long tool loop could compact after every tool response for the lifetime of
+the turn.
+
+These tests drive ``run_conversation()`` through real tool iterations with a
+compressor that always demands compression and assert ``_compress_context``
+fires at most ``max_compression_attempts`` times per turn — no source
+inspection, only observable behavior.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from run_agent import AIAgent
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _tool_call(i: int):
+    return SimpleNamespace(
+        id=f"call_{i}",
+        type="function",
+        function=SimpleNamespace(name="web_search", arguments='{"query": "x"}'),
+    )
+
+
+def _tool_response(i: int):
+    msg = SimpleNamespace(
+        content=None,
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=[_tool_call(i)],
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="tool_calls")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _stop_response():
+    msg = SimpleNamespace(
+        content="done",
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=None,
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _make_tool_defs(*names: str) -> list:
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": n,
+                "description": f"{n} tool",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+        for n in names
+    ]
+
+
+def _pressured_compressor() -> MagicMock:
+    """A compressor that always reports context pressure after tools run.
+
+    ``should_defer_preflight_to_real_usage`` returns True so the turn-start
+    preflight and the pre-API pressure gate stand down — isolating the
+    post-tool gate as the only compression site under test.
+    """
+    compressor = MagicMock()
+    compressor.protect_first_n = 3
+    compressor.protect_last_n = 20
+    compressor.threshold_tokens = 10_000
+    compressor.context_length = 200_000
+    compressor.last_prompt_tokens = 150_000
+    compressor.should_compress.return_value = True
+    compressor.should_defer_preflight_to_real_usage.return_value = True
+    compressor.get_active_compression_failure_cooldown.return_value = None
+    return compressor
+
+
+@pytest.fixture()
+def agent():
+    with (
+        patch("run_agent.get_tool_definitions", return_value=_make_tool_defs("web_search")),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        a = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            max_iterations=10,
+        )
+    a.client = MagicMock()
+    a._cached_system_prompt = "You are helpful."
+    a._use_prompt_caching = False
+    a._disable_streaming = True
+    a.tool_delay = 0
+    a.save_trajectories = False
+    a.compression_enabled = True
+    a.context_compressor = _pressured_compressor()
+    return a
+
+
+def _run_tool_loop(agent, n_tool_iterations: int):
+    """Drive one turn: ``n_tool_iterations`` tool calls, then a stop."""
+    responses = [_tool_response(i) for i in range(n_tool_iterations)]
+    responses.append(_stop_response())
+    agent.client.chat.completions.create.side_effect = responses
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    with (
+        patch.object(agent, "_compress_context", side_effect=_fake_compress),
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+        patch(
+            "run_agent.handle_function_call",
+            lambda name, args, task_id=None, **kwargs: json.dumps({"ok": True}),
+        ),
+    ):
+        result = agent.run_conversation("do a lot of tool work")
+
+    return result, compress_calls
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestPostToolCompressionAttemptCap:
+    def test_post_tool_compression_capped_at_default_three(self, agent):
+        """7 tool iterations under constant pressure → exactly 3 compactions.
+
+        Before the fix the post-tool gate re-fired after every tool response
+        (7 compactions here); the shared per-turn counter caps it at the
+        resolved default of 3.
+        """
+        assert agent.max_compression_attempts == 3  # config default
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=7)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 3, (
+            f"post-tool compression must stop at the per-turn cap (3), "
+            f"got {len(compress_calls)} compactions"
+        )
+
+    def test_post_tool_compression_honors_configured_cap(self, agent):
+        """A raised compression.max_attempts cap lets more rounds run."""
+        agent.max_compression_attempts = 5
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=8)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 5
+
+    def test_post_tool_compression_shares_counter_with_pre_api_gate(self, agent):
+        """Pre-API compactions consume the same per-turn budget.
+
+        Let the pre-API pressure gate fire once (defer disabled for the first
+        check), then keep the pressure on through tool iterations: the
+        combined total must still respect the cap.
+        """
+        # First pre-API check does not defer → pre-API gate fires once;
+        # afterwards defer again so only the post-tool gate keeps firing.
+        defers = iter([False])
+        agent.context_compressor.should_defer_preflight_to_real_usage.side_effect = (
+            lambda _t: next(defers, True)
+        )
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=7)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 3, (
+            "pre-API and post-tool compactions must share one per-turn "
+            f"attempt budget, got {len(compress_calls)} total compactions"
+        )
+
+    def test_cap_is_per_turn_not_per_session(self, agent):
+        """A fresh turn gets a fresh attempt budget."""
+        _result, first = _run_tool_loop(agent, n_tool_iterations=5)
+        agent.client.chat.completions.create.side_effect = None
+        _result, second = _run_tool_loop(agent, n_tool_iterations=5)
+
+        assert len(first) == 3
+        assert len(second) == 3
diff --git a/tests/run_agent/test_preflight_compression_cap_e2e.py b/tests/run_agent/test_preflight_compression_cap_e2e.py
new file mode 100644
index 000000000000..e29b1ecc44f5
--- /dev/null
+++ b/tests/run_agent/test_preflight_compression_cap_e2e.py
@@ -0,0 +1,186 @@
+"""E2E: compression.max_attempts=6 drives a 4th+ preflight compaction pass.
+
+The turn-start preflight loop in ``agent/turn_context.py`` was hardcoded to
+``range(3)``: even when every pass made real progress and the request stayed
+over threshold, the 4th pass never ran, regardless of configuration.  The
+loop now sizes itself from the same resolved ``compression.max_attempts`` cap
+as the conversation loop's compression sites.
+
+This test builds a real ``AIAgent`` from a config with
+``compression.max_attempts: 6`` (the config-driven path through
+``agent_init``), then drives a full ``run_conversation()`` turn in which the
+estimated request size keeps shrinking ~10% per compaction but stays above
+threshold — the exact "progress, but not enough yet" shape that legitimately
+needs more than three rounds.  With cap=6 the preflight must run a 4th pass
+(and ultimately all six).
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _config(max_attempts) -> dict:
+    return {
+        "compression": {
+            "enabled": True,
+            "threshold": 0.50,
+            "target_ratio": 0.20,
+            "protect_first_n": 3,
+            "protect_last_n": 20,
+            "max_attempts": max_attempts,
+        },
+        "prompt_caching": {"cache_ttl": "5m"},
+        "sessions": {},
+        "bedrock": {},
+    }
+
+
+def _stop_response():
+    msg = SimpleNamespace(
+        content="done",
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=None,
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _make_agent(monkeypatch, tmp_path: Path, *, max_attempts) -> AIAgent:
+    from hermes_cli import config as config_mod
+
+    monkeypatch.setattr(
+        config_mod, "load_config", lambda: _config(max_attempts)
+    )
+    db = SessionDB(db_path=tmp_path / "state.db")
+    with (
+        contextlib.redirect_stdout(io.StringIO()),
+        patch("run_agent.get_tool_definitions", return_value=[]),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        agent = AIAgent(
+            base_url="https://openrouter.ai/api/v1",
+            api_key="test-key",
+            model="test/model",
+            enabled_toolsets=[],
+            disabled_toolsets=[],
+            quiet_mode=True,
+            skip_memory=True,
+            skip_context_files=True,
+            session_db=db,
+            session_id="preflight-cap-e2e",
+        )
+    agent.client = MagicMock()
+    agent._cached_system_prompt = "You are helpful."
+    agent._use_prompt_caching = False
+    agent._disable_streaming = True
+    agent.tool_delay = 0
+    agent.save_trajectories = False
+    return agent
+
+
+def test_preflight_runs_fourth_compaction_pass_at_cap_six(monkeypatch, tmp_path):
+    agent = _make_agent(monkeypatch, tmp_path, max_attempts=6)
+    # Config-driven attach seam (agent_init) resolved the raised cap.
+    assert agent.max_compression_attempts == 6
+
+    # Keep the request permanently over threshold while every compaction
+    # makes material (~10% > the 5% progress floor) headway.
+    compressor = agent.context_compressor
+    compressor.threshold_tokens = 50_000
+
+    estimate_state = {"tokens": 1_000_000.0, "calls": 0}
+
+    def _shrinking_estimate(*_args, **_kwargs):
+        if estimate_state["calls"]:
+            estimate_state["tokens"] *= 0.9
+        estimate_state["calls"] += 1
+        return int(estimate_state["tokens"])
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    # 60 messages > protect_first_n + protect_last_n + 1, so the cheap
+    # preflight count gate opens without patching internals.
+    history = [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
+        for i in range(60)
+    ]
+    agent.client.chat.completions.create.return_value = _stop_response()
+
+    with (
+        patch(
+            "agent.turn_context.estimate_request_tokens_rough",
+            side_effect=_shrinking_estimate,
+        ),
+        patch.object(agent, "_compress_context", side_effect=_fake_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)
+
+    assert result["completed"] is True
+    # The old hardcoded range(3) made a 4th pass impossible; cap=6 must
+    # deliver it (and, with steady progress over threshold, all six).
+    assert len(compress_calls) >= 4, (
+        f"expected a 4th preflight compaction pass at cap=6, "
+        f"got {len(compress_calls)} passes"
+    )
+    assert len(compress_calls) == 6
+
+
+def test_preflight_still_stops_at_default_three(monkeypatch, tmp_path):
+    """Unset compression.max_attempts keeps the historical 3-pass behavior."""
+    agent = _make_agent(monkeypatch, tmp_path, max_attempts=None)
+    assert agent.max_compression_attempts == 3
+
+    compressor = agent.context_compressor
+    compressor.threshold_tokens = 50_000
+
+    estimate_state = {"tokens": 1_000_000.0, "calls": 0}
+
+    def _shrinking_estimate(*_args, **_kwargs):
+        if estimate_state["calls"]:
+            estimate_state["tokens"] *= 0.9
+        estimate_state["calls"] += 1
+        return int(estimate_state["tokens"])
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    history = [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
+        for i in range(60)
+    ]
+    agent.client.chat.completions.create.return_value = _stop_response()
+
+    with (
+        patch(
+            "agent.turn_context.estimate_request_tokens_rough",
+            side_effect=_shrinking_estimate,
+        ),
+        patch.object(agent, "_compress_context", side_effect=_fake_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)
+
+    assert result["completed"] is True
+    assert len(compress_calls) == 3