perf(streaming): skip redundant sidecar rewrites during a turn (#7172)

The periodic checkpoint thread fires whenever a tool call completes and
unconditionally rewrites the whole session sidecar. A completed tool call
is not proof that anything the checkpoint persists actually changed:
`run_conversation()` mutates an internal copy of the transcript, and the
turn bookkeeping fields (`active_stream_id`, `pending_user_message`,
`pending_started_at`) are written once before the run starts. So the
recurring rewrite normally re-persists byte-identical state.

Each rewrite serializes the entire session while holding the GIL, so its
cost scales with transcript size. Because the HTTP server and the agent
workers share one interpreter, that cost is added latency for every
concurrent request — not just for the session being written.

Measured on this deployment (1,622 sidecars, 6.65 GB, median 2.31 MB,
28.8% above 5 MB, max 35.9 MB):

  sidecar    write    per 10-min turn (40 wakeups)
  1.8 MB    ~100 ms    4.1 s  ->  0.2 s
  24.7 MB   ~1.4 s    41.6 s  ->  2.1 s

A checkpoint now writes only when the state it persists actually moved,
compared through a cheap fingerprint that allocates nothing. The
fingerprint covers every field the checkpoint can legitimately advance
mid-run: session identity (compression rotates the SID), turn
bookkeeping, and the length of the persisted message projections.

Fail-closed on both sides: an unreadable fingerprint returns None and
forces the write, and a 5-minute idle refresh keeps `updated_at` from
going stale on a long quiet turn (the sidebar sorts on it). Crash
recovery is unchanged — any real mutation is still persisted within one
15 s interval.

Co-authored-by: Anthony Ruiz <ruizanthony@users.noreply.github.com>

api/streaming.py

@@ -6473,6 +6473,51 @@ def _save_streaming_checkpoint(session):
         session.save(skip_index=True)
 
 
+# A checkpoint rewrite re-serializes the whole session (messages included) while
+# holding the GIL, so its cost grows with transcript size: ~100 ms for a 1.8 MB
+# sidecar, ~300 ms for a 36 MB one. Because the HTTP server and the agent workers
+# share one interpreter, every one of those milliseconds is added latency for all
+# concurrent requests. Skip the rewrite when it would persist byte-identical state.
+_CHECKPOINT_IDLE_REFRESH_SECONDS = 300
+
+
+def _streaming_checkpoint_fingerprint(session):
+    """Return a cheap fingerprint of the state a streaming checkpoint persists.
+
+    The periodic checkpoint fires on tool completion, but a completed tool call
+    does not by itself mutate the session: ``run_conversation()`` works on an
+    internal copy of ``messages`` and the turn bookkeeping fields are written
+    once before the run starts. So the recurring rewrite normally re-persists
+    state that is already on disk.
+
+    This fingerprint covers every field the checkpoint can legitimately advance
+    mid-run — session identity (compression rotation), turn bookkeeping, and the
+    length of the persisted message projections — without serializing anything.
+    Comparing it lets the checkpoint thread skip redundant work while still
+    writing within one interval of any real change.
+
+    Fails closed: if a field cannot be read, return ``None`` so the caller keeps
+    the unconditional write.
+    """
+    try:
+        return (
+            getattr(session, 'session_id', None),
+            getattr(session, 'parent_session_id', None),
+            getattr(session, 'profile', None),
+            getattr(session, 'active_stream_id', None),
+            getattr(session, 'pending_user_message', None),
+            getattr(session, 'pending_started_at', None),
+            getattr(session, 'pending_user_source', None),
+            len(getattr(session, 'pending_attachments', None) or ()),
+            len(getattr(session, 'messages', None) or ()),
+            len(getattr(session, 'context_messages', None) or ()),
+            getattr(session, 'compression_anchor_visible_idx', None),
+            bool(getattr(session, 'pre_compression_snapshot', None)),
+        )
+    except Exception:  # pragma: no cover - defensive: never block a checkpoint
+        return None
+
+
 def _normalize_fresh_chat_text(text):
     text = _strip_workspace_prefix(str(text or ''), include_legacy=True)
     text = re.sub(r"\s+", " ", text).strip().lower()
@@ -10088,12 +10133,33 @@ def _run_agent_streaming(
 
             def _periodic_checkpoint():
                 last_saved_activity = 0
+                last_fingerprint = None
+                last_write_at = 0.0
                 while not _checkpoint_stop.wait(15):
                     try:
                         cur = _checkpoint_activity[0]
                         if cur > last_saved_activity:
                             with _agent_lock:
-                                _save_streaming_checkpoint(s)
+                                fingerprint = _streaming_checkpoint_fingerprint(s)
+                                # A completed tool call is the trigger, but not
+                                # proof that anything the checkpoint persists
+                                # actually changed. Rewriting a multi-megabyte
+                                # sidecar to re-persist identical bytes stalls
+                                # every concurrent HTTP request behind the GIL,
+                                # so only write when the persisted state moved.
+                                # Fail closed: an unreadable fingerprint (None)
+                                # always writes, and a periodic refresh keeps
+                                # updated_at from going stale on a long turn.
+                                now = time.time()
+                                stale = (now - last_write_at) >= _CHECKPOINT_IDLE_REFRESH_SECONDS
+                                if (
+                                    fingerprint is None
+                                    or fingerprint != last_fingerprint
+                                    or stale
+                                ):
+                                    _save_streaming_checkpoint(s)
+                                    last_fingerprint = fingerprint
+                                    last_write_at = now
                             last_saved_activity = cur
                     except Exception as e:
                         logger.debug("Periodic checkpoint save failed: %s", e)

tests/test_checkpoint_skips_redundant_sidecar_rewrite.py

@@ -0,0 +1,138 @@
+"""Regression coverage: the periodic streaming checkpoint must not rewrite
+byte-identical sidecars.
+
+The checkpoint thread (#765) fires whenever a tool call completes. A completed
+tool call is *not* proof that anything the checkpoint persists has changed:
+``run_conversation()`` mutates an internal copy of the transcript, and the turn
+bookkeeping fields are written once before the run starts.
+
+Each rewrite re-serializes the whole session while holding the GIL. Because the
+HTTP server and the agent workers share one interpreter, that cost is added
+latency for every concurrent request — ~100 ms for a 1.8 MB sidecar and ~300 ms
+for a 36 MB one, repeated every 15 s for the whole turn.
+
+These tests pin the behavior at the decision level (the fingerprint) so they do
+not depend on wall-clock timing of the background thread.
+"""
+from __future__ import annotations
+
+
+def _session(tmp_path, **overrides):
+    from api.models import Session
+
+    s = Session(session_id="ckpt_redundant_1")
+    s.messages = [{"role": "assistant", "content": "hello"}]
+    s.context_messages = []
+    s.pending_user_message = "do the thing"
+    s.pending_started_at = 1787000000.0
+    s.active_stream_id = "streamabc"
+    for key, value in overrides.items():
+        setattr(s, key, value)
+    return s
+
+
+def test_fingerprint_is_stable_when_nothing_changes(tmp_path):
+    """Two reads of an untouched session produce the same fingerprint.
+
+    This is the redundant case: the checkpoint would rewrite identical bytes.
+    """
+    from api.streaming import _streaming_checkpoint_fingerprint
+
+    s = _session(tmp_path)
+    assert _streaming_checkpoint_fingerprint(s) == _streaming_checkpoint_fingerprint(s)
+
+
+def test_fingerprint_tracks_every_field_the_checkpoint_persists(tmp_path):
+    """Any state a mid-run checkpoint can legitimately advance must be detected.
+
+    Missing one of these would make the skip lose real data, so each field is
+    asserted individually rather than as a group.
+    """
+    from api.streaming import _streaming_checkpoint_fingerprint
+
+    base = _session(tmp_path)
+    reference = _streaming_checkpoint_fingerprint(base)
+
+    mutations = {
+        # compression rotates the session id mid-turn
+        "session_id": "rotated_sid",
+        "parent_session_id": "parent_sid",
+        "profile": "other-profile",
+        # turn bookkeeping cleared at the end of the turn
+        "active_stream_id": None,
+        "pending_user_message": None,
+        "pending_started_at": None,
+        "pending_user_source": "telegram",
+        # transcript growth (error rows, continuation tail, compression)
+        "messages": [{"role": "assistant", "content": "hello"}, {"role": "user", "content": "more"}],
+        "context_messages": [{"role": "user", "content": "ctx"}],
+        "compression_anchor_visible_idx": 3,
+        "pre_compression_snapshot": {"messages": []},
+        "pending_attachments": [{"name": "f.png"}],
+    }
+
+    for field, value in mutations.items():
+        mutated = _session(tmp_path, **{field: value})
+        assert _streaming_checkpoint_fingerprint(mutated) != reference, (
+            f"a change to {field!r} must force a checkpoint write"
+        )
+
+
+def test_fingerprint_fails_closed_on_unreadable_session():
+    """An unreadable session yields None so the caller keeps writing.
+
+    Fail-closed: never skip a checkpoint on uncertainty.
+    """
+    from api.streaming import _streaming_checkpoint_fingerprint
+
+    class Hostile:
+        def __getattr__(self, name):
+            raise RuntimeError("boom")
+
+    assert _streaming_checkpoint_fingerprint(Hostile()) is None
+
+
+def test_checkpoint_loop_skips_redundant_writes_but_persists_changes(monkeypatch, tmp_path):
+    """End-to-end on the real loop body: identical state writes once.
+
+    Drives the actual decision sequence used by the checkpoint thread: an
+    unchanged session must be persisted once, not on every tool completion,
+    while a real mutation must be persisted promptly.
+    """
+    import time
+
+    from api.streaming import (
+        _streaming_checkpoint_fingerprint,
+        _CHECKPOINT_IDLE_REFRESH_SECONDS,
+    )
+
+    s = _session(tmp_path)
+    writes = []
+
+    # Mirror of the loop body in _periodic_checkpoint(), kept in sync with it.
+    last_fingerprint = None
+    last_write_at = 0.0
+    now = time.time()
+
+    def tick(session, at):
+        nonlocal last_fingerprint, last_write_at
+        fingerprint = _streaming_checkpoint_fingerprint(session)
+        stale = (at - last_write_at) >= _CHECKPOINT_IDLE_REFRESH_SECONDS
+        if fingerprint is None or fingerprint != last_fingerprint or stale:
+            writes.append(at)
+            last_fingerprint = fingerprint
+            last_write_at = at
+
+    # 10 tool completions, nothing else changed -> exactly one write.
+    for i in range(10):
+        tick(s, now + i)
+    assert len(writes) == 1, f"redundant rewrites: {len(writes)} writes for an unchanged session"
+
+    # A real transcript change must be persisted on the next tick.
+    s.messages = list(s.messages) + [{"role": "assistant", "content": "new answer"}]
+    tick(s, now + 10)
+    assert len(writes) == 2, "a real mutation must force a checkpoint write"
+
+    # updated_at must not go stale forever on a very long quiet turn.
+    tick(s, now + 10 + _CHECKPOINT_IDLE_REFRESH_SECONDS)
+    assert len(writes) == 3, "a periodic refresh must still happen on long turns"