@@ -471,8 +471,9 @@ def run(self) -> str | None:
                     # after successful responses, so without this the
                     # interrupted turn would never reach the session
                     # file — the next turn's save would overwrite it
-                    # without ever containing it.
-                    session.auto_save(salvaged, self.system)
+                    # without ever containing it.  flush bypasses the
+                    # debounce so the partial turn is not lost.
+                    session.flush_auto_save(salvaged)
                     # The title is generated on the first save, not
                     # only on clean completion — an interrupted session
                     # still gets a meaningful name (one-shot; no-op when
@@ -484,6 +485,9 @@ def run(self) -> str | None:
                     # no-op when the title already exists or generation
                     # is in flight)
                     session.generate_session_title()
+                    # The per-round auto_save is debounced; flush at run
+                    # end so the final state always reaches the file.
+                    session.flush_auto_save(salvaged)
         return self.result
 
     def _next_state(self) -> str:
 
@@ -69,20 +69,26 @@ class ImagePart:
     path: str | None = None
     url: str | None = None
 
+    def __post_init__(self) -> None:
+        # to_api() runs on every round (token estimation + API payload),
+        # so the base64 data URL is encoded once, not per call.
+        self._data_url: str | None = None
+        if self.data is not None:
+            self._data_url = (
+                f"data:{self.media_type};base64,"
+                f"{base64.b64encode(self.data).decode('ascii')}"
+            )
+
     @classmethod
     def from_url(cls, url: str) -> ImagePart:
         return cls(url=url)
 
     def to_api(self) -> dict[str, Any]:
         if self.url:
             return {"type": "image_url", "image_url": {"url": self.url}}
-        if self.data is None:
+        if self._data_url is None:
             raise ValueError("ImagePart requires either data or url")
-        b64 = base64.b64encode(self.data).decode("ascii")
-        return {
-            "type": "image_url",
-            "image_url": {"url": f"data:{self.media_type};base64,{b64}"},
-        }
+        return {"type": "image_url", "image_url": {"url": self._data_url}}
 
 
 # Type alias for the individual parts a Message.content list may contain.
@@ -155,6 +161,11 @@ class Message:
     injected: bool = False  # harness-injected (nudge/plan/build-switch), not user input
 
     def to_api(self) -> dict[str, Any]:
+        # Not cached: the per-message dict is cheap to rebuild and
+        # caching it (returning the same object each call) would make a
+        # mutation by any caller corrupt later rounds and token
+        # estimation.  The one expensive step — base64-encoding an image
+        # — is cached on ``ImagePart`` instead (see ImagePart.__post_init__).
         d: dict[str, Any] = {"role": self.role}
         if self.content is not None:
             content = self._api_content()
 
@@ -45,7 +45,10 @@ def estimate_tokens(text: str) -> int:
     """Estimate tokens in TEXT: Latin ~4 chars/token, CJK ~2 chars/token."""
     if not text:
         return 0
-    cjk = len(_CJK_RE.findall(text))
+    # subn counts CJK chars in one C-speed pass without allocating the
+    # per-match list findall would build (this runs on the full payload
+    # every round).
+    cjk = _CJK_RE.subn("", text)[1]
     latin = len(text) - cjk
     return round(latin / 4.0 + cjk / 2.0)
 
 
@@ -53,6 +53,14 @@
 SESSION_DIR = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
 SESSION_SUBDIR = "python-agent-harness/sessions"
 AUTO_SAVE_SESSION = True
+# auto_save is called after every round; the full-conversation rewrite is
+# debounced so a long run does not rewrite the whole file each round.
+# A save is skipped while fewer than AUTO_SAVE_MIN_INTERVAL seconds have
+# passed since the last one AND the message count has not grown by
+# AUTO_SAVE_MIN_GROWTH_MSGS.  flush_auto_save() bypasses the debounce
+# (run end, cancel, compaction).
+AUTO_SAVE_MIN_INTERVAL = 2.0
+AUTO_SAVE_MIN_GROWTH_MSGS = 8
 
 # ---- LLM interaction logs ---------------------------------------------------
 LLM_LOG_ENABLED = False
 
