Repository: phidatahq/phidata
File: libs/agno/tests/integration/agent/test_session.py (Python)

Code before:
        return f"The shopping list is now {session_state['shopping_list']}"

    agent = Agent(
        model=OpenAIChat(id="gpt-4o-mini"),
        db=shared_db,
        session_id=str(uuid.uuid4()),
        session_state={"shopping_list": []},
        tools=[add_item],
        markdown=True,
    )

    response = agent.run("Add apples to my shopping list")

    # Verify RunOutput has session_state field (structure test)
    assert response.session_state is not None, "RunOutput should have session_state"
    assert isinstance(response.session_state, dict), "session_state should be a dict"
    assert "shopping_list" in response.session_state, "shopping_list key should be present"
    assert isinstance(response.session_state["shopping_list"], list), "shopping_list should be a list"

    # If tools were executed, verify state was updated
    if response.tools and len(response.tools) > 0:
        tool_was_called = any(tool.tool_name == "add_item" for tool in response.tools)
        if tool_was_called:
            assert len(response.session_state.get("shopping_list", [])) > 0, (
                "Shopping list should have items if tool was called"
            )


def test_session_state_in_run_completed_event_stream(shared_db):
    """Test that RunCompletedEvent contains session_state in streaming mode."""

    # Reuse the existing shopping list pattern
    def add_item(session_state: Dict[str, Any], item: str) -> str:
        """Add an item to the shopping list."""
        session_state["shopping_list"].append(item)
        return f"The shopping list is now {session_state['shopping_list']}"

    agent = Agent(
        model=OpenAIChat(id="gpt-4o-mini"),
        db=shared_db,
        session_id=str(uuid.uuid4()),
        session_state={"shopping_list": ["bananas"]},
        tools=[add_item],
        markdown=True,
    )

    run_completed_event = None
    tool_called_in_stream = False

    for event in agent.run("Add oranges to my shopping list", stream=True, stream_intermediate_steps=True):
        # Check if tool was called during streaming

Review comment (bug):
We need to fail if the Agent didn't call the tools, as we can't test if the session_state management happens as expected.

This is the "generative test" problem. We are depending on the model calling the tool to test something unrelated... we could mock the run itself to enforce the call to `add_item`. That would fix it

This applies to all these tests

Code after:
        session_state={"shopping_list": ["bananas"]},
        tools=[add_item],
        instructions="You help manage shopping lists.",
        markdown=True,
    )

    run_completed_event = None

    for event in agent.run("Add oranges to my shopping list", stream=True, stream_intermediate_steps=True):
        if hasattr(event, "event") and event.event == "RunCompleted":
            run_completed_event = event
            break

    # Verify RunCompletedEvent structure
    assert run_completed_event is not None, "Should receive RunCompleted event"
    assert run_completed_event.session_state is not None, "RunCompletedEvent should have session_state"
    assert isinstance(run_completed_event.session_state, dict), "session_state should be a dict"
    assert "shopping_list" in run_completed_event.session_state, "shopping_list key should be present"
    assert "bananas" in run_completed_event.session_state.get("shopping_list", []), "Initial item should be preserved"

    # Verify state was updated by the tool
    assert len(run_completed_event.session_state.get("shopping_list", [])) == 2, "Shopping list should have 2 items"
    assert "oranges" in run_completed_event.session_state["shopping_list"], "Shopping list should contain oranges"


async def test_session_state_in_run_output_async(shared_db):
    """Test that RunOutput contains session_state in async non-streaming mode."""
    session_id = str(uuid.uuid4())
    agent = Agent(
        db=shared_db,
        session_id=session_id,
        session_state={"shopping_list": []},
        tools=[async_add_item],
        instructions="You help manage shopping lists.",
        markdown=True,
    )

    response = await agent.arun("Add apples to my shopping list")

    # Verify RunOutput has session_state
    assert response.session_state is not None, "RunOutput should have session_state"
    assert isinstance(response.session_state, dict), "session_state should be a dict"
    assert "shopping_list" in response.session_state, "shopping_list key should be present"
    assert isinstance(response.session_state["shopping_list"], list), "shopping_list should be a list"

    # Verify state was updated by the tool
    assert len(response.session_state.get("shopping_list", [])) == 1, "Shopping list should have 1 item"
    assert "apples" in response.session_state["shopping_list"], "Shopping list should contain apples"


async def test_session_state_in_run_completed_event_stream_async(shared_db):