diff --git a/agent/prompt_cache_scope.py b/agent/prompt_cache_scope.py index 1ffca5dd63..53ec690819 100644 --- a/agent/prompt_cache_scope.py +++ b/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]: diff --git a/hermes_state.py b/hermes_state.py index 0909289e45..b10270b94c 100644 --- a/hermes_state.py +++ b/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") diff --git a/tests/agent/test_declared_conversation_scope.py b/tests/agent/test_declared_conversation_scope.py index 21b2bc2b28..66ed0fcae3 100644 --- a/tests/agent/test_declared_conversation_scope.py +++ b/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