Python: fix(devui): page list_items past checkpoint items

python/packages/devui/agent_framework_devui/_conversations.py

@@ -516,7 +516,9 @@ async def list_items(
         if after:
             # Find the index after the cursor
             for i, item in enumerate(items):
-                if item.id == after:
+                # Checkpoint items are plain dicts, not models.
+                item_id = cast(dict[str, Any], item).get("id") if isinstance(item, dict) else item.id
+                if item_id == after:
                     start_idx = i + 1
                     break
 

python/packages/devui/tests/devui/test_conversations.py

@@ -5,6 +5,7 @@
 from typing import cast
 
 import pytest
+from agent_framework import WorkflowCheckpoint
 from openai.types.conversations import InputTextContent
 from openai.types.conversations.message import Message as OpenAIMessage
 from openai.types.responses import ResponseInputFile, ResponseInputImage, ResponseOutputRefusal
@@ -369,6 +370,38 @@ async def test_list_items_pagination():
     assert has_more is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("order", ["asc", "desc"])
+async def test_list_items_pagination_with_checkpoint_items(order: str):
+    """Paging with `after` must not fail when checkpoint items (plain dicts) are listed."""
+    store = InMemoryConversationStore()
+    conversation = store.create_conversation()
+    await store.add_items(
+        conversation.id,
+        items=[{"role": "user", "content": "first"}, {"role": "assistant", "content": "second"}],
+    )
+    storage = store._conversations[conversation.id]["checkpoint_storage"]
+    for _ in range(2):
+        await storage.save(WorkflowCheckpoint(workflow_name="workflow", graph_signature_hash="hash"))
+
+    def item_id(item: object) -> str:
+        # Checkpoint items are plain dicts at runtime; the rest are models.
+        return cast(str, item["id"] if isinstance(item, dict) else getattr(item, "id"))
+
+    all_items, _ = await store.list_items(conversation.id, order=order)
+    all_ids = [item_id(item) for item in all_items]
+
+    paged_ids: list[str] = []
+    after: str | None = None
+    has_more = True
+    while has_more:
+        page, has_more = await store.list_items(conversation.id, limit=1, after=after, order=order)
+        after = item_id(page[-1])
+        paged_ids.append(after)
+
+    assert paged_ids == all_ids
+
+
 @pytest.mark.asyncio
 async def test_list_items_converts_function_calls():
     """Test that list_items properly converts function calls to ResponseFunctionToolCallItem."""