fix: only drop an interrupted partial for the runaway repetition shape

is_repetition_dominated is tuned for the truncated-continuation nudge, where a
false positive merely skips a continuation. At the interrupt checkpoints the
same verdict DROPS the partial from history and tells the model the reply
degenerated, so a legitimately repetitive but correct reply - twelve distinct
INSERT rows sharing a long prefix trip the 60-char window scan - was erased and
mislabelled. Gate the two checkpoints on is_runaway_repetition: dominated AND,
when the text has line structure, at most half of its non-empty lines distinct.
Byte-identical repeated lines (the #112764 shape) still qualify; distinct batch
rows no longer do. The continuation path keeps the looser predicate.

The plain interrupt site now mirrors the redirect placeholder (empty content,
display_kind=hidden, api_content=[response interrupted]) so the bracketed
placeholder no longer surfaces as an assistant bubble in transcript replays.

agent/conversation_loop.py

@@ -28,7 +28,7 @@ from agent.prompt_caching import (
     strip_anthropic_cache_control,
     strip_anthropic_tool_cache_control,
 )
-from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED, is_repetition_dominated
+from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED, is_runaway_repetition
 from agent.runtime_cwd import resolve_agent_cwd
 from agent.surface_switch import (
     identity_line_value, note_inert_pinned_tools, split_runtime_boundary, stage_surface_switch_note,
@@ -289,9 +289,10 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text
     visible = agent._strip_think_blocks(getattr(agent, "_current_streamed_assistant_text", "") or "").strip()
 
     checkpoint_parts = [_INTERRUPT_SCAFFOLD_MARKER]
-    if is_repetition_dominated(visible):
-        # Same guard as the truncated-continuation path: the looped bytes must reach neither the
-        # replayed correction nor the placeholder below (empty ``visible`` takes the hidden shape).
+    if is_runaway_repetition(visible):
+        # Runaway shape only (a correct batch-style partial stays replayable): the looped bytes must
+        # reach neither the replayed correction nor the placeholder below (empty ``visible`` takes
+        # the hidden shape).
         checkpoint_parts.append(REPETITION_LOOP_INTERRUPTED)
         visible = ""
     elif visible:

agent/repetition_guard.py

@@ -28,6 +28,11 @@ _DOMINANCE_RATIO = 0.5
 # to know the reply degenerated and was cut off.
 REPETITION_LOOP_INTERRUPTED = "[the reply degenerated into a repetition loop and was interrupted]"
 
+# ``is_runaway_repetition``: a multi-line partial must be mostly copies of a few lines. Batch-style
+# output (distinct INSERT rows, similar table rows) shares long prefixes and trips the window
+# scan, but every line is distinct; a loop re-emits the same line(s).
+_RUNAWAY_DISTINCT_LINE_RATIO = 0.5
+
 
 def is_repetition_dominated(text: str) -> bool:
     """True when a single 60+ char substring recurs often enough to cover at least half
@@ -61,6 +66,22 @@ def is_repetition_dominated(text: str) -> bool:
     return False
 
 
+def is_runaway_repetition(text: str) -> bool:
+    """Stricter than :func:`is_repetition_dominated`: also require the runaway shape.
+
+    An interrupt checkpoint DROPS the partial when this fires, so a legitimately repetitive but
+    correct reply (distinct batch rows) must not qualify: repeated windows have to dominate AND,
+    when the text has line structure, at most half of its non-empty lines may be distinct.
+    """
+    if not is_repetition_dominated(text):
+        return False
+    lines = [line.strip() for line in text.splitlines()]
+    lines = [line for line in lines if line]
+    if len(lines) < _MIN_REPEAT_COUNT:
+        return True  # no line structure to judge by: a dominated single-line loop
+    return len(set(lines)) <= len(lines) * _RUNAWAY_DISTINCT_LINE_RATIO
+
+
 def _line_repetition_dominated(text: str, n: int) -> bool:
     """True when a single normalized line covers half the fragment via repeats."""
     counts = Counter(norm for norm in (line.strip() for line in text.splitlines()) if norm)

agent/turn_api_call.py

@@ -17,7 +17,7 @@ from typing import Any, Dict, Optional
 from agent.error_classifier import FailoverReason
 from agent.agent_runtime_helpers import _INTERRUPTED_PLACEHOLDER
 from agent.message_metadata import append_message
-from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED, is_repetition_dominated
+from agent.repetition_guard import REPETITION_LOOP_INTERRUPTED, is_runaway_repetition
 from agent.turn_failure_copy import site_copy, stamp_failure
 
 logger = logging.getLogger("agent.conversation_loop")
@@ -187,10 +187,14 @@ def handle_api_interrupt(
     _partial = agent._strip_think_blocks(
         getattr(agent, "_current_streamed_assistant_text", "") or ""
     ).strip()
-    if _partial and is_repetition_dominated(_partial):
+    if _partial and is_runaway_repetition(_partial):
         # The interrupted row is replayed next turn; looped bytes there re-seed the loop
-        # (#112764). Keep the neutral placeholder the sanitizer would heal to anyway.
-        append_message(messages, {"role": "assistant", "content": _INTERRUPTED_PLACEHOLDER})
+        # (#112764). Same hidden shape as the redirect placeholder: nothing visible in the
+        # transcript, a neutral api_content so the pre-call sanitizer does not re-heal it.
+        append_message(messages, {
+            "role": "assistant", "content": "", "display_kind": "hidden",
+            "api_content": _INTERRUPTED_PLACEHOLDER,
+        })
         final_response = REPETITION_LOOP_INTERRUPTED
     elif _partial:
         append_message(messages, {"role": "assistant", "content": _partial})

tests/agent/test_turn_api_call_interrupt.py

@@ -45,11 +45,29 @@ def test_repetition_dominated_partial_is_not_kept_as_the_interrupted_row():
 
     messages, verdict = _interrupt(looped)
 
-    assert (messages[-1]["role"], messages[-1]["content"]) == ("assistant", _INTERRUPTED_PLACEHOLDER)
+    # Same hidden shape as the redirect placeholder: no visible bubble in transcript replays.
+    assert messages[-1]["role"] == "assistant"
+    assert messages[-1]["content"] == ""
+    assert messages[-1]["display_kind"] == "hidden"
+    assert messages[-1]["api_content"] == _INTERRUPTED_PLACEHOLDER
     assert verdict.final_response == REPETITION_LOOP_INTERRUPTED
     assert "I. I. I." not in verdict.final_response
 
 
+def test_distinct_batch_rows_are_not_mistaken_for_a_loop():
+    """Legitimately repetitive output (distinct INSERT rows sharing a long prefix) trips the
+    window scan but is not a runaway loop: the partial must stay the interrupted row and must
+    not be relabelled as a degenerate reply."""
+    rows = "\n".join(
+        f"INSERT INTO users (id, name, email, created_at) VALUES ({i}, 'user{i}', 'user{i}@example.com', NOW());"
+        for i in range(12)
+    )
+    messages, verdict = _interrupt(rows)
+
+    assert (messages[-1]["role"], messages[-1]["content"]) == ("assistant", rows)
+    assert verdict.final_response == rows
+
+
 def test_ordinary_partial_is_kept_as_the_interrupted_row():
     messages, verdict = _interrupt("Visible draft.")