Keep legacy history pagination addressable

Author: lihongguang-0014

src/opensquilla/application/session_history.py

@@ -16,9 +16,10 @@
 
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import Any, Protocol, cast
+from typing import Protocol
 
 from opensquilla.history_cursor import (
+    HISTORY_CURSOR_MAX_INTEGER,
     HistoryCursor,
     HistoryCursorInvalidatedError,
 )
@@ -146,13 +147,19 @@ def cursor_for_entry(entry: object) -> HistoryCursor | None:
     """Return the stable integer cursor used by the history Port."""
 
     created_at = getattr(entry, "created_at", None)
-    stable_id = getattr(entry, "id", None) or getattr(entry, "message_id", None)
-    if created_at in {None, ""} or stable_id in {None, ""}:
-        return None
-    try:
-        return int(cast(Any, created_at)), int(cast(Any, stable_id))
-    except (TypeError, ValueError):
+    stable_id = getattr(entry, "id", None)
+    if (
+        not isinstance(created_at, int)
+        or isinstance(created_at, bool)
+        or not isinstance(stable_id, int)
+        or isinstance(stable_id, bool)
+        or created_at < 0
+        or stable_id < 0
+        or created_at > HISTORY_CURSOR_MAX_INTEGER
+        or stable_id > HISTORY_CURSOR_MAX_INTEGER
+    ):
         return None
