[eric] agents: a turn that GREW past the trigger breaks, so the valve is not sat out by a big first reading (ENG-418)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U6zrBsUCNzpMBnov3rTVYV

backend/apps/agents/core/fault_injection.py

@@ -26,8 +26,30 @@ KNOWN_FAULTS: Set[str] = {
     "transport_death",   # the CLI's pipe dies, not the provider (ENG-382 respawn-not-rebuild)
     "empty_finish",      # a turn ends with no answer after tool work (ENG-354, ENG-390)
     "dead_lane",         # the router has already given up on the credential (ENG-414 preflight)
+    "cli_context_squeeze",  # a tiny context window, so autocompact thrash is drillable (ENG-418)
 }
 
+# The window `cli_context_squeeze` pretends the model has, and the number is load-bearing.
+#
+# A turn does not start at zero: system prompt plus ~50 tool schemas cost 30,257 tokens on a plain
+# agent session (measured live 2026-08-28). Our compaction trigger is 18% of the window, so below a
+# ~168K window the trigger sits UNDER that floor, every turn starts over it, and
+# `maybe_break_midturn` correctly refuses forever (a rebuild that failed to shrink must run rather
+# than break-loop). A drill there reads as "our valve never fires" about a branch that is inert by
+# design: measured, at a 30,000 window the breaker ran 0 times while the CLI thrashed to death.
+#
+# 250,000 puts the trigger at 45,000, clear of the floor and reachable in a few file reads, so the
+# valve is genuinely eligible and its firing means something.
+CLI_SQUEEZE_WINDOW = 250_000
+
+# What a turn costs before it does anything, measured. Not a limit; the eligibility arithmetic below.
+TURN_BASELINE_TOKENS = 30_257
+
+# Smaller windows are still useful, because they reproduce the CLI's own autocompact thrash on
+# demand. They just cannot say anything about OUR breaker, and a drill must never be able to report
+# that silently.
+VALVE_ELIGIBLE_WINDOW = 170_000
+
 
 def armed(name: str) -> bool:
     """True when this fault was deliberately armed. Never true in a shipped build."""
@@ -38,6 +60,34 @@ def armed(name: str) -> bool:
     return name in (wanted & KNOWN_FAULTS)
 
 
+def squeezed_context_window() -> int:
+    """The pretend window when `cli_context_squeeze` is armed, else 0.
+
+    It has to be ONE number feeding BOTH the CLI's autocompact and our own compaction trigger, or
+    the drill measures a configuration that cannot exist: squeeze only the CLI and it dies at 30K
+    while our valve waits for 180K, so "our valve never engaged" would be an artifact of the
+    harness rather than a finding about the code."""
+    if not armed("cli_context_squeeze"):
+        return 0
+    raw = os.environ.get("OSW_FAULT_CLI_WINDOW", "").strip()
+    n = CLI_SQUEEZE_WINDOW
+    if raw:
+        try:
+            p_n = int(raw)
+        except ValueError:
+            p_n = 0
+        if p_n > 0:
+            n = p_n
+    if n < VALVE_ELIGIBLE_WINDOW:
+        logger.warning(
+            "cli_context_squeeze window %d puts the compaction trigger under a turn's ~%d-token "
+            "baseline, so OUR mid-turn breaker is INELIGIBLE and cannot fire at any input. This "
+            "run can reproduce the CLI's autocompact thrash; it can say nothing about our valve.",
+            n, TURN_BASELINE_TOKENS,
+        )
+    return n
+
+
 P_FIRED: Set[str] = set()
 
 

backend/apps/agents/core/models.py

@@ -196,6 +196,14 @@ class AgentSession(BaseModel):
     # Absolute token ceiling so big-window models don't sit at 650K before marking; the marker fires at the TIGHTER of the pct or this cap, so it's never "just 65%".
     compact_abs_ceiling_tokens: int = 180_000
     compacted_through_msg_id: Optional[str] = None
