diff --git a/agent/turn_context.py b/agent/turn_context.py
index 3ffaa3f7cd..e250b79cdd 100644
--- a/agent/turn_context.py
+++ b/agent/turn_context.py
@@ -405,6 +405,21 @@ def compression_made_progress(
 _compression_made_progress = compression_made_progress
 
 
+class PreflightCompressionTimedOut(RuntimeError):
+    """Raised when an oversized turn cannot safely finish preflight."""
+
+
+def _fail_closed_after_preflight_timeout(agent, request_tokens: int) -> None:
+    """Stop an oversized turn instead of sending its unchanged provider payload."""
+    if not getattr(agent, "_last_compression_timed_out", False):
+        return
+    raise PreflightCompressionTimedOut(
+        "Context compression timed out before it could commit while the request "
+        f"was still approximately {request_tokens:,} tokens. The provider call "
+        "was not sent. Run /compress and wait for it to finish, then retry."
+    )
+
+
 def _review_fork_first_request_pending(agent: Any) -> bool:
     """Whether a detached review fork has yet to send its first provider request.
 
@@ -1169,6 +1184,7 @@ def build_turn_context(
                 if not _compression_made_progress(
                     _orig_len, len(messages), _orig_tokens, _preflight_tokens
                 ):
+                    _fail_closed_after_preflight_timeout(agent, _preflight_tokens)
                     _preflight_compression_blocked = True
                     break  # Cannot compress further: neither rows nor tokens moved
                 conversation_history = conversation_history_after_compression(
diff --git a/run_agent.py b/run_agent.py
index d056fbd698..9f145baddf 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -8068,6 +8068,11 @@ class AIAgent:
         auto-compress abort.  Auto-compress callers use the default
         ``force=False``.
         """
+        # Per-attempt signal consumed by turn-start preflight. A stalled
+        # compression must not be mistaken for a structural no-op and followed
+        # by the oversized provider request it was meant to prevent.
+        self._last_compression_timed_out = False
+
         from agent.conversation_compression import (
             CompressionCommitFence,
             compress_context,
@@ -8194,6 +8199,7 @@ class AIAgent:
                     timeout_cause["progress_observed"] = progress_observed
 
                 def _on_timeout(idle, waited, since_progress):
+                    self._last_compression_timed_out = True
                     total_exhausted = timeout_cause["total_exhausted"]
                     progress_observed = timeout_cause["progress_observed"]
                     if total_exhausted:
diff --git a/tests/agent/test_compress_context_progress_timeout.py b/tests/agent/test_compress_context_progress_timeout.py
index f0e52986d1..7bed4a32fc 100644
--- a/tests/agent/test_compress_context_progress_timeout.py
+++ b/tests/agent/test_compress_context_progress_timeout.py
@@ -467,6 +467,7 @@ class TestCompressContextForwarderOwnsTimeout:
 
         assert out_msgs is original
         assert out_prompt == "sys"
+        assert agent._last_compression_timed_out is True
         assert calls["n"] == 1
         agent._emit_warning.assert_called_once()
         assert agent.context_compressor._consecutive_timeout_failures == 1
@@ -643,5 +644,6 @@ class TestCompressContextForwarderOwnsTimeout:
             commit_fence=fence,
         )
         assert seen["fence"] is fence
+        assert agent._last_compression_timed_out is False
         assert prompt == "sys"
         assert msgs[0]["content"] == "ok"
diff --git a/tests/agent/test_preflight_compression_timeout_fail_closed.py b/tests/agent/test_preflight_compression_timeout_fail_closed.py
new file mode 100644
index 0000000000..05b9eabc0a
--- /dev/null
+++ b/tests/agent/test_preflight_compression_timeout_fail_closed.py
@@ -0,0 +1,23 @@
+"""Regression coverage for stalled preflight compression."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from agent.turn_context import (
+    PreflightCompressionTimedOut,
+    _fail_closed_after_preflight_timeout,
+)
+
+
+def test_preflight_timeout_blocks_unchanged_provider_payload():
+    agent = SimpleNamespace(_last_compression_timed_out=True)
+
+    with pytest.raises(PreflightCompressionTimedOut, match="provider call was not sent"):
+        _fail_closed_after_preflight_timeout(agent, 190_035)
+
+
+def test_structural_noop_keeps_existing_preflight_behavior():
+    agent = SimpleNamespace(_last_compression_timed_out=False)
+
+    _fail_closed_after_preflight_timeout(agent, 190_035)