logger.py: batch-checkpoint refactor (see app.py commit for details)

core/logger.py

@@ -228,7 +228,9 @@ class SessionLogger:
 
         logger._last_confirmed_state = logger._state_hash()
         logger._ensure_worker()
-        logger._enqueue_upload()
+        # NOTE: caller (on_recover_session in app.py) must call
+        # checkpoint() after this returns, to persist the
+        # 'session_resumed' snapshot appended above.
         return logger
 
     def _filename(self):
@@ -267,7 +269,6 @@ class SessionLogger:
             "submitted_at": _now_iso(),
             "char_count": len(text),
         }
-        self._enqueue_upload()
 
     def set_discussion_plan(self, text):
         if not self.session_id:
@@ -276,7 +277,6 @@ class SessionLogger:
             "text": text,
             "extracted_at": _now_iso(),
         }
-        self._enqueue_upload()
 
     def set_perfect_answer(self, text):
         if not self.session_id:
@@ -286,7 +286,6 @@ class SessionLogger:
             "generated_at": _now_iso(),
             "char_count": len(text),
         }
-        self._enqueue_upload()
 
     def set_dominant_framework(self, framework):
         """Record which of the 3 frameworks (consequence / duty_rights /
@@ -295,7 +294,6 @@ class SessionLogger:
         if not self.session_id:
             return
         self.dominant_framework = framework
-        self._enqueue_upload()
 
     def add_feedback(self, message_index, kind, message_excerpt=""):
         """Record a participant aha! / ugh. reaction on a bot message.
@@ -304,6 +302,10 @@ class SessionLogger:
               "stress" (made participant uncomfortable),
               "none" (cancel, second click on the already-active emoji).
         Every click (including cancels) is appended with its own timestamp.
+
+        NOTE: in-memory only. The next checkpoint() (triggered by next chat
+        turn / end_session) will persist these events. Direct upload was
+        removed in the batch-push refactor to reduce HF commit pressure.
         """
         if not self.session_id:
             return
@@ -314,7 +316,6 @@ class SessionLogger:
             "timestamp": _now_iso(),
             "message_excerpt": (message_excerpt or "")[:200],
         })
-        self._enqueue_upload()
 
     def get_current_kind(self, message_index):
         """Return the current ('aha' / 'stress' / None) reaction for a given
@@ -327,7 +328,11 @@ class SessionLogger:
 
     def add_argument_snapshot(self, trigger, text, context=None):
         """Snapshot the participant's notepad textbox at an interaction event.
-        The bot NEVER sees this; logger-only side-channel."""
+        The bot NEVER sees this; logger-only side-channel.
+
+        NOTE: in-memory only. The next checkpoint() (triggered by next chat
+        turn / submit_initial / end_session) will persist these snapshots.
+        """
         if not self.session_id:
             return
         text = text or ""
@@ -343,18 +348,19 @@ class SessionLogger:
         if context:
             snap["context"] = context
         self.argument_snapshots.append(snap)
-        self._enqueue_upload()
 
     def add_manager_verdict(self, verdict_dict):
         """Append a manager verdict to the session-level history list.
         Called once per chat turn, right after the manager LLM returns.
 
         `verdict_dict` should be the dict form of `ManagerVerdict.to_dict()`.
+
+        NOTE: in-memory only. The next checkpoint() (called right after
+        the parent chat turn completes) will persist this verdict.
         """
         if not self.session_id:
             return
         self.manager_history.append(dict(verdict_dict))
-        self._enqueue_upload()
 
     def add_turn(self, user_message, manager_verdict, plan_control,
                  assistant_response, scratch=None, latency_s=None):
@@ -376,7 +382,8 @@ class SessionLogger:
         if latency_s:
             turn["latency_s"] = latency_s
         self.turns.append(turn)
-        self._enqueue_upload()
+        # NOTE: in-memory only. app.py's per-turn handler must call
+        # checkpoint() right after _log_turn() to persist this turn.
 
     def _derive_feedback_current(self):
         """Collapse feedback events to {message_index: latest_kind},
@@ -468,6 +475,26 @@ class SessionLogger:
             "session_summary": self._derive_session_summary(),
         }
 
+    def checkpoint(self):
+        """Public entry-point: explicitly push the current session state
+        to HF. Called from app.py at well-defined trigger points:
+
+          - per chat turn (after _log_turn in on_send)
+          - submit_initial (after add_argument_snapshot in on_submit_initial_confirm)
+          - session recovery (after from_dict in on_recover_session)
+
+        end_session_confirm uses flush() instead (synchronous, retry-heavy).
+        end_session_arm is NOT a checkpoint trigger — the arm event is
+        captured in-memory and ride along the next push.
+
+        Replaces the older behaviour where every add_xxx / set_xxx method
+        triggered its own upload. That created O(N) HF commits per session
+        and caused 256-commits/hour rate-limit pressure during concurrent
+        prod runs. New behaviour: O(turns) commits per session, with the
+        per-turn push carrying all interim feedback/snapshot/manager events
+        in a single cumulative payload."""
+        self._enqueue_upload()
+
     def _enqueue_upload(self):
         """Push the latest payload onto the upload queue. If a payload is
         already queued, it gets superseded (only the newest survives).