Merge branch 'main' into content-extension

tests/common/mcp.py

@@ -16,13 +16,10 @@ MCP_TOOLGROUP_ID = "mcp::localmcp"
 
 def default_tools():
     """Default tools for backward compatibility."""
-    from mcp import types
     from mcp.server.fastmcp import Context
 
-    async def greet_everyone(
-        url: str, ctx: Context
-    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
-        return [types.TextContent(type="text", text="Hello, world!")]
+    async def greet_everyone(url: str, ctx: Context) -> str:
+        return "Hello, world!"
 
     async def get_boiling_point(liquid_name: str, celsius: bool = True) -> int:
         """
@@ -45,7 +42,6 @@ def default_tools():
 
 def dependency_tools():
     """Tools with natural dependencies for multi-turn testing."""
-    from mcp import types
     from mcp.server.fastmcp import Context
 
     async def get_user_id(username: str, ctx: Context) -> str:
@@ -106,7 +102,7 @@ def dependency_tools():
         else:
             access = "no"
 
-        return [types.TextContent(type="text", text=access)]
+        return access
 
     async def get_experiment_id(experiment_name: str, ctx: Context) -> str:
         """
@@ -245,7 +241,6 @@ def make_mcp_server(required_auth_token: str | None = None, tools: dict[str, Cal
     try:
         yield {"server_url": server_url}
     finally:
-        print("Telling SSE server to exit")
         server_instance.should_exit = True
         time.sleep(0.5)
 
@@ -269,4 +264,3 @@ def make_mcp_server(required_auth_token: str | None = None, tools: dict[str, Cal
 
         AppStatus.should_exit = False
         AppStatus.should_exit_event = None
-        print("SSE server exited")

tests/external/llama-stack-api-weather/pyproject.toml

@@ -3,7 +3,7 @@ name = "llama-stack-api-weather"
 version = "0.1.0"
 description = "Weather API for Llama Stack"
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.12"
 dependencies = ["llama-stack", "pydantic"]
 
 [build-system]

tests/external/llama-stack-provider-kaze/pyproject.toml

@@ -3,7 +3,7 @@ name = "llama-stack-provider-kaze"
 version = "0.1.0"
 description = "Kaze weather provider for Llama Stack"
 readme = "README.md"
-requires-python = ">=3.10"
+requires-python = ">=3.12"
 dependencies = ["llama-stack", "pydantic", "aiohttp"]
 
 [build-system]

tests/integration/fixtures/common.py

@@ -270,7 +270,7 @@ def openai_client(client_with_models):
 
 @pytest.fixture(params=["openai_client", "client_with_models"])
 def compat_client(request, client_with_models):
-    if isinstance(client_with_models, LlamaStackAsLibraryClient):
+    if request.param == "openai_client" and isinstance(client_with_models, LlamaStackAsLibraryClient):
         # OpenAI client expects a server, so unless we also rewrite OpenAI client's requests
         # to go via the Stack library client (which itself rewrites requests to be served inline),
         # we cannot do this.

tests/integration/non_ci/responses/fixtures/test_cases/responses.yaml

@@ -137,7 +137,7 @@ test_response_multi_turn_tool_execution:
         server_url: "<FILLED_BY_TEST_RUNNER>"
       output: "yes"
     - case_id: "experiment_results_lookup"
-      input: "I need to get the results for the 'boiling_point' experiment. First, get the experiment ID for 'boiling_point', then use that ID to get the experiment results. Tell me what you found."
+      input: "I need to get the results for the 'boiling_point' experiment. First, get the experiment ID for 'boiling_point', then use that ID to get the experiment results. Tell me the boiling point in Celsius."
       tools:
       - type: mcp
         server_label: "localmcp"
@@ -149,7 +149,7 @@ test_response_multi_turn_tool_execution_streaming:
   test_params:
     case:
     - case_id: "user_permissions_workflow"
-      input: "Help me with this security check: First, get the user ID for 'charlie', then get the permissions for that user ID, and finally check if that user can access 'secret_file.txt'. Stream your progress as you work through each step."
+      input: "Help me with this security check: First, get the user ID for 'charlie', then get the permissions for that user ID, and finally check if that user can access 'secret_file.txt'. Stream your progress as you work through each step. Return only one tool call per step. Summarize the final result with a single 'yes' or 'no' response."
       tools:
       - type: mcp
         server_label: "localmcp"
@@ -157,7 +157,7 @@ test_response_multi_turn_tool_execution_streaming:
       stream: true
       output: "no"
     - case_id: "experiment_analysis_streaming"
-      input: "I need a complete analysis: First, get the experiment ID for 'chemical_reaction', then get the results for that experiment, and tell me if the yield was above 80%. Please stream your analysis process."
+      input: "I need a complete analysis: First, get the experiment ID for 'chemical_reaction', then get the results for that experiment, and tell me if the yield was above 80%. Return only one tool call per step.  Please stream your analysis process."
       tools:
       - type: mcp
         server_label: "localmcp"

tests/integration/non_ci/responses/test_responses.py

@@ -363,6 +363,9 @@ def test_response_non_streaming_file_search_empty_vector_store(request, compat_c
     ids=case_id_generator,
 )
 def test_response_non_streaming_mcp_tool(request, compat_client, text_model_id, case):
+    if not isinstance(compat_client, LlamaStackAsLibraryClient):
+        pytest.skip("in-process MCP server is only supported in library client")
+
     with make_mcp_server() as mcp_server_info:
         tools = case["tools"]
         for tool in tools:
@@ -381,12 +384,18 @@ def test_response_non_streaming_mcp_tool(request, compat_client, text_model_id,
         assert list_tools.type == "mcp_list_tools"
         assert list_tools.server_label == "localmcp"
         assert len(list_tools.tools) == 2
-        assert {t.name for t in list_tools.tools} == {"get_boiling_point", "greet_everyone"}
+        assert {t.name for t in list_tools.tools} == {
+            "get_boiling_point",
+            "greet_everyone",
+        }
 
         call = response.output[1]
         assert call.type == "mcp_call"
         assert call.name == "get_boiling_point"
-        assert json.loads(call.arguments) == {"liquid_name": "myawesomeliquid", "celsius": True}
+        assert json.loads(call.arguments) == {
+            "liquid_name": "myawesomeliquid",
+            "celsius": True,
+        }
         assert call.error is None
         assert "-100" in call.output
 
@@ -485,8 +494,11 @@ def test_response_non_streaming_multi_turn_image(request, compat_client, text_mo
     responses_test_cases["test_response_multi_turn_tool_execution"]["test_params"]["case"],
     ids=case_id_generator,
 )
-def test_response_non_streaming_multi_turn_tool_execution(request, compat_client, text_model_id, case):
+def test_response_non_streaming_multi_turn_tool_execution(compat_client, text_model_id, case):
     """Test multi-turn tool execution where multiple MCP tool calls are performed in sequence."""
+    if not isinstance(compat_client, LlamaStackAsLibraryClient):
+        pytest.skip("in-process MCP server is only supported in library client")
+
     with make_mcp_server(tools=dependency_tools()) as mcp_server_info:
         tools = case["tools"]
         # Replace the placeholder URL with the actual server URL
@@ -541,8 +553,11 @@ def test_response_non_streaming_multi_turn_tool_execution(request, compat_client
     responses_test_cases["test_response_multi_turn_tool_execution_streaming"]["test_params"]["case"],
     ids=case_id_generator,
 )
-async def test_response_streaming_multi_turn_tool_execution(request, compat_client, text_model_id, case):
+def test_response_streaming_multi_turn_tool_execution(compat_client, text_model_id, case):
     """Test streaming multi-turn tool execution where multiple MCP tool calls are performed in sequence."""
+    if not isinstance(compat_client, LlamaStackAsLibraryClient):
+        pytest.skip("in-process MCP server is only supported in library client")
+
     with make_mcp_server(tools=dependency_tools()) as mcp_server_info:
         tools = case["tools"]
         # Replace the placeholder URL with the actual server URL
@@ -572,6 +587,105 @@ async def test_response_streaming_multi_turn_tool_execution(request, compat_clie
             f"Last chunk should be response.completed, got {chunks[-1].type}"
         )
 
+        # Verify tool call streaming events are present
+        chunk_types = [chunk.type for chunk in chunks]
+
+        # Should have function call arguments delta events for tool calls
+        delta_events = [chunk for chunk in chunks if chunk.type == "response.function_call_arguments.delta"]
+        done_events = [chunk for chunk in chunks if chunk.type == "response.function_call_arguments.done"]
+
+        # Should have output item events for tool calls
+        item_added_events = [chunk for chunk in chunks if chunk.type == "response.output_item.added"]
+        item_done_events = [chunk for chunk in chunks if chunk.type == "response.output_item.done"]
+
+        # Verify we have substantial streaming activity (not just batch events)
+        assert len(chunks) > 10, f"Expected rich streaming with many events, got only {len(chunks)} chunks"
+
+        # Since this test involves MCP tool calls, we should see streaming events
+        assert len(delta_events) > 0, f"Expected function_call_arguments.delta events, got chunk types: {chunk_types}"
+        assert len(done_events) > 0, f"Expected function_call_arguments.done events, got chunk types: {chunk_types}"
+
+        # Should have output item events for function calls
+        assert len(item_added_events) > 0, f"Expected response.output_item.added events, got chunk types: {chunk_types}"
+        assert len(item_done_events) > 0, f"Expected response.output_item.done events, got chunk types: {chunk_types}"
+
+        # Verify delta events have proper structure
+        for delta_event in delta_events:
+            assert hasattr(delta_event, "delta"), "Delta event should have 'delta' field"
+            assert hasattr(delta_event, "item_id"), "Delta event should have 'item_id' field"
+            assert hasattr(delta_event, "sequence_number"), "Delta event should have 'sequence_number' field"
+            assert delta_event.delta, "Delta should not be empty"
+
+        # Verify done events have proper structure
+        for done_event in done_events:
+            assert hasattr(done_event, "arguments"), "Done event should have 'arguments' field"
+            assert hasattr(done_event, "item_id"), "Done event should have 'item_id' field"
+            assert done_event.arguments, "Final arguments should not be empty"
+
+        # Verify output item added events have proper structure
+        for added_event in item_added_events:
+            assert hasattr(added_event, "item"), "Added event should have 'item' field"
+            assert hasattr(added_event, "output_index"), "Added event should have 'output_index' field"
+            assert hasattr(added_event, "sequence_number"), "Added event should have 'sequence_number' field"
+            assert hasattr(added_event, "response_id"), "Added event should have 'response_id' field"
+            assert added_event.item.type in ["function_call", "mcp_call"], "Added item should be a tool call"
+            assert added_event.item.status == "in_progress", "Added item should be in progress"
+            assert added_event.response_id, "Response ID should not be empty"
+            assert isinstance(added_event.output_index, int), "Output index should be integer"
+            assert added_event.output_index >= 0, "Output index should be non-negative"
+
+        # Verify output item done events have proper structure
+        for done_event in item_done_events:
+            assert hasattr(done_event, "item"), "Done event should have 'item' field"
+            assert hasattr(done_event, "output_index"), "Done event should have 'output_index' field"
+            assert hasattr(done_event, "sequence_number"), "Done event should have 'sequence_number' field"
+            assert hasattr(done_event, "response_id"), "Done event should have 'response_id' field"
+            assert done_event.item.type in ["function_call", "mcp_call"], "Done item should be a tool call"
+            # Note: MCP calls don't have a status field, only function calls do
+            if done_event.item.type == "function_call":
+                assert done_event.item.status == "completed", "Function call should be completed"
+            assert done_event.response_id, "Response ID should not be empty"
+            assert isinstance(done_event.output_index, int), "Output index should be integer"
+            assert done_event.output_index >= 0, "Output index should be non-negative"
+
+        # Group function call argument events by item_id (these should have proper tracking)
+        function_call_events_by_item_id = {}
+        for chunk in chunks:
+            if hasattr(chunk, "item_id") and chunk.type in [
+                "response.function_call_arguments.delta",
+                "response.function_call_arguments.done",
+            ]:
+                item_id = chunk.item_id
+                if item_id not in function_call_events_by_item_id:
+                    function_call_events_by_item_id[item_id] = []
+                function_call_events_by_item_id[item_id].append(chunk)
+
+        for item_id, related_events in function_call_events_by_item_id.items():
+            # Should have at least one delta and one done event for a complete function call
+            delta_events = [e for e in related_events if e.type == "response.function_call_arguments.delta"]
+            done_events = [e for e in related_events if e.type == "response.function_call_arguments.done"]
+
+            assert len(delta_events) > 0, f"Item {item_id} should have at least one delta event"
+            assert len(done_events) == 1, f"Item {item_id} should have exactly one done event"
+
+            # Verify all events have the same item_id
+            for event in related_events:
+                assert event.item_id == item_id, f"Event should have consistent item_id {item_id}, got {event.item_id}"
+
+        # Basic pairing check: each output_item.added should be followed by some activity
+        # (but we can't enforce strict 1:1 pairing due to the complexity of multi-turn scenarios)
+        assert len(item_added_events) > 0, "Should have at least one output_item.added event"
+
+        # Verify response_id consistency across all events
+        response_ids = set()
+        for chunk in chunks:
+            if hasattr(chunk, "response_id"):
+                response_ids.add(chunk.response_id)
+            elif hasattr(chunk, "response") and hasattr(chunk.response, "id"):
+                response_ids.add(chunk.response.id)
+
+        assert len(response_ids) == 1, f"All events should reference the same response_id, found: {response_ids}"
+
         # Get the final response from the last chunk
         final_chunk = chunks[-1]
         if hasattr(final_chunk, "response"):
@@ -634,7 +748,7 @@ async def test_response_streaming_multi_turn_tool_execution(request, compat_clie
         },
     ],
 )
-def test_response_text_format(request, compat_client, text_model_id, text_format):
+def test_response_text_format(compat_client, text_model_id, text_format):
     if isinstance(compat_client, LlamaStackAsLibraryClient):
         pytest.skip("Responses API text format is not yet supported in library client.")
 
@@ -653,7 +767,7 @@ def test_response_text_format(request, compat_client, text_model_id, text_format
 
 
 @pytest.fixture
-def vector_store_with_filtered_files(request, compat_client, text_model_id, tmp_path_factory):
+def vector_store_with_filtered_files(compat_client, text_model_id, tmp_path_factory):
     """Create a vector store with multiple files that have different attributes for filtering tests."""
     if isinstance(compat_client, LlamaStackAsLibraryClient):
         pytest.skip("Responses API file search is not yet supported in library client.")
@@ -713,7 +827,9 @@ def vector_store_with_filtered_files(request, compat_client, text_model_id, tmp_
 
         # Attach file to vector store with attributes
         file_attach_response = compat_client.vector_stores.files.create(
-            vector_store_id=vector_store.id, file_id=file_response.id, attributes=file_data["attributes"]
+            vector_store_id=vector_store.id,
+            file_id=file_response.id,
+            attributes=file_data["attributes"],
         )
 
         # Wait for attachment

tests/integration/vector_io/test_openai_vector_stores.py

@@ -10,10 +10,11 @@ import uuid
 from io import BytesIO
 
 import pytest
-from llama_stack_client import BadRequestError, LlamaStackClient
+from llama_stack_client import BadRequestError
 from openai import BadRequestError as OpenAIBadRequestError
 
 from llama_stack.apis.vector_io import Chunk
+from llama_stack.core.library_client import LlamaStackAsLibraryClient
 
 logger = logging.getLogger(__name__)
 
@@ -476,9 +477,6 @@ def test_openai_vector_store_attach_file(compat_client_with_empty_stores, client
     """Test OpenAI vector store attach file."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files attach is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -527,9 +525,6 @@ def test_openai_vector_store_attach_files_on_creation(compat_client_with_empty_s
     """Test OpenAI vector store attach files on creation."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files attach is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create some files and attach them to the vector store
@@ -583,9 +578,6 @@ def test_openai_vector_store_list_files(compat_client_with_empty_stores, client_
     """Test OpenAI vector store list files."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files list is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -598,16 +590,20 @@ def test_openai_vector_store_list_files(compat_client_with_empty_stores, client_
             file_buffer.name = f"openai_test_{i}.txt"
             file = compat_client.files.create(file=file_buffer, purpose="assistants")
 
-        compat_client.vector_stores.files.create(
+        response = compat_client.vector_stores.files.create(
             vector_store_id=vector_store.id,
             file_id=file.id,
         )
+        assert response is not None
+        assert response.status == "completed", (
+            f"Failed to attach file {file.id} to vector store {vector_store.id}: {response=}"
+        )
         file_ids.append(file.id)
 
     files_list = compat_client.vector_stores.files.list(vector_store_id=vector_store.id)
     assert files_list
     assert files_list.object == "list"
-    assert files_list.data
+    assert files_list.data is not None
     assert not files_list.has_more
     assert len(files_list.data) == 3
     assert set(file_ids) == {file.id for file in files_list.data}
@@ -643,12 +639,13 @@ def test_openai_vector_store_list_files_invalid_vector_store(compat_client_with_
     """Test OpenAI vector store list files with invalid vector store ID."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files list is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
+    if isinstance(compat_client, LlamaStackAsLibraryClient):
+        errors = ValueError
+    else:
+        errors = (BadRequestError, OpenAIBadRequestError)
 
-    with pytest.raises((BadRequestError, OpenAIBadRequestError)):
+    with pytest.raises(errors):
         compat_client.vector_stores.files.list(vector_store_id="abc123")
 
 
@@ -656,9 +653,6 @@ def test_openai_vector_store_retrieve_file_contents(compat_client_with_empty_sto
     """Test OpenAI vector store retrieve file contents."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files retrieve contents is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -686,9 +680,15 @@ def test_openai_vector_store_retrieve_file_contents(compat_client_with_empty_sto
         file_id=file.id,
     )
 
-    assert file_contents
-    assert file_contents.content[0]["type"] == "text"
-    assert file_contents.content[0]["text"] == test_content.decode("utf-8")
+    assert file_contents is not None
+    assert len(file_contents.content) == 1
+    content = file_contents.content[0]
+
+    # llama-stack-client returns a model, openai-python is a badboy and returns a dict
+    if not isinstance(content, dict):
+        content = content.model_dump()
+    assert content["type"] == "text"
+    assert content["text"] == test_content.decode("utf-8")
     assert file_contents.filename == file_name
     assert file_contents.attributes == attributes
 
@@ -697,9 +697,6 @@ def test_openai_vector_store_delete_file(compat_client_with_empty_stores, client
     """Test OpenAI vector store delete file."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files list is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -752,9 +749,6 @@ def test_openai_vector_store_delete_file_removes_from_vector_store(compat_client
     """Test OpenAI vector store delete file removes from vector store."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files attach is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -793,9 +787,6 @@ def test_openai_vector_store_update_file(compat_client_with_empty_stores, client
     """Test OpenAI vector store update file."""
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files update is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store
@@ -841,9 +832,6 @@ def test_create_vector_store_files_duplicate_vector_store_name(compat_client_wit
     """
     skip_if_provider_doesnt_support_openai_vector_stores(client_with_models)
 
-    if isinstance(compat_client_with_empty_stores, LlamaStackClient):
-        pytest.skip("Vector Store Files create is not yet supported with LlamaStackClient")
-
     compat_client = compat_client_with_empty_stores
 
     # Create a vector store with files

tests/unit/providers/agent/test_agent_meta_reference.py

@@ -0,0 +1,347 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+import json
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from llama_stack.apis.agents import Session
+from llama_stack.core.datatypes import User
+from llama_stack.providers.inline.agents.meta_reference.persistence import (
+    AgentPersistence,
+    AgentSessionInfo,
+)
+from llama_stack.providers.utils.kvstore import KVStore
+
+
+@pytest.fixture
+def mock_kvstore():
+    return AsyncMock(spec=KVStore)
+
+
+@pytest.fixture
+def mock_policy():
+    return []
+
+
+@pytest.fixture
+def agent_persistence(mock_kvstore, mock_policy):
+    return AgentPersistence(agent_id="test-agent-123", kvstore=mock_kvstore, policy=mock_policy)
+
+
+@pytest.fixture
+def sample_session():
+    return AgentSessionInfo(
+        session_id="session-123",
+        session_name="Test Session",
+        started_at=datetime.now(UTC),
+        owner=User(principal="user-123", attributes=None),
+        turns=[],
+        identifier="test-session",
+        type="session",
+    )
+
+
+@pytest.fixture
+def sample_session_json(sample_session):
+    return sample_session.model_dump_json()
+
+
+class TestAgentPersistenceListSessions:
+    def setup_mock_kvstore(self, mock_kvstore, session_keys=None, turn_keys=None, invalid_keys=None, custom_data=None):
+        """Helper to setup mock kvstore with sessions, turns, and custom/invalid data
+
+        Args:
+            mock_kvstore: The mock KVStore object
+            session_keys: List of session keys or dict mapping keys to custom session data
+            turn_keys: List of turn keys or dict mapping keys to custom turn data
+            invalid_keys: Dict mapping keys to invalid/corrupt data
+            custom_data: Additional custom data to add to the mock responses
+        """
+        all_keys = []
+        mock_data = {}
+
+        # session keys
+        if session_keys:
+            if isinstance(session_keys, dict):
+                all_keys.extend(session_keys.keys())
+                mock_data.update({k: json.dumps(v) if isinstance(v, dict) else v for k, v in session_keys.items()})
+            else:
+                all_keys.extend(session_keys)
+                for key in session_keys:
+                    session_id = key.split(":")[-1]
+                    mock_data[key] = json.dumps(
+                        {
+                            "session_id": session_id,
+                            "session_name": f"Session {session_id}",
+                            "started_at": datetime.now(UTC).isoformat(),
+                            "turns": [],
+                        }
+                    )
+
+        # turn keys
+        if turn_keys:
+            if isinstance(turn_keys, dict):
+                all_keys.extend(turn_keys.keys())
+                mock_data.update({k: json.dumps(v) if isinstance(v, dict) else v for k, v in turn_keys.items()})
+            else:
+                all_keys.extend(turn_keys)
+                for key in turn_keys:
+                    parts = key.split(":")
+                    session_id = parts[-2]
+                    turn_id = parts[-1]
+                    mock_data[key] = json.dumps(
+                        {
+                            "turn_id": turn_id,
+                            "session_id": session_id,
+                            "input_messages": [],
+                            "started_at": datetime.now(UTC).isoformat(),
+                        }
+                    )
+
+        if invalid_keys:
+            all_keys.extend(invalid_keys.keys())
+            mock_data.update(invalid_keys)
+
+        if custom_data:
+            mock_data.update(custom_data)
+
+        values_list = list(mock_data.values())
+        mock_kvstore.values_in_range.return_value = values_list
+
+        async def mock_get(key):
+            return mock_data.get(key)
+
+        mock_kvstore.get.side_effect = mock_get
+
+        return mock_data
+
+    @pytest.mark.parametrize(
+        "scenario",
+        [
+            {
+                # from this issue: https://github.com/meta-llama/llama-stack/issues/3048
+                "name": "reported_bug",
+                "session_keys": ["session:test-agent-123:1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d"],
+                "turn_keys": [
+                    "session:test-agent-123:1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d:eb7e818f-41fb-49a0-bdd6-464974a2d2ad"
+                ],
+                "expected_sessions": ["1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d"],
+            },
+            {
+                "name": "basic_filtering",
+                "session_keys": ["session:test-agent-123:session-1", "session:test-agent-123:session-2"],
+                "turn_keys": ["session:test-agent-123:session-1:turn-1", "session:test-agent-123:session-1:turn-2"],
+                "expected_sessions": ["session-1", "session-2"],
+            },
+            {
+                "name": "multiple_turns_per_session",
+                "session_keys": ["session:test-agent-123:session-456"],
+                "turn_keys": [
+                    "session:test-agent-123:session-456:turn-789",
+                    "session:test-agent-123:session-456:turn-790",
+                ],
+                "expected_sessions": ["session-456"],
+            },
+            {
+                "name": "multiple_sessions_with_turns",
+                "session_keys": ["session:test-agent-123:session-1", "session:test-agent-123:session-2"],
+                "turn_keys": [
+                    "session:test-agent-123:session-1:turn-1",
+                    "session:test-agent-123:session-1:turn-2",
+                    "session:test-agent-123:session-2:turn-3",
+                ],
+                "expected_sessions": ["session-1", "session-2"],
+            },
+        ],
+    )
+    async def test_list_sessions_key_filtering(self, agent_persistence, mock_kvstore, scenario):
+        self.setup_mock_kvstore(mock_kvstore, session_keys=scenario["session_keys"], turn_keys=scenario["turn_keys"])
+
+        with patch("llama_stack.providers.inline.agents.meta_reference.persistence.log") as mock_log:
+            result = await agent_persistence.list_sessions()
+
+        assert len(result) == len(scenario["expected_sessions"])
+        session_ids = {s.session_id for s in result}
+        for expected_id in scenario["expected_sessions"]:
+            assert expected_id in session_ids
+
+        # no errors should be logged
+        mock_log.error.assert_not_called()
+
+    @pytest.mark.parametrize(
+        "error_scenario",
+        [
+            {
+                "name": "invalid_json",
+                "valid_keys": ["session:test-agent-123:valid-session"],
+                "invalid_data": {"session:test-agent-123:invalid-json": "corrupted-json-data{"},
+                "expected_valid_sessions": ["valid-session"],
+                "expected_error_count": 1,
+            },
+            {
+                "name": "missing_fields",
+                "valid_keys": ["session:test-agent-123:valid-session"],
+                "invalid_data": {
+                    "session:test-agent-123:invalid-schema": json.dumps(
+                        {
+                            "session_id": "invalid-schema",
+                            "session_name": "Missing Fields",
+                            # missing `started_at` and `turns`
+                        }
+                    )
+                },
+                "expected_valid_sessions": ["valid-session"],
+                "expected_error_count": 1,
+            },
+            {
+                "name": "multiple_invalid",
+                "valid_keys": ["session:test-agent-123:valid-session-1", "session:test-agent-123:valid-session-2"],
+                "invalid_data": {
+                    "session:test-agent-123:corrupted-json": "not-valid-json{",
+                    "session:test-agent-123:incomplete-data": json.dumps({"incomplete": "data"}),
+                },
+                "expected_valid_sessions": ["valid-session-1", "valid-session-2"],
+                "expected_error_count": 2,
+            },
+        ],
+    )
+    async def test_list_sessions_error_handling(self, agent_persistence, mock_kvstore, error_scenario):
+        session_keys = {}
+        for key in error_scenario["valid_keys"]:
+            session_id = key.split(":")[-1]
+            session_keys[key] = {
+                "session_id": session_id,
+                "session_name": f"Valid {session_id}",
+                "started_at": datetime.now(UTC).isoformat(),
+                "turns": [],
+            }
+
+        self.setup_mock_kvstore(mock_kvstore, session_keys=session_keys, invalid_keys=error_scenario["invalid_data"])
+
+        with patch("llama_stack.providers.inline.agents.meta_reference.persistence.log") as mock_log:
+            result = await agent_persistence.list_sessions()
+
+        # only valid sessions should be returned
+        assert len(result) == len(error_scenario["expected_valid_sessions"])
+        session_ids = {s.session_id for s in result}
+        for expected_id in error_scenario["expected_valid_sessions"]:
+            assert expected_id in session_ids
+
+        # error should be logged
+        assert mock_log.error.call_count > 0
+        assert mock_log.error.call_count == error_scenario["expected_error_count"]
+
+    async def test_list_sessions_empty(self, agent_persistence, mock_kvstore):
+        mock_kvstore.values_in_range.return_value = []
+
+        result = await agent_persistence.list_sessions()
+
+        assert result == []
+        mock_kvstore.values_in_range.assert_called_once_with(
+            start_key="session:test-agent-123:", end_key="session:test-agent-123:\xff\xff\xff\xff"
+        )
+
+    async def test_list_sessions_properties(self, agent_persistence, mock_kvstore):
+        session_data = {
+            "session_id": "session-123",
+            "session_name": "Test Session",
+            "started_at": datetime.now(UTC).isoformat(),
+            "owner": {"principal": "user-123", "attributes": None},
+            "turns": [],
+        }
+
+        self.setup_mock_kvstore(mock_kvstore, session_keys={"session:test-agent-123:session-123": session_data})
+
+        result = await agent_persistence.list_sessions()
+
+        assert len(result) == 1
+        assert isinstance(result[0], Session)
+        assert result[0].session_id == "session-123"
+        assert result[0].session_name == "Test Session"
+        assert result[0].turns == []
+        assert hasattr(result[0], "started_at")
+
+    async def test_list_sessions_kvstore_exception(self, agent_persistence, mock_kvstore):
+        mock_kvstore.values_in_range.side_effect = Exception("KVStore error")
+
+        with pytest.raises(Exception, match="KVStore error"):
+            await agent_persistence.list_sessions()
+
+    async def test_bug_data_loss_with_real_data(self, agent_persistence, mock_kvstore):
+        # tests the handling of the issue reported in: https://github.com/meta-llama/llama-stack/issues/3048
+        session_data = {
+            "session_id": "1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d",
+            "session_name": "Test Session",
+            "started_at": datetime.now(UTC).isoformat(),
+            "turns": [],
+        }
+
+        turn_data = {
+            "turn_id": "eb7e818f-41fb-49a0-bdd6-464974a2d2ad",
+            "session_id": "1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d",
+            "input_messages": [
+                {"role": "user", "content": "if i had a cluster i would want to call it persistence01", "context": None}
+            ],
+            "steps": [
+                {
+                    "turn_id": "eb7e818f-41fb-49a0-bdd6-464974a2d2ad",
+                    "step_id": "c0f797dd-3d34-4bc5-a8f4-db6af9455132",
+                    "started_at": "2025-08-05T14:31:50.000484Z",
+                    "completed_at": "2025-08-05T14:31:51.303691Z",
+                    "step_type": "inference",
+                    "model_response": {
+                        "role": "assistant",
+                        "content": "OK, I can create a cluster named 'persistence01' for you.",
+                        "stop_reason": "end_of_turn",
+                        "tool_calls": [],
+                    },
+                }
+            ],
+            "output_message": {
+                "role": "assistant",
+                "content": "OK, I can create a cluster named 'persistence01' for you.",
+                "stop_reason": "end_of_turn",
+                "tool_calls": [],
+            },
+            "output_attachments": [],
+            "started_at": "2025-08-05T14:31:49.999950Z",
+            "completed_at": "2025-08-05T14:31:51.305384Z",
+        }
+
+        mock_data = {
+            "session:test-agent-123:1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d": json.dumps(session_data),
+            "session:test-agent-123:1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d:eb7e818f-41fb-49a0-bdd6-464974a2d2ad": json.dumps(
+                turn_data
+            ),
+        }
+
+        mock_kvstore.values_in_range.return_value = list(mock_data.values())
+
+        async def mock_get(key):
+            return mock_data.get(key)
+
+        mock_kvstore.get.side_effect = mock_get
+
+        with patch("llama_stack.providers.inline.agents.meta_reference.persistence.log") as mock_log:
+            result = await agent_persistence.list_sessions()
+
+        assert len(result) == 1
+        assert result[0].session_id == "1f08fd1c-5a9d-459d-a00b-36d4dfa49b7d"
+
+        # confirm no errors logged
+        mock_log.error.assert_not_called()
+
+    async def test_list_sessions_key_range_construction(self, agent_persistence, mock_kvstore):
+        mock_kvstore.values_in_range.return_value = []
+
+        await agent_persistence.list_sessions()
+
+        mock_kvstore.values_in_range.assert_called_once_with(
+            start_key="session:test-agent-123:", end_key="session:test-agent-123:\xff\xff\xff\xff"
+        )

tests/unit/providers/agents/meta_reference/test_openai_responses.py

@@ -272,7 +272,9 @@ async def test_create_openai_response_with_tool_call_type_none(openai_responses_
 
     # Check that we got the content from our mocked tool execution result
     chunks = [chunk async for chunk in result]
-    assert len(chunks) == 2  # Should have response.created and response.completed
+    # Should have: response.created, output_item.added, function_call_arguments.delta,
+    # function_call_arguments.done, output_item.done, response.completed
+    assert len(chunks) == 6
 
     # Verify inference API was called correctly (after iterating over result)
     first_call = mock_inference_api.openai_chat_completion.call_args_list[0]
@@ -284,11 +286,17 @@ async def test_create_openai_response_with_tool_call_type_none(openai_responses_
     assert chunks[0].type == "response.created"
     assert len(chunks[0].response.output) == 0
 
+    # Check streaming events
+    assert chunks[1].type == "response.output_item.added"
+    assert chunks[2].type == "response.function_call_arguments.delta"
+    assert chunks[3].type == "response.function_call_arguments.done"
+    assert chunks[4].type == "response.output_item.done"
+
     # Check response.completed event (should have the tool call)
-    assert chunks[1].type == "response.completed"
-    assert len(chunks[1].response.output) == 1
-    assert chunks[1].response.output[0].type == "function_call"
-    assert chunks[1].response.output[0].name == "get_weather"
+    assert chunks[5].type == "response.completed"
+    assert len(chunks[5].response.output) == 1
+    assert chunks[5].response.output[0].type == "function_call"
+    assert chunks[5].response.output[0].name == "get_weather"
 
 
 async def test_create_openai_response_with_multiple_messages(openai_responses_impl, mock_inference_api):