+    # Where the last mid-turn break happened. The anti-loop: if a rebuild lands back at or above it,
+    # the rebuild did not shrink anything and breaking again would loop forever.
+    last_break_input_tokens: int = 0
+    # Session-lifetime compaction counts, for the fleet question this class has never been able to
+    # answer: how often does a real user's chat hit a compact boundary, and how often do we catch it
+    # first? Per-turn counters die with the turn, so a block envelope carried nothing.
+    cli_compactions: int = 0
+    midturn_breaks: int = 0
     # Aux-LLM distilled summary of the turns dropped by compaction, cached against the cutoff id it was built for; keeps the gist of old history on a rebuild instead of a hard drop.
     compacted_summary: Optional[str] = None
     compacted_summary_through: Optional[str] = None

backend/apps/agents/manager/configure_provider_env.py

@@ -262,6 +262,16 @@ async def configure_provider_env(
                 p_env.setdefault(k, v)
         except Exception as e:
             logger.debug("node trust: skipped for the CLI (%s)", e)
+        # Drill only: shrink the CLI's autocompact window so the thrash class is reproducible in
+        # seconds instead of needing a real 180K-token session. The SAME number scales our own
+        # compaction trigger (context_budget.effective_window), so the drill preserves the
+        # production race between the two rather than inventing one (ENG-418).
+        from backend.apps.agents.core.fault_injection import armed as p_fault_armed
+        from backend.apps.agents.core.fault_injection import squeezed_context_window
+        if p_fault_armed("cli_context_squeeze"):
+            p_squeeze = squeezed_context_window()
+            p_env["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] = str(p_squeeze)
+            logger.warning("[cli-context] DRILL: CLI autocompact window forced to %d", p_squeeze)
 
     # Fault-injection seam: lets a QA harness front the provider with a local proxy (mid-run 401 drills, ENG-302 family). Absent in prod, so every branch's real base URL stands.
     p_base_override = os.environ.get("OPENSWARM_ANTHROPIC_BASE_OVERRIDE")

backend/apps/agents/manager/context_budget.py

@@ -21,6 +21,14 @@ from backend.apps.agents.manager.streaming.state import TurnState
 logger = logging.getLogger(__name__)
 
 
+@typechecked
+def effective_window(session: AgentSession) -> int:
+    """The window every budget here reasons about, so a squeeze drill scales US and the CLI from
+    the same number and the race between the two compactions stays the production one (ENG-418)."""
+    from backend.apps.agents.core.fault_injection import squeezed_context_window
+    return squeezed_context_window() or max(1, session.context_window)
+
+
 @typechecked
 def compact_ceiling_tokens(session: AgentSession) -> int:
     """The absolute ceiling, with a drill override.
@@ -38,6 +46,13 @@ def compact_ceiling_tokens(session: AgentSession) -> int:
             return session.compact_abs_ceiling_tokens
         if override > 0:
             return override
+    from backend.apps.agents.core.fault_injection import squeezed_context_window
+    p_squeeze = squeezed_context_window()
+    if p_squeeze:
+        # Scaled, not replaced: the ceiling is 18% of a real 1M window, and a drill that kept 180K
+        # against a 30K window would put the ceiling six times past the wall it is meant to beat.
+        p_real = max(1, session.context_window)
+        return max(1, int(session.compact_abs_ceiling_tokens * p_squeeze / p_real))
     return session.compact_abs_ceiling_tokens
 
 
@@ -46,12 +61,16 @@ def compact_trigger_tokens(session: AgentSession) -> int:
     """The token count where compaction fires: the TIGHTER of the pct threshold and the
     absolute ceiling (on a 200K window the pct wins at 130K; on a 1M window the ceiling
     wins at 180K, not 650K)."""
-    window = max(1, session.context_window)
+    window = effective_window(session)
     ceiling = compact_ceiling_tokens(session)
     abs_pct = min(1.0, ceiling / window)
     return int(window * min(session.compact_threshold_pct, abs_pct))
 
 
+# How much a turn must ADD before breaking it is worth the rebuild it costs. Same reasoning as the
+# 20K the pre-nudge compaction is gated on: below this, the break spends more than it reclaims.
+MIN_TURN_GROWTH_TOKENS = 20_000
+
 CONTINUATION_PROMPT = (
     "Continue the task exactly where you left off. Your earlier progress in this chat is "
     "summarized above; do not redo completed steps, pick up at the next unfinished one."
@@ -74,29 +93,54 @@ def maybe_break_midturn(session: AgentSession, turn: TurnState, msg_usage: Dict)
     except Exception:
         return False
     if total <= 0:
-        # No usage on this message. On the codex/GPT lane assistant messages NEVER carry usage
-        # (it arrives only on the ResultMessage, at turn end), so this is not a hiccup: the breaker
-        # is inert for that entire session and one giant turn can run to the context ceiling with
-        # nothing watching. A guard may never disable itself in silence, so it names what it just
-        # stopped protecting. Once per turn: this path runs on every assistant message.
-        if not turn.usage_absence_reported:
-            turn.usage_absence_reported = True
-            logger.warning(
-                "[context-break] session %s on model %s sends no per-message usage, so the "
-                "mid-turn context breaker cannot run for it; this turn is unprotected against a "
-                "single-turn context blowout (ENG-391)",
-                getattr(session, "id", "?"), getattr(session, "model", "?"),
-            )
+        # No usage on THIS message, which on the Anthropic lane is ordinary: mid-stream assistant
+        # messages carry output usage only, and a later one carries the real input count. Reporting
+        # here claimed the whole turn was unprotected on turns that were fine, and a liveness signal
+        # that cries wolf hides the case it exists for. The honest claim is only available at turn
+        # end, so report_usage_liveness() makes it there (ENG-391, corrected ENG-418).
         return False
     # Keep the session's counter honest mid-turn: a broken turn never gets its ResultMessage accounting, and the next pre-send guard reads this.
+    turn.saw_usable_usage = True
     session.tokens["input"] = total
     turn.last_step_input = total
-    if total < compact_trigger_tokens(session):
+    p_trigger = compact_trigger_tokens(session)
+    # Once per turn, say the guard is watching and what it is watching for. "The valve did not fire"
+    # has three indistinguishable causes from outside (no usage at all, never eligible because the
+    # turn started over the trigger, or simply never crossed), and each is a different bug. One line
+    # per turn separates them, and it is the liveness signal this class has been missing.
+    if not turn.usage_seen_reported:
+        turn.usage_seen_reported = True
+        turn.first_input_reading = total
+        logger.info(
+            "[context-break] session %s: watching, first reading %d against trigger %d (window %d)",
+            getattr(session, "id", "?"), total, p_trigger, effective_window(session),
+        )
+    if total < p_trigger:
         turn.saw_input_below_trigger = True
         return False
-    if turn.context_break_fired or not turn.saw_input_below_trigger:
+    if turn.context_break_fired:
+        return False
+    # A turn is breakable when it CROSSED the trigger, or when it has GROWN materially past where it
+    # started. The second half is the one that was missing, and it is not an edge case: measured live
+    # 2026-08-28, the first usage reading a turn ever delivers was 94,404 against a 45,000 trigger,
+    # so `saw_input_below_trigger` was never set and the breaker sat out the entire turn. In
+    # production that is every long chat and every resumed session near its ceiling -- exactly the
+    # 925K/1M blowout with no compact boundary that this guard was written for (ENG-418).
+    p_grew = total - turn.first_input_reading >= MIN_TURN_GROWTH_TOKENS
+    if not (turn.saw_input_below_trigger or p_grew):
+        return False
+    # The anti-loop, and the reason growth is safe to act on: a rebuild that failed to shrink lands
+    # back at or above the last break, and breaking it again would rebuild forever. It must RUN.
+    if session.last_break_input_tokens and turn.first_input_reading >= session.last_break_input_tokens:
+        logger.warning(
+            "[context-break] session %s: the last break rebuilt to %d, no smaller than the %d it "
+            "broke at, so this turn runs unbroken rather than looping",
+            getattr(session, "id", "?"), turn.first_input_reading, session.last_break_input_tokens,
+        )
         return False
     turn.context_break_fired = True
+    session.midturn_breaks += 1
+    session.last_break_input_tokens = total
     maybe_compact(session, force=True)
     session.needs_fresh_session = True
     session.pending_continuation = True
@@ -104,6 +148,26 @@ def maybe_break_midturn(session: AgentSession, turn: TurnState, msg_usage: Dict)
     return True
 
 
+@typechecked
+def report_usage_liveness(session: AgentSession, turn: TurnState) -> bool:
+    """At turn end, say out loud if the breaker never had a number to work with.
+
+    On the codex/GPT lane assistant messages NEVER carry usage (it arrives only on the
+    ResultMessage), so the breaker is inert for that whole session and one giant turn can run to
+    the context ceiling with nothing watching. A guard may not disable itself in silence, and this
+    is the only point where "never" is a fact rather than a guess."""
+    if turn.saw_usable_usage or turn.usage_absence_reported:
+        return False
+    turn.usage_absence_reported = True
+    logger.warning(
+        "[context-break] session %s on model %s sent no per-message usage for the WHOLE turn, so "
+        "the mid-turn context breaker never ran; that turn was unprotected against a single-turn "
+        "context blowout (ENG-391)",
+        getattr(session, "id", "?"), getattr(session, "model", "?"),
+    )
+    return True
+
+
 @typechecked
 def maybe_compact(session: AgentSession, force: bool = False) -> bool:
     """Mark history for compaction when ctx_used_pct >= compact_threshold_pct (or force).

backend/apps/agents/manager/run/TurnRunner.py

@@ -146,6 +146,7 @@ class TurnRunner(AgentManagerProtocol):
                         note_core_mcp_health(session, session_id, raw)
                     if p_subtype == "compact_boundary":
                         turn.compact_boundaries += 1
+                        session.cli_compactions += 1
                     elif p_subtype == "api_retry":
                         note_provider_retry(session_id, raw, turn)
 

backend/apps/agents/manager/run/handle_run_error.py

@@ -104,6 +104,10 @@ def p_report_model_error(subkind: str, session_id: str, session: AgentSession, t
             "input_tokens": int((session.tokens or {}).get("input", 0) or 0),
             "tool_calls": count_tool_calls(session),
             "compacted": bool(session.needs_fresh_session),
+            # `compacted` is a pending-rebuild flag, not a history. These two are the history: how
+            # many times the CLI compacted this chat, and how many times we broke the turn first.
+            "cli_compactions": int(getattr(session, "cli_compactions", 0) or 0),
+            "midturn_breaks": int(getattr(session, "midturn_breaks", 0) or 0),
             "history_prefix_sent": session.history_prefix_sent,
             "delegated": p_used_delegation(session),
         })

backend/apps/agents/manager/streaming/handle_result_message.py

@@ -73,6 +73,14 @@ async def handle_result_message(
     api_type: Optional[str],
     global_settings: object,
 ) -> None:
+    # The turn is over, so "the breaker never had a number" is now a fact rather than a guess.
+    # Reporting it per-message cried wolf on every healthy Anthropic turn (ENG-391 / ENG-418).
+    try:
+        from backend.apps.agents.manager.context_budget import report_usage_liveness
+        report_usage_liveness(session, turn)
+    except Exception:
+        logging.getLogger(__name__).debug("usage liveness report skipped", exc_info=True)
+
     # ResultMessage carries the AUTHORITATIVE per-turn output_tokens count. Some providers (notably OpenAI/Gemini through 9Router) only populate `usage.output_tokens` here, not on individual AssistantMessages. Fold this into the running turn aggregate BEFORE emitting the final consolidated thinking message, so the bubble's tokens segment reflects ground truth on those providers too.
     try:
         result_usage = getattr(message, "usage", None) or {}

backend/apps/agents/manager/streaming/state.py

@@ -69,5 +69,8 @@ class TurnState(BaseModel):
     # Said once per turn when the provider sends no usage at all, so the breaker being
     # structurally inert on that lane is visible instead of silent (ENG-391).
     usage_absence_reported: bool = False
+    saw_usable_usage: bool = False
+    usage_seen_reported: bool = False
+    first_input_reading: int = 0
     # The LAST inference step's request size (input + cache read + cache creation): the true live context. The ResultMessage's usage sums these across every step of the turn, which is billing, not context.
     last_step_input: int = 0

backend/tests/test_cli_context_squeeze.py

@@ -0,0 +1,160 @@
+"""ENG-418: the autocompact-thrash class is drillable, and our own valve is proven to fire on it.
+
+Before this, the mid-turn breaker could only be reached by paying for a genuine 180K-token turn.
+Three attempts to fire it live cost real money and failed. A guard that is never executed is
+indistinguishable from one that was never needed, so the drill is the point, not a convenience.
+"""
+
+import pytest
+
+from backend.apps.agents.core import fault_injection as fi
+from backend.apps.agents.core.models import AgentSession
+from backend.apps.agents.manager.context_budget import (
+    compact_ceiling_tokens,
+    compact_trigger_tokens,
+    effective_window,
+    maybe_break_midturn,
+)
+from backend.apps.agents.manager.streaming.state import TurnState
+
+
+@pytest.fixture(autouse=True)
+def p_clean(monkeypatch):
+    monkeypatch.delenv("OSW_FAULT", raising=False)
+    monkeypatch.delenv("OSW_FAULT_CLI_WINDOW", raising=False)
+    monkeypatch.delenv("OSW_COMPACT_CEILING_TOKENS", raising=False)
+
+
+def p_session() -> AgentSession:
+    s = AgentSession(name="t", model="opus-5")
+    s.context_window = 1_000_000
+    return s
+
+
+def test_unset_is_exactly_todays_behaviour():
+    s = p_session()
+    assert fi.squeezed_context_window() == 0
+    assert effective_window(s) == 1_000_000
+    assert compact_trigger_tokens(s) == 180_000
+
+
+def test_the_squeeze_scales_us_and_the_cli_from_ONE_number(monkeypatch):
+    """The trap this exists to avoid: squeeze only the CLI and it dies at 30K while our valve waits
+    for 180K, so "the valve never engaged" would be an artifact of the harness. Ratio preserved."""
+    s = p_session()
+    p_real_ratio = compact_trigger_tokens(s) / effective_window(s)
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    assert effective_window(s) == fi.CLI_SQUEEZE_WINDOW
+    assert compact_trigger_tokens(s) / effective_window(s) == pytest.approx(p_real_ratio)
+    assert compact_ceiling_tokens(s) == 45_000
+
+
+def test_the_cli_gets_the_same_number_this_process_reasons_about(monkeypatch):
+    """Two places deriving the same budget from different inputs is how they drift; assert they
+    agree by construction, and that the env var is the one the CLI actually reads."""
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    monkeypatch.setenv("OSW_FAULT_CLI_WINDOW", "450000")
+    src = open("backend/apps/agents/manager/configure_provider_env.py", encoding="utf-8").read()
+    assert 'p_env["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] = str(p_squeeze)' in src
+    assert fi.squeezed_context_window() == 450_000
+    assert effective_window(p_session()) == 450_000
+
+
+def test_the_valve_actually_FIRES_under_the_squeeze(monkeypatch):
+    """The liveness assertion the issue asks for. Not "it does not crash": it fires, on usage a
+    squeezed session really produces, and it arms the continuation that saves the work."""
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    s = p_session()
+    t = TurnState()
+    assert maybe_break_midturn(s, t, {"input_tokens": 35_000}) is False, "must see a below-trigger step first"
+    assert t.saw_input_below_trigger
+    assert maybe_break_midturn(s, t, {"input_tokens": 60_000}) is True, "the valve did not engage"
+    assert s.pending_continuation and s.needs_fresh_session
+    assert s.pending_continuation_prompt, "a break with no continuation loses the work silently"
+    assert t.context_break_fired
+    assert maybe_break_midturn(s, t, {"input_tokens": 90_000}) is False, "it must fire once per turn"
+
+
+def test_without_the_squeeze_the_same_usage_does_nothing(monkeypatch):
+    """The negative control, inline: 60,000 tokens is a rounding error against a real 1M window, so
+    a valve that fired here would be firing on nothing."""
+    s = p_session()
+    t = TurnState()
+    maybe_break_midturn(s, t, {"input_tokens": 35_000})
+    assert maybe_break_midturn(s, t, {"input_tokens": 60_000}) is False
+    assert not s.pending_continuation
+
+
+def test_a_junk_window_falls_back_to_the_documented_default(monkeypatch):
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    for junk in ("abc", "-5", "0", ""):
+        monkeypatch.setenv("OSW_FAULT_CLI_WINDOW", junk)
+        assert fi.squeezed_context_window() == fi.CLI_SQUEEZE_WINDOW
+
+
+def test_it_is_a_declared_fault_so_a_typo_cannot_arm_nothing_quietly(monkeypatch):
+    assert "cli_context_squeeze" in fi.KNOWN_FAULTS
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeez")
+    assert fi.unknown_faults() == {"cli_context_squeez"}
+    assert fi.squeezed_context_window() == 0
+
+
+def test_a_valve_ineligible_window_still_works_but_SAYS_SO(monkeypatch):
+    """The trap that cost this drill two runs. At 30,000 the trigger is 5,400 against a ~30,257
+    baseline, so every turn starts over it and the breaker correctly refuses forever; measured
+    live, it ran 0 times while the CLI thrashed to death. That window is still worth having (it
+    reproduces the CLI's thrash on demand), so it is allowed and announced, never silent."""
+    import logging
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    monkeypatch.setenv("OSW_FAULT_CLI_WINDOW", "30000")
+    p_seen = []
+    h = logging.Handler()
+    h.emit = lambda r: p_seen.append(r.getMessage())
+    fi.logger.addHandler(h)
+    try:
+        assert fi.squeezed_context_window() == 30_000, "the small window is usable, not refused"
+    finally:
+        fi.logger.removeHandler(h)
+    assert any("INELIGIBLE" in m for m in p_seen), "a drill that cannot test the valve must say so"
+
+
+def test_the_default_window_leaves_the_valve_eligible():
+    """The arithmetic, asserted rather than trusted: trigger must clear a real turn's baseline."""
+    import os
+    os.environ["OSW_FAULT"] = "cli_context_squeeze"
+    try:
+        assert compact_trigger_tokens(p_session()) > fi.TURN_BASELINE_TOKENS
+        assert fi.CLI_SQUEEZE_WINDOW >= fi.VALVE_ELIGIBLE_WINDOW
+    finally:
+        os.environ.pop("OSW_FAULT", None)
+
+
+
+
+
+def test_the_envelope_carries_the_compaction_HISTORY_not_just_a_pending_flag():
+    """`compacted` is bool(needs_fresh_session), a pending-rebuild flag that says nothing about how
+    often this chat has actually compacted. The fleet question ENG-418 asks needs the counts."""
+    src = open("backend/apps/agents/manager/run/handle_run_error.py", encoding="utf-8").read()
+    assert '"cli_compactions"' in src and '"midturn_breaks"' in src
+
+
+def test_both_counters_survive_the_turn_that_incremented_them(monkeypatch):
+    """Per-turn counters die with the turn, which is why a block envelope carried nothing."""
+    monkeypatch.setenv("OSW_FAULT", "cli_context_squeeze")
+    s = p_session()
+    assert s.midturn_breaks == 0
+    t = TurnState()
+    maybe_break_midturn(s, t, {"input_tokens": 35_000})
+    assert maybe_break_midturn(s, t, {"input_tokens": 60_000}) is True
+    assert s.midturn_breaks == 1
+    t2 = TurnState()
+    maybe_break_midturn(s, t2, {"input_tokens": 20_000})
+    assert maybe_break_midturn(s, t2, {"input_tokens": 60_000}) is True
+    assert s.midturn_breaks == 2, "the count is the session's, not the turn's"
+
+
+def test_the_cli_side_counter_is_wired_to_the_boundary_event():
+    src = open("backend/apps/agents/manager/run/TurnRunner.py", encoding="utf-8").read()
+    i = src.index('if p_subtype == "compact_boundary":')
+    assert "session.cli_compactions += 1" in src[i:i + 200]

backend/tests/test_context_budget.py

@@ -210,15 +210,55 @@ def test_midturn_break_fires_once_per_turn():
     assert cb.maybe_break_midturn(s, t, p_usage(300_000)) is False
 
 
-def test_turn_already_over_trigger_at_start_never_breaks():
-    # A rebuild that failed to shrink must RUN, not break-loop forever.
+def test_a_turn_that_starts_high_and_does_not_grow_is_left_alone():
+    """A rebuild that failed to shrink must RUN. Growth, not an absolute reading, is what makes a
+    turn this guard's business."""
     s = p_session_with(messages=10, input_tokens=0, context_window=1_000_000)
     t = TurnState()
     assert cb.maybe_break_midturn(s, t, p_usage(500_000)) is False
-    assert cb.maybe_break_midturn(s, t, p_usage(600_000)) is False
+    assert cb.maybe_break_midturn(s, t, p_usage(505_000)) is False, "5K is not material growth"
     assert t.context_break_fired is False
 
 
+def test_a_turn_that_starts_high_and_GROWS_does_break():
+    """The hole this closed. Measured live: the first usage reading a turn delivers was 94,404
+    against a 45,000 trigger, so the old below-trigger-first rule sat the guard out for the whole
+    turn -- which in production is every long chat and every resumed session near its ceiling."""
+    s = p_session_with(messages=10, input_tokens=0, context_window=1_000_000)
+    t = TurnState()
+    assert cb.maybe_break_midturn(s, t, p_usage(500_000)) is False
+    assert cb.maybe_break_midturn(s, t, p_usage(600_000)) is True
+    assert s.pending_continuation and s.needs_fresh_session
+    assert s.last_break_input_tokens == 600_000
+
+
+def test_a_rebuild_that_did_not_shrink_cannot_break_LOOP():
+    """The anti-loop, at the session level where it belongs: break once, and if the rebuild lands
+    back at or above where we broke, the next turn RUNS instead of rebuilding forever."""
+    s = p_session_with(messages=10, input_tokens=0, context_window=1_000_000)
+    t = TurnState()
+    cb.maybe_break_midturn(s, t, p_usage(500_000))
+    assert cb.maybe_break_midturn(s, t, p_usage(600_000)) is True
+
+    t2 = TurnState()   # the rebuilt turn, no smaller than the break
+    assert cb.maybe_break_midturn(s, t2, p_usage(600_000)) is False
+    assert cb.maybe_break_midturn(s, t2, p_usage(700_000)) is False
+    assert t2.context_break_fired is False
+
+
+def test_a_rebuild_that_DID_shrink_is_protected_again():
+    """The other direction, which is the half that is easy to lose: the anti-loop must not become a
+    one-break-per-session cap."""
+    s = p_session_with(messages=10, input_tokens=0, context_window=1_000_000)
+    t = TurnState()
+    cb.maybe_break_midturn(s, t, p_usage(500_000))
+    assert cb.maybe_break_midturn(s, t, p_usage(600_000)) is True
+
+    t2 = TurnState()
+    assert cb.maybe_break_midturn(s, t2, p_usage(40_000)) is False
+    assert cb.maybe_break_midturn(s, t2, p_usage(300_000)) is True, "a shrunk rebuild is breakable again"
+
+
 def test_midturn_break_zero_or_garbage_usage_is_inert():
     s = p_session_with(messages=10, input_tokens=7, context_window=1_000_000)
     t = TurnState()

backend/tests/test_fault_injection.py

@@ -61,6 +61,7 @@ WIRED_IN = {
     "transport_death": TURN_RUNNER,
     "empty_finish": "backend/apps/agents/manager/streaming/handle_assistant_message.py",
     "dead_lane": "backend/apps/agents/manager/run/lane_preflight.py",
+    "cli_context_squeeze": "backend/apps/agents/manager/configure_provider_env.py",
 }
 
 

backend/tests/test_midturn_breaker_liveness.py

@@ -71,25 +71,53 @@ def test_a_turn_that_STARTS_over_the_trigger_is_left_alone():
 
 
 def test_no_usage_at_all_is_ANNOUNCED_not_swallowed(p_logs):
-    """The GPT lane. It cannot run; it must say whose session it just stopped protecting."""
+    """The GPT lane. It cannot run; it must say whose session it just stopped protecting.
+
+    Announced at TURN END, because that is the only point where "never" is a fact. Reporting it
+    from the first usage-less message claimed the whole turn was unprotected on healthy Anthropic
+    turns too, and a liveness signal that cries wolf hides the case it exists for (ENG-418)."""
     s, t = p_session(), TurnState()
-    assert p_cb.maybe_break_midturn(s, t, {}) is False
+    for _ in range(12):
+        assert p_cb.maybe_break_midturn(s, t, {}) is False
+    assert not [r for r in p_logs if "never ran" in r.getMessage()], "not before the turn ends"
+    assert p_cb.report_usage_liveness(s, t) is True
     said = " ".join(r.getMessage() for r in p_logs)
-    assert "cannot run" in said
+    assert "never ran" in said
     assert "s-gpt" in said and "cx/gpt-5.6" in said, "name the session and the lane, not just the class"
     assert "ENG-391" in said
 
 
 def test_the_announcement_is_once_per_turn_not_per_message(p_logs):
-    """It runs on EVERY assistant message; a per-message warning would be its own bug."""
+    """A turn end can be reached more than once on a retry; one warning per turn, not per pass."""
     s, t = p_session(), TurnState()
     for _ in range(12):
         p_cb.maybe_break_midturn(s, t, {})
-    assert len([r for r in p_logs if "cannot run" in r.getMessage()]) == 1
+    for _ in range(3):
+        p_cb.report_usage_liveness(s, t)
+    assert len([r for r in p_logs if "never ran" in r.getMessage()]) == 1
+
+
+def test_one_usage_less_message_among_good_ones_says_NOTHING(p_logs):
+    """The false alarm this replaced. Observed live on sonnet-5: the same turn logged "cannot run"
+    AND recorded input=103,436, which is only written when usage IS present. Mid-stream assistant
+    messages carrying output-only usage are ordinary on the Anthropic lane."""
+    s, t = p_session(), TurnState()
+    p_cb.maybe_break_midturn(s, t, {})
+    p_cb.maybe_break_midturn(s, t, {"input_tokens": 500})
+    p_cb.maybe_break_midturn(s, t, {})
+    assert p_cb.report_usage_liveness(s, t) is False
+    assert not [r for r in p_logs if "never ran" in r.getMessage()]
+
+
+def test_the_turn_end_report_is_actually_wired_to_the_turn_end():
+    """A liveness report nobody calls is the very shape this file exists to prevent."""
+    src = open("backend/apps/agents/manager/streaming/handle_result_message.py", encoding="utf-8").read()
+    assert "report_usage_liveness(session, turn)" in src
 
 
 def test_a_lane_WITH_usage_never_triggers_the_warning(p_logs):
     """The innocent case: Anthropic sends usage, so nothing is inert and nothing should be said."""
     s, t = p_session(), TurnState()
     p_cb.maybe_break_midturn(s, t, {"input_tokens": 500})
-    assert not [r for r in p_logs if "cannot run" in r.getMessage()]
+    p_cb.report_usage_liveness(s, t)
+    assert not [r for r in p_logs if "never ran" in r.getMessage()]