test(telemetry): cover TOIN save failure and pass-exhaustion paths

Two paths introduced by writing outside the lock had no direct coverage:
a backend.save() raising must leave _dirty armed and the save clock
unadvanced so the next record_* trigger retries (structured
toin_save_failed warning asserted), and a mutation racing every pass
must exhaust the bounded passes without finalizing.

tests/test_toin_fixes.py

@@ -11,6 +11,7 @@ This file tests all the fixes made to the TOIN implementation:
 """
 
 import json
+import logging
 import tempfile
 import threading
 import time
@@ -776,3 +777,78 @@ class TestAutoSaveLockScope:
         assert len(writes) == 2
         assert any(signature.structure_hash in key for key in writes[1]["patterns"])
         assert toin._dirty is False
+
+    def test_save_failure_keeps_dirty_state_for_retry(self):
+        class _Capture(logging.Handler):
+            def __init__(self):
+                super().__init__()
+                self.records: list[logging.LogRecord] = []
+
+            def emit(self, record: logging.LogRecord) -> None:
+                self.records.append(record)
+
+        class _FailingBackend:
+            def save(self, data: dict) -> None:
+                raise OSError("disk full")
+
+        toin = ToolIntelligenceNetwork(TOINConfig(auto_save_interval=600))
+        toin._backend = _FailingBackend()
+        toin._dirty = True
+        toin._last_save_time = time.time() - 2 * toin._config.auto_save_interval
+        before = toin._last_save_time
+
+        logger = logging.getLogger("headroom.telemetry.toin")
+        capture = _Capture()
+        logger.addHandler(capture)
+        try:
+            toin._maybe_auto_save()
+        finally:
+            logger.removeHandler(capture)
+
+        # Failed write must not advance the save clock — the next record_*
+        # trigger should retry from a fresh snapshot.
+        assert toin._dirty is True
+        assert toin._last_save_time == before
+        assert any(
+            record.levelno == logging.WARNING
+            and getattr(record, "event", None) == "toin_save_failed"
+            for record in capture.records
+        )
+
+    def test_mutations_on_every_pass_exhaust_and_stay_dirty(self):
+        writes: list[dict] = []
+
+        toin = ToolIntelligenceNetwork(TOINConfig(auto_save_interval=600))
+
+        class _MutatingBackend:
+            """Commits a new pattern during every write, racing each pass."""
+
+            def save(self, data: dict) -> None:
+                writes.append(data)
+                saved_interval = toin._config.auto_save_interval
+                # Silence the trailing auto-save inside the nested commit.
+                toin._config.auto_save_interval = 0
+                try:
+                    items = [{"id": len(writes), "score": 1}]
+                    toin.record_compression(
+                        tool_signature=ToolSignature.from_items(items),
+                        original_count=1,
+                        compressed_count=1,
+                        original_tokens=10,
+                        compressed_tokens=5,
+                        strategy="smart_sample",
+                        items=items,
+                    )
+                finally:
+                    toin._config.auto_save_interval = saved_interval
+
+        toin._backend = _MutatingBackend()
+        toin._dirty = True
+        toin._last_save_time = time.time() - 2 * 600
+
+        toin._maybe_auto_save()
+
+        # Both bounded passes wrote; neither could finalize because every pass
+        # was invalidated mid-write. Dirty stays armed for the next trigger.
+        assert len(writes) == toin._SAVE_PASSES
+        assert toin._dirty is True