+    return created_at, stable_id
 
 
 def paginate_transcript(

src/opensquilla/session/storage.py

@@ -14128,7 +14128,9 @@ async def get_canonical_transcript_page(
         one SQLite read snapshot. ``before`` keeps its historical precedence
         over ``after`` when both cursors exist. A supplied cursor must identify
         an anchor in this session; missing, foreign, or deleted anchors fail
-        closed instead of being treated as an unpositioned latest read.
+        closed instead of being treated as an unpositioned latest read. Legacy
+        archived rows without an original integer id are not keyset-addressable
+        and are omitted while canonical coverage reports the archive incomplete.
         """
         page_size = max(1, int(limit))
         fetch_size = page_size + 1
@@ -14235,6 +14237,7 @@ async def get_canonical_transcript_page(
                     schema_version
                 FROM compacted_transcript_entries
                 WHERE session_id = ?
+                  AND original_entry_id IS NOT NULL
                   AND EXISTS (SELECT 1 FROM cursor_anchor)
                   {archived_cursor_clause}
                 ORDER BY
@@ -14288,13 +14291,6 @@ async def get_canonical_transcript_page(
         entries = entries[:page_size]
         if not ascending:
             entries.reverse()
-        if has_more and entries:
-            continuation_entry = entries[-1] if ascending else entries[0]
-            # Legacy compacted rows can lack their original transcript id. The
-            # archive is already reported incomplete; do not advertise another
-            # keyset page when its boundary cannot form the numeric cursor.
-            if continuation_entry.id is None:
-                has_more = False
         return entries, has_more
 
     @_serialized_read

tests/test_application/test_session_history.py

@@ -12,6 +12,7 @@
     HistoryPage,
     SessionHistoryApplication,
     SessionHistoryQuery,
+    cursor_for_entry,
     paginate_transcript,
 )
 from opensquilla.history_cursor import HistoryCursorInvalidatedError
@@ -209,6 +210,15 @@ def test_paginate_transcript_preserves_latest_window_and_valid_cursors() -> None
     assert backward_more is True
 
 
+def test_cursor_for_entry_uses_only_bounded_integer_transcript_ids() -> None:
+    assert cursor_for_entry(SimpleNamespace(id=0, message_id="999", created_at=42)) == (
+        42,
+        0,
+    )
+    assert cursor_for_entry(SimpleNamespace(id=None, message_id="7", created_at=42)) is None
+    assert cursor_for_entry(SimpleNamespace(id=1 << 63, message_id="1", created_at=42)) is None
+
+
 @pytest.mark.parametrize("direction", ["before", "after"])
 def test_paginate_transcript_rejects_missing_cursor(direction: str) -> None:
     kwargs = {direction: (99, 99)}

tests/test_gateway/test_rpc_chat_history.py

@@ -160,48 +160,34 @@ async def test_chat_history_returns_pagination_metadata_with_legacy_messages() -
 
 
 @pytest.mark.asyncio
-async def test_chat_history_does_not_publish_unusable_legacy_null_id_cursor(
-    tmp_path,
-) -> None:
-    storage = SessionStorage(str(tmp_path / "history-legacy-null-id.db"))
-    await storage.connect()
-    manager = SessionManager(storage, inject_time_prefix=False)
+async def test_chat_history_does_not_publish_unusable_legacy_null_id_cursor() -> None:
     session_key = "agent:main:webchat:legacy-null-id"
-    node = await manager.create(session_key)
-    try:
-        await storage.conn.executemany(
-            """
-            INSERT INTO compacted_transcript_entries (
-                session_id, session_key, compaction_id, compaction_index,
-                original_entry_id, message_id, role, content, created_at, archived_at
-            ) VALUES (?, ?, ?, ?, NULL, ?, 'user', ?, ?, ?)
-            """,
-            [
-                (
-                    node.session_id,
-                    session_key,
-                    "legacy-compaction",
-                    0,
-                    f"legacy-{index}",
-                    f"legacy message {index}",
-                    index,
-                    index,
-                )
-                for index in (1, 2)
-            ],
-        )
-        await storage.conn.commit()
+    legacy_entry = TranscriptEntry(
+        id=None,
+        session_id="legacy-session",
+        session_key=session_key,
+        message_id="legacy-2",
+        role="user",
+        content="legacy message 2",
+        created_at=2,
+    )
+    manager = _FakePagedSessionManager(
+        [legacy_entry],
+        page={
+            "entries": [legacy_entry],
+            "has_more": True,
+            "canonical_complete": False,
+        },
+    )
 
-        result = await _handle_chat_history(
-            {"sessionKey": session_key, "limit": 1},
-            RpcContext(
-                conn_id="test",
-                principal=SimpleNamespace(role="operator"),
-                session_manager=manager,
-            ),
-        )
-    finally:
-        await storage.close()
+    result = await _handle_chat_history(
+        {"sessionKey": session_key, "limit": 1},
+        RpcContext(
+            conn_id="test",
+            principal=SimpleNamespace(role="operator"),
+            session_manager=manager,
+        ),
+    )
 
     assert [message["message_id"] for message in result["messages"]] == ["legacy-2"]
     assert result["canonical_complete"] is False

tests/test_session/test_manager.py

@@ -4070,6 +4070,58 @@ async def test_canonical_transcript_page_crosses_multiple_compaction_boundaries(
     assert [entry.message_id for entry in forward] == [entry.message_id for entry in loaded]
 
 
+@pytest.mark.asyncio
+async def test_canonical_page_skips_null_legacy_ids_without_hiding_valid_rows(manager):
+    node = await manager.create("agent:main:webchat:legacy-null-page")
+    await manager._storage.conn.executemany(
+        """
+        INSERT INTO compacted_transcript_entries (
+            session_id, session_key, compaction_id, compaction_index,
+            original_entry_id, message_id, role, content, created_at, archived_at
+        ) VALUES (?, ?, ?, ?, ?, ?, 'user', ?, ?, ?)
+        """,
+        [
+            (
+                node.session_id,
+                node.session_key,
+                "legacy-compaction",
+                0,
+                original_id,
+                f"legacy-{created_at}",
+                f"legacy message {created_at}",
+                created_at,
+                created_at,
+            )
+            for created_at, original_id in ((1, 1), (2, None), (3, 3))
+        ],
+    )
+    await manager._storage.conn.commit()
+
+    latest = await manager.get_canonical_transcript_page(node.session_key, limit=1)
+    before = await manager.get_canonical_transcript_page(
+        node.session_key,
+        limit=1,
+        before=(3, 3),
+    )
+    after = await manager.get_canonical_transcript_page(
+        node.session_key,
+        limit=1,
+        after=(1, 1),
+    )
+    full = await manager.get_canonical_transcript_page(node.session_key, limit=2)
+
+    assert [(entry.created_at, entry.id) for entry in latest.entries] == [(3, 3)]
+    assert latest.has_more is True
+    assert latest.canonical_complete is False
+    assert [(entry.created_at, entry.id) for entry in before.entries] == [(1, 1)]
+    assert before.has_more is False
+    assert [(entry.created_at, entry.id) for entry in after.entries] == [(3, 3)]
+    assert after.has_more is False
+    assert [(entry.created_at, entry.id) for entry in full.entries] == [(1, 1), (3, 3)]
+    assert full.has_more is False
+    assert full.canonical_complete is False
+
+
 @pytest.mark.asyncio
 async def test_canonical_transcript_page_preserves_turn_context(manager):
     """Paged canonical reads must keep turn_context on active and archived rows."""