fix(cache): make the conversation generation survive a backwards clock

Self-review of the generation marker. MAX(ended_at) alone is wall-clock: an
NTP correction between two resets writes a SMALLER boundary, MAX keeps
returning the older one, and the next conversation silently reuses the
previous generation -- two conversations on one routing key, which is the
defect this PR exists to remove.

latest_conversation_boundary now returns (count, latest_ended_at) and the
marker is 'count:ended_at'. The two halves fail under different conditions --
a backwards clock defeats the timestamp, retention pruning of an old ended row
decrements the count -- so a generation repeats only if both happen at once.
The pair is deliberately biased toward changing: a spurious change costs one
cold prompt-cache bucket, a repeat would merge two conversations.

Pinned by test_a_backwards_clock_does_not_reuse_a_generation, which rewrites
the second boundary to land before the first and asserts three conversations
still resolve to three distinct scopes.

Refs #96811

agent/prompt_cache_scope.py

@@ -116,8 +116,13 @@ def _conversation_generation(session_key: str, session_db: Any) -> str:
     - stable across a host's per-response physical ids (no boundary is written
       when nothing was reset, so every reply hashes the same value), and
     - rotating on every conversation replacement, ``/new`` and the policy
-      auto-resets alike, monotonically — ``ended_at`` only moves forward, so a
-      retired generation can never be reused.
+      auto-resets alike.
+
+    The marker pairs the boundary COUNT with the latest ``ended_at`` because
+    each alone can repeat a previous generation under a different rare
+    condition — a backwards clock correction defeats the timestamp, retention
+    pruning defeats the count — and the two do not fail together (see
+    ``SessionDB.latest_conversation_boundary``).
 
     No counter is introduced anywhere: the marker is read from state the
     reset paths already write, and it is read on the memoized resolution path,
@@ -132,8 +137,10 @@ def _conversation_generation(session_key: str, session_db: Any) -> str:
     boundary = reader(session_key)
     if boundary is None:
         return ""
-    # Fixed-point so the carrier is byte-identical across repr differences.
-    return f"{float(boundary):.6f}"
+    # (crossings, ended_at). Fixed-point on the timestamp so the carrier is
+    # byte-identical across repr differences between platforms.
+    crossings, ended_at = boundary
+    return f"{int(crossings)}:{float(ended_at):.6f}"
 
 
 def declared_conversation_scope(agent: Any) -> Optional[str]:

hermes_state.py

@@ -14134,8 +14134,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         session = self.get_session(session_id)
         return bool(session and self._is_explicit_fork_child_row(session))
 
-    def latest_conversation_boundary(self, session_key: str) -> Optional[float]:
-        """When the most recent conversation boundary closed for *session_key*.
+    def latest_conversation_boundary(
+        self, session_key: str
+    ) -> Optional[Tuple[int, float]]:
+        """How many conversation boundaries this key has crossed, and when.
 
         A boundary is a row this key ended at an intentional conversation
         break — the ``_RESET_END_REASONS`` set (``/new``, ``/switch``, idle,
@@ -14144,19 +14146,29 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         so the two agree on where one conversation stops and the next begins
         and cannot drift.
 
-        Returns ``None`` when the key has never been reset.  The value is a
-        monotonically advancing ``ended_at``: each reset appends a newer
-        boundary, so a generation derived from it can never roll back to a
-        previous one.  Read-only; ``agent/prompt_cache_scope.py`` uses it to
-        keep a host-declared conversation key from outliving the conversation
-        it names.
+        Returns ``(count, latest_ended_at)``, or ``None`` when the key has
+        never been reset.  BOTH halves are reported because each one alone has
+        a narrow way to repeat a previous generation:
+
+        - ``ended_at`` is wall-clock, so a backwards NTP correction between two
+          resets writes a SMALLER boundary and ``MAX`` keeps returning the older
+          one — the next conversation would reuse the previous generation;
+        - ``count`` survives that, but retention pruning of an old ended row
+          decrements it.
+
+        The two fail under different conditions, so the pair only repeats a
+        generation if both happen at once.  A spurious change is merely one cold
+        prompt-cache bucket; a repeated one would put two conversations on the
+        same routing key, so the pair is biased toward changing.  Read-only;
+        ``agent/prompt_cache_scope.py`` uses it to keep a host-declared
+        conversation key from outliving the conversation it names.
         """
         if not session_key:
             return None
         with self._read_ctx() as conn:
             row = conn.execute(
                 f"""
-                SELECT MAX(ended_at) AS boundary
+                SELECT COUNT(*) AS crossings, MAX(ended_at) AS boundary
                 FROM sessions
                 WHERE session_key = ?
                   AND ended_at IS NOT NULL
@@ -14165,7 +14177,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 (session_key,),
             ).fetchone()
         boundary = row["boundary"] if row is not None else None
-        return float(boundary) if boundary is not None else None
+        if boundary is None:
+            return None
+        return int(row["crossings"] or 0), float(boundary)
 
     def _is_compression_child_row(self, child: Dict[str, Any]) -> bool:
         parent_id = child.get("parent_session_id")

tests/agent/test_declared_conversation_scope.py

@@ -507,3 +507,30 @@ class TestConversationGenerationRotates:
         agent = _agent("sess-A", LegacyDB(), self.KEY)
         scope = declared_conversation_scope(agent)
         assert scope is not None and scope.startswith("gwk_")
+
+    def test_a_backwards_clock_does_not_reuse_a_generation(self, db):
+        """An NTP correction between two resets must not merge them.
+
+        ``MAX(ended_at)`` alone would keep returning the earlier, larger
+        timestamp; the boundary COUNT is what separates them.
+        """
+        first = self._keyed(db, "sess-clock-1")
+        scope_first = resolve_prompt_cache_scope(first)
+        db.end_session("sess-clock-1", "session_reset")
+
+        second = self._keyed(db, "sess-clock-2")
+        scope_second = resolve_prompt_cache_scope(second)
+        db.end_session("sess-clock-2", "session_reset")
+        # The clock went backwards: this boundary lands BEFORE the first one.
+        with db._lock:
+            db._conn.execute(
+                "UPDATE sessions SET ended_at = ("
+                "  SELECT MIN(ended_at) FROM sessions WHERE ended_at IS NOT NULL"
+                ") - 60 WHERE id = ?",
+                ("sess-clock-2",),
+            )
+            db._conn.commit()
+
+        third = self._keyed(db, "sess-clock-3")
+        scope_third = resolve_prompt_cache_scope(third)
+        assert len({scope_first, scope_second, scope_third}) == 3