fix(memory): degrade gracefully after repeated at-capacity consolidation failures (#42405)

Builds on the zero-match feedback fix (previous commit) to close the silent-hang
symptom: when memory is at capacity, a failed `add`/`replace`/`remove`
consolidation could loop the whole turn to iteration-budget exhaustion and
deliver no user-facing reply.

#41755 turned the at-capacity overflow error into a *commanded* in-turn retry
("...then retry this add — all in this turn"); combined with the fragile
substring-only `replace`/`remove` matching (LLMs can't reliably re-quote a long
entry verbatim), the model loops add↔replace on inexact guesses until the turn
dies. The existing tool_guardrails halt would catch this, but hard_stop_enabled
is opt-in (off by default), so a default install still hangs.

This fixes it at the memory layer without changing global guardrail behavior:
- MemoryStore tracks per-turn consolidation failures; after a cap (3) it drops
  the "retry in this turn" instruction and returns a terminal "leave memory
  unchanged, continue your reply" result, so a failed memory side effect can
  never block the turn's reply.
- The counter resets on any successful write (progress) and at each turn
  boundary (turn_context.reset_consolidation_failures, guarded via getattr so
  plugin memory stores without the method are a no-op).

Co-authored-by: liuhao1024 <sunsky.lau@gmail.com>

tests/tools/test_memory_tool.py

@@ -330,6 +330,10 @@ class TestMemoryStoreReplace:
         store.add("memory", "fact A")
         result = store.replace("memory", "nonexistent", "new")
         assert result["success"] is False
+        assert "No entry matched" in result["error"]
+        # Zero-match must return current entries so the agent can self-correct
+        # instead of looping blindly (#42405, co-author #42417).
+        assert result["current_entries"] == ["fact A"]
 
     def test_replace_ambiguous_match(self, store):
         store.add("memory", "server A runs nginx")
@@ -361,14 +365,119 @@ class TestMemoryStoreRemove:
         assert len(store.memory_entries) == 0
 
     def test_remove_no_match(self, store):
+        store.add("memory", "fact A")
         result = store.remove("memory", "nonexistent")
         assert result["success"] is False
+        assert "No entry matched" in result["error"]
+        # Zero-match must return current entries (#42405, co-author #42417).
+        assert result["current_entries"] == ["fact A"]
 
     def test_remove_empty_old_text(self, store):
         result = store.remove("memory", "  ")
         assert result["success"] is False
 
 
+class TestMemoryConsolidationGracefulDegrade:
+    """Fix #3 for #42405: a failed at-capacity consolidation must never loop the
+    turn to budget exhaustion — after a per-turn cap of failures, memory ops
+    return a terminal 'stop, continue your reply' result instead of the
+    'retry — all in this turn' instruction."""
+
+    def test_zero_match_failures_degrade_after_cap(self, store):
+        store.add("memory", "fact A")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        # First `cap` failures still hand back previews + the self-correct hint.
+        for _ in range(cap):
+            r = store.replace("memory", "nonexistent", "new")
+            assert r["success"] is False
+            assert "current_entries" in r  # actionable feedback, keep trying
+            assert "retry with the exact text" in r["error"]
+        # The next failure degrades: terminal, no retry instruction.
+        r = store.replace("memory", "nonexistent", "new")
+        assert r["success"] is False
+        assert r["done"] is True
+        assert "current_entries" not in r
+        assert "continue with your reply" in r["error"]
+
+    def test_add_overflow_degrades_after_cap(self, store):
+        # Fill near the 500-char user/memory limit so add() overflows.
+        store.add("memory", "x" * 200)
+        store.add("memory", "y" * 200)
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        big = "z" * 200
+        for _ in range(cap):
+            r = store.add("memory", big)
+            assert r["success"] is False
+            assert "retry this add" in r["error"]  # still instructs in-turn retry
+        r = store.add("memory", big)
+        assert r["success"] is False
+        assert r["done"] is True
+        assert "continue with your reply" in r["error"]
+
+    def test_failures_mix_across_actions_share_one_budget(self, store):
+        store.add("memory", "fact A")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        # Interleave replace + remove failures — they share the per-turn counter.
+        actions = [lambda: store.replace("memory", "nope", "x"),
+                   lambda: store.remove("memory", "nope")]
+        for i in range(cap):
+            assert actions[i % 2]()["success"] is False
+        # cap+1th failure (any action) degrades.
+        r = store.remove("memory", "nope")
+        assert "continue with your reply" in r["error"]
+
+    def test_success_resets_failure_budget(self, store):
+        store.add("memory", "real entry")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        for _ in range(cap):
+            store.replace("memory", "nonexistent", "new")
+        # A successful op resets the counter — progress was made.
+        ok = store.replace("memory", "real entry", "updated entry")
+        assert ok["success"] is True
+        # Now a fresh failure is treated as the first again (still actionable).
+        r = store.replace("memory", "nonexistent", "new")
+        assert "current_entries" in r
+        assert "continue with your reply" not in r["error"]
+
+    def test_reset_consolidation_failures_clears_budget(self, store):
+        store.add("memory", "fact A")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        for _ in range(cap + 1):
+            store.replace("memory", "nonexistent", "new")
+        # New turn boundary resets the budget.
+        store.reset_consolidation_failures()
+        r = store.replace("memory", "nonexistent", "new")
+        assert "current_entries" in r  # actionable again, not degraded
+        assert "continue with your reply" not in r["error"]
+
+    def test_apply_batch_failures_count_toward_budget(self, store):
+        """apply_batch is the primary at-capacity consolidation path; its
+        failures must also degrade so a looping batch can't exhaust the turn
+        (#42405 whole-bug-class — sibling call path)."""
+        store.add("memory", "fact A")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        bad_batch = [{"action": "replace", "old_text": "nope", "content": "x"}]
+        for _ in range(cap):
+            r = store.apply_batch("memory", bad_batch)
+            assert r["success"] is False
+            assert "current_entries" in r  # still actionable under cap
+        r = store.apply_batch("memory", bad_batch)
+        assert r["success"] is False
+        assert r["done"] is True
+        assert "continue with your reply" in r["error"]
+
+    def test_apply_batch_and_single_op_share_budget(self, store):
+        """A batch failure followed by single-op failures shares one counter."""
+        store.add("memory", "fact A")
+        cap = store._MAX_CONSOLIDATION_FAILURES_PER_TURN
+        store.apply_batch("memory", [{"action": "remove", "old_text": "nope"}])
+        for _ in range(cap - 1):
+            store.replace("memory", "nope", "x")
+        # cap reached across batch + single ops → next degrades.
+        r = store.replace("memory", "nope", "x")
+        assert "continue with your reply" in r["error"]
+
+
 class TestMemoryStorePersistence:
     def test_save_and_load_roundtrip(self, tmp_path, monkeypatch):
         monkeypatch.setattr("tools.memory_tool.get_memory_dir", lambda: tmp_path)