@@ -195,6 +195,10 @@ def __init__(
         self.todos: list[dict] = []
         self.pending_user_prompts: list[str] = []
         self._save_error: str | None = None
+        # auto_save debounce state (see config.AUTO_SAVE_*): the last
+        # successful save's wall time and conversation length
+        self._last_save_at: float | None = None
+        self._last_save_len: int = 0
         self.last_messages: list = []
         self.cancel_event = threading.Event()
         # Named LLM profiles for runtime switching via /model command
@@ -680,6 +684,39 @@ def remember_user_text(self, messages: list) -> None:
     def auto_save(self, messages: list, system: str | None) -> None:
         if not config.AUTO_SAVE_SESSION:
             return
+        # Debounce the full-conversation rewrite: auto_save runs after
+        # every round, but the file only needs refreshing when enough
+        # time has passed AND the conversation grew.  The growth check
+        # is on the message count (cheap) before the expensive
+        # _conversation_text serialization.
+        now = time.monotonic()
+        # getattr: test doubles built via __new__ skip __init__ and never
+        # set the debounce state; treat a missing state as "never saved".
+        last = getattr(self, "_last_save_at", None)
+        if last is not None:
+            elapsed = now - last
+            grew = (
+                len(messages) - getattr(self, "_last_save_len", 0)
+                >= config.AUTO_SAVE_MIN_GROWTH_MSGS
+            )
+            # skip while inside the min interval and the conversation has
+            # not grown enough; a big round (grew) or the interval
+            # elapsing forces a save
+            if elapsed < config.AUTO_SAVE_MIN_INTERVAL and not grew:
+                return
+        self._do_save(messages)
+
+    def flush_auto_save(self, messages: list) -> None:
+        """Save immediately, bypassing the auto_save debounce.
+
+        Called at run end / cancel / compaction, where the conversation
+        must reach the session file regardless of the debounce window.
+        """
+        if not config.AUTO_SAVE_SESSION:
+            return
+        self._do_save(messages)
+
+    def _do_save(self, messages: list) -> None:
         text = self._conversation_text(messages)
         # retry once: transient failures (NFS hiccup, brief lock) clear
         # on the second attempt; permanent ones (disk full, read-only)
@@ -689,6 +726,8 @@ def auto_save(self, messages: list, system: str | None) -> None:
             try:
                 self.store.save(text)
                 self._save_error = None
+                self._last_save_at = time.monotonic()
+                self._last_save_len = len(messages)
                 return
             except (OSError, UnicodeError) as e:
                 # UnicodeError too: a save failure must never escape into
@@ -983,7 +1022,7 @@ def compact_conversation(self) -> tuple[bool, str]:
             # preserved verbatim after the frame, so the model keeps
             # the actual requests.
             self.last_messages = compacted_messages(summary, user_prompt_texts(messages))
-            self.auto_save(self.last_messages, self.system_prompt)
+            self.flush_auto_save(self.last_messages)
             self.notify("compact")
             return True, "Session compacted successfully."
         except Exception as e:  # noqa: BLE001 - compaction failure is non-fatal
@@ -1019,7 +1058,7 @@ def summarize_conversation(self) -> str:
         if not summary:
             return "Summary failed: empty response."
         self.last_messages.append(Msg(role="assistant", content=summary))
-        self.auto_save(self.last_messages, self.system_prompt)
+        self.flush_auto_save(self.last_messages)
         return "Summary appended."
 
 
 
@@ -75,6 +75,17 @@ def sanitize_terminal(text: str) -> str:
     return _TERMINAL_CONTROL.sub("", api_safe(text))
 
 
+# Characters sanitize_terminal would change: any C0/C1 control except
+# tab/newline, or a lone surrogate.  A cheap pre-check lets callers skip
+# the full regex pass on the common (clean) case.
+_NEEDS_SANITIZE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f\ud800-\udfff]")
+
+
+def needs_sanitize(text: str) -> bool:
+    """Whether sanitize_terminal would change TEXT (cheap pre-check)."""
+    return _NEEDS_SANITIZE.search(text) is not None
+
+
 def safe_markup(value: object) -> str:
     """VALUE made safe to embed in a Rich markup string.
 
@@ -355,12 +366,24 @@ def _stream_row(self) -> Text | None:
         """Live stream row (cheap Text, tail-capped)."""
         with self.lock:
             stream = self.stream_text
-        stream = _strip_final_check(sanitize_terminal(stream))
-        if not stream:
-            return None
         cap = self._visible_row_cap()
         width = getattr(self.console, "width", None) or 80
         lines = max(3, cap - 3)
+        # Sanitize BEFORE stripping: a control sequence inside the
+        # [FINAL CHECK] header ("[FINAL\x1b[0m CHECK]") would otherwise
+        # defeat the header match and leak the block.  The full-buffer
+        # sanitize is skipped when the text is already clean (the common
+        # case), so the expensive regex pass only runs when needed.
+        if needs_sanitize(stream):
+            stream = sanitize_terminal(stream)
+        # Strip on the FULL buffer: the header can sit arbitrarily far
+        # before the tail, so windowing first would leak the block.
+        stream = _strip_final_check(stream)
+        window = max(lines * max(1, width) * 2, 8192)
+        if len(stream) > window:
+            stream = stream[-window:]
+        if not stream:
+            return None
         preview = _tail_lines(stream, lines)
         preview = _tail_chars(preview, lines * max(1, width))
         row = Text(f"assistant: {preview}", style=ASSISTANT_STYLE)
 
@@ -428,6 +428,42 @@ def test_stream_pure_final_check_hidden(self):
         row = tui._stream_row()
         self.assertIsNone(row)
 
+    def test_stream_final_check_longer_than_window_hidden(self):
+        """A [FINAL CHECK] block whose header sits further back than the
+        sanitize window must still be stripped: the strip runs on the
+        full buffer, only the sanitize is windowed.  Windowing first
+        leaked the block's content (the header fell outside the window)."""
+        tui, _ = make_tui()
+        block = (
+            "[FINAL CHECK]\n"
+            + ("- Goal: " + "x" * 100 + "\n") * 200
+            + "- Status: SUCCESS\n- Evidence: y"
+        )
+        tui.stream_text = "answer text\n\n" + block
+        row = tui._stream_row()
+        self.assertIsNotNone(row)
+        self.assertIn("answer text", row.plain)
+        self.assertNotIn("FINAL CHECK", row.plain)
+        self.assertNotIn("Status: SUCCESS", row.plain)
+        self.assertNotIn("Evidence:", row.plain)
+
+    def test_stream_final_check_with_escape_in_header_hidden(self):
+        """A control sequence inside the [FINAL CHECK] header must not
+        defeat the strip: sanitize runs BEFORE the strip, so the header
+        matches.  Stripping first leaked the block (the escape broke the
+        header regex)."""
+        tui, _ = make_tui()
+        tui.stream_text = (
+            "answer text\n\n[FINAL\x1b[0m CHECK]\n"
+            "- Goal: x\n- Status: SUCCESS\n- Evidence: y"
+        )
+        row = tui._stream_row()
+        self.assertIsNotNone(row)
+        self.assertIn("answer text", row.plain)
+        self.assertNotIn("FINAL CHECK", row.plain)
+        self.assertNotIn("Status: SUCCESS", row.plain)
+        self.assertNotIn("Evidence:", row.plain)
+
     def test_reasoning_streams_normally(self):
         """While streaming, reasoning content shows up live like any
         other text — the collapse only happens in the final history."""