/tests/integration_test_conversations_sdk.py

"""
Integration tests for the Conversations API using the SDK.
"""

import asyncio
import logging
import uuid
from typing import Any, List, Optional

import pytest
import requests
from letta_client import APIError, AsyncLetta, Letta
from letta_client.types import MessageCreateParam

logger = logging.getLogger(__name__)

TEST_MODEL_HANDLE = "anthropic/claude-haiku-4-5"


@pytest.fixture
def async_client(server_url: str) -> AsyncLetta:
    """Create an async Letta client."""
    return AsyncLetta(base_url=server_url)


@pytest.fixture
def otid_test_agent(client: Letta):
    """Create a test agent for otid/lock tests using openai/gpt-4o-mini (always available)."""
    agent_state = client.agents.create(
        name=f"test_conversations_{uuid.uuid4().hex[:8]}",
        model="openai/gpt-4o-mini",
        embedding="openai/text-embedding-3-small",
        memory_blocks=[
            {"label": "human", "value": "Test user"},
            {"label": "persona", "value": "You are a helpful assistant."},
        ],
    )
    yield agent_state
    client.agents.delete(agent_id=agent_state.id)


@pytest.fixture
def client(server_url: str) -> Letta:
    """Create a Letta client."""
    return Letta(base_url=server_url)


@pytest.fixture
def agent(client: Letta):
    """Create a test agent."""
    agent_state = client.agents.create(
        name=f"test_conversations_{uuid.uuid4().hex[:8]}",
        model=TEST_MODEL_HANDLE,
        memory_blocks=[
            {"label": "human", "value": "Test user"},
            {"label": "persona", "value": "You are a helpful assistant."},
        ],
    )
    yield agent_state
    # Cleanup
    client.agents.delete(agent_id=agent_state.id)


class TestConversationsSDK:
    """Test conversations using the SDK client."""

    @staticmethod
    def _get_system_message_content(server_url: str, message_id: str) -> str:
        """Fetch a system message by ID via REST and return its content."""
        response = requests.get(f"{server_url}/v1/messages/{message_id}")
        assert response.status_code == 200, f"Failed to retrieve message {message_id}: {response.text}"

        payload = response.json()
        assert isinstance(payload, list) and len(payload) == 1, f"Expected one message for {message_id}, got: {payload}"
        message = payload[0]
        assert message.get("message_type") == "system_message", f"Expected system message, got: {message}"
        assert isinstance(message.get("content"), str), f"Expected string content, got: {message}"
        return message["content"]

    @staticmethod
    def _assert_system_metadata_identifiers(content: str, agent_id: str, conversation_id: str):
        """Assert system prompt metadata includes expected IDs and excludes deprecated date line."""
        assert "<memory_metadata>" in content
        assert f"- AGENT_ID: {agent_id}" in content
        assert f"- CONVERSATION_ID: {conversation_id}" in content
        assert "- System prompt last recompiled:" in content
        assert "- The current system date is:" not in content

    def test_create_conversation(self, client: Letta, agent):
        """Test creating a conversation for an agent."""
        conversation = client.conversations.create(agent_id=agent.id)

        assert conversation.id is not None
        assert conversation.id.startswith("conv-")
        assert conversation.agent_id == agent.id

    def test_list_conversations(self, client: Letta, agent):
        """Test listing conversations for an agent."""
        # Create multiple conversations
        conv1 = client.conversations.create(agent_id=agent.id)
        conv2 = client.conversations.create(agent_id=agent.id)

        # List conversations
        conversations = client.conversations.list(agent_id=agent.id)

        assert len(conversations) >= 2
        conv_ids = [c.id for c in conversations]
        assert conv1.id in conv_ids
        assert conv2.id in conv_ids

    def test_list_conversations_sort_by_last_message_at(self, client: Letta, agent):
        """Test listing conversations sorted by last_message_at."""
        conv_older = client.conversations.create(agent_id=agent.id)
        conv_newer = client.conversations.create(agent_id=agent.id)

        list(
            client.conversations.messages.create(
                conversation_id=conv_older.id,
                messages=[{"role": "user", "content": "hello older conversation"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conv_newer.id,
                messages=[{"role": "user", "content": "hello newer conversation"}],
            )
        )

        sorted_desc = client.conversations.list(agent_id=agent.id, order_by="last_message_at", order="desc")
        sorted_ids = [c.id for c in sorted_desc]

        assert conv_older.id in sorted_ids
        assert conv_newer.id in sorted_ids
        assert sorted_ids.index(conv_newer.id) < sorted_ids.index(conv_older.id)

    def test_retrieve_conversation(self, client: Letta, agent):
        """Test retrieving a specific conversation."""
        # Create a conversation
        created = client.conversations.create(agent_id=agent.id)

        # Retrieve it (should have system message from creation)
        retrieved = client.conversations.retrieve(conversation_id=created.id)

        assert retrieved.id == created.id
        assert retrieved.agent_id == created.agent_id
        # Conversation should have 1 system message immediately after creation
        assert len(retrieved.in_context_message_ids) == 1
        assert retrieved.in_context_message_ids[0].startswith("message-")

        # Send a message to the conversation
        list(
            client.conversations.messages.create(
                conversation_id=created.id,
                messages=[{"role": "user", "content": "Hello!"}],
            )
        )

        # Retrieve again and check in_context_message_ids is populated
        retrieved_with_messages = client.conversations.retrieve(conversation_id=created.id)

        # System message + user + assistant messages should be in the conversation
        assert len(retrieved_with_messages.in_context_message_ids) >= 3  # system + user + assistant
        # All IDs should be strings starting with "message-"
        for msg_id in retrieved_with_messages.in_context_message_ids:
            assert isinstance(msg_id, str)
            assert msg_id.startswith("message-")

        # Verify message ordering by listing messages in ascending order (oldest first)
        messages = list(client.conversations.messages.list(conversation_id=created.id, order="asc"))
        assert len(messages) >= 3  # system + user + assistant
        # First message should be system message for this conversation.
        assert messages[0].message_type == "system_message", f"First message should be system_message, got {messages[0].message_type}"
        # Second message should be user message
        assert messages[1].message_type == "user_message", f"Second message should be user_message, got {messages[1].message_type}"

    def test_default_system_message_metadata_has_agent_and_default_conversation(self, client: Letta, agent, server_url: str):
        """Default conversation system prompt should include AGENT_ID and CONVERSATION_ID=default."""
        assert agent.message_ids and len(agent.message_ids) > 0, "Agent should have an initial system message"

        content = self._get_system_message_content(server_url=server_url, message_id=agent.message_ids[0])
        self._assert_system_metadata_identifiers(content=content, agent_id=agent.id, conversation_id="default")

    def test_conversation_system_message_metadata_has_agent_and_conversation_id(self, client: Letta, agent, server_url: str):
        """Conversation-scoped system prompt should include AGENT_ID and the concrete conversation ID."""
        conversation = client.conversations.create(agent_id=agent.id)
        retrieved = client.conversations.retrieve(conversation_id=conversation.id)

        assert retrieved.in_context_message_ids and len(retrieved.in_context_message_ids) > 0
        system_message_id = retrieved.in_context_message_ids[0]

        content = self._get_system_message_content(server_url=server_url, message_id=system_message_id)
        self._assert_system_metadata_identifiers(content=content, agent_id=agent.id, conversation_id=conversation.id)

    def test_client_skills_are_rendered_in_conversation_system_prompt(self, client: Letta, agent, server_url: str):
        """Client skills are request-scoped: injected at LLM request time, never persisted to DB system message."""
        conversation = client.conversations.create(agent_id=agent.id)

        client_skills = [
            {
                "name": "debugging-checklist",
                "description": "Use this skill to debug stream wiring issues.",
                "location": "/tmp/.skills/debugging-checklist/SKILL.md",
            },
            {
                "name": "api-ops",
                "description": "Operational API runbook.",
                "location": "/tmp/.skills/api-ops/SKILL.md",
            },
        ]

        response_messages = list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[
                    {
                        "role": "user",
                        "content": "What skills are available? Provide the exact name for each skill.",
                    }
                ],
                client_skills=client_skills,
            )
        )

        # Persisted system message should NOT contain skills (they are request-scoped only)
        retrieved_after_first_turn = client.conversations.retrieve(conversation_id=conversation.id)
        first_system_message_id = retrieved_after_first_turn.in_context_message_ids[0]
        first_system_content = self._get_system_message_content(server_url=server_url, message_id=first_system_message_id)

        assert "<available_skills>" not in first_system_content, "Skills should not be persisted in the DB system message"

        # Agent should still see skills at request time and respond about them
        assistant_messages = [m for m in response_messages if getattr(m, "message_type", None) == "assistant_message"]
        assert assistant_messages, "Expected at least one assistant message in response stream"

        assistant_content = " ".join(m.content for m in assistant_messages if isinstance(getattr(m, "content", None), str))
        assert "debugging-checklist" in assistant_content
        assert "api-ops" in assistant_content

        # Second turn: persisted system prompt should remain stable (no drift)
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[
                    {
                        "role": "user",
                        "content": "Repeat the skill names again.",
                    }
                ],
                client_skills=client_skills,
            )
        )

        retrieved_after_second_turn = client.conversations.retrieve(conversation_id=conversation.id)
        second_system_message_id = retrieved_after_second_turn.in_context_message_ids[0]
        second_system_content = self._get_system_message_content(server_url=server_url, message_id=second_system_message_id)

        assert second_system_content == first_system_content

    def test_send_message_to_conversation(self, client: Letta, agent):
        """Test sending a message to a conversation."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send a message (returns a stream)
        stream = client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=[{"role": "user", "content": "Hello, how are you?"}],
        )

        # Consume the stream to get messages
        messages = list(stream)

        # Check response contains messages
        assert len(messages) > 0
        # Should have at least an assistant message
        message_types = [m.message_type for m in messages if hasattr(m, "message_type")]
        assert "assistant_message" in message_types

    def test_list_conversation_messages(self, client: Letta, agent):
        """Test listing messages from a conversation."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send a message to create some history (consume the stream)
        stream = client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=[{"role": "user", "content": "Say 'test response' back to me."}],
        )
        list(stream)  # Consume stream

        # List messages
        messages = list(client.conversations.messages.list(conversation_id=conversation.id))

        assert len(messages) >= 2  # At least user + assistant
        message_types = [m.message_type for m in messages]
        assert "user_message" in message_types
        assert "assistant_message" in message_types

        # Send another message and check that old and new messages are both listed
        first_message_count = len(messages)
        stream = client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=[{"role": "user", "content": "This is a follow-up message."}],
        )
        list(stream)  # Consume stream

        # List messages again
        updated_messages = list(client.conversations.messages.list(conversation_id=conversation.id))

        # Should have more messages now (at least 2 more: user + assistant)
        assert len(updated_messages) >= first_message_count + 2

    def test_conversation_isolation(self, client: Letta, agent):
        """Test that conversations are isolated from each other."""
        # Create two conversations
        conv1 = client.conversations.create(agent_id=agent.id)
        conv2 = client.conversations.create(agent_id=agent.id)

        # Send different messages to each (consume streams)
        list(
            client.conversations.messages.create(
                conversation_id=conv1.id,
                messages=[{"role": "user", "content": "Remember the word: APPLE"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conv2.id,
                messages=[{"role": "user", "content": "Remember the word: BANANA"}],
            )
        )

        # List messages from each conversation
        conv1_messages = list(client.conversations.messages.list(conversation_id=conv1.id))
        conv2_messages = list(client.conversations.messages.list(conversation_id=conv2.id))

        # Check messages are separate
        conv1_content = " ".join([m.content for m in conv1_messages if hasattr(m, "content") and m.content])
        conv2_content = " ".join([m.content for m in conv2_messages if hasattr(m, "content") and m.content])

        assert "APPLE" in conv1_content
        assert "BANANA" in conv2_content
        # Each conversation should only have its own word
        assert "BANANA" not in conv1_content or "APPLE" not in conv2_content

        # Ask what word was remembered and make sure it's different for each conversation
        conv1_recall = list(
            client.conversations.messages.create(
                conversation_id=conv1.id,
                messages=[{"role": "user", "content": "What word did I ask you to remember? Reply with just the word."}],
            )
        )
        conv2_recall = list(
            client.conversations.messages.create(
                conversation_id=conv2.id,
                messages=[{"role": "user", "content": "What word did I ask you to remember? Reply with just the word."}],
            )
        )

        # Get the assistant responses
        conv1_response = " ".join([m.content for m in conv1_recall if hasattr(m, "message_type") and m.message_type == "assistant_message"])
        conv2_response = " ".join([m.content for m in conv2_recall if hasattr(m, "message_type") and m.message_type == "assistant_message"])

        assert "APPLE" in conv1_response.upper(), f"Conv1 should remember APPLE, got: {conv1_response}"
        assert "BANANA" in conv2_response.upper(), f"Conv2 should remember BANANA, got: {conv2_response}"

        # Each conversation has its own system message (created on first message)
        conv1_system_id = conv1_messages[0].id
        conv2_system_id = conv2_messages[0].id
        assert conv1_system_id != conv2_system_id, "System messages should have different IDs for different conversations"

    def test_fork_conversation_via_rest_shares_non_system_message_ids(self, client: Letta, agent, server_url: str):
        """Fork via REST and verify memory continuity plus shared non-system message IDs in-order."""
        source = client.conversations.create(agent_id=agent.id)

        list(
            client.conversations.messages.create(
                conversation_id=source.id,
                messages=[{"role": "user", "content": "my favorite color is red"}],
            )
        )

        fork_response = requests.post(f"{server_url}/v1/conversations/{source.id}/fork")
        assert fork_response.status_code == 200, f"Fork request failed: {fork_response.text}"
        forked_conversation = fork_response.json()
        forked_id = forked_conversation["id"]

        fork_reply_stream = list(
            client.conversations.messages.create(
                conversation_id=forked_id,
                messages=[{"role": "user", "content": "whats my favorite color?"}],
            )
        )

        fork_reply_text = " ".join(
            m.content
            for m in fork_reply_stream
            if hasattr(m, "message_type") and m.message_type == "assistant_message" and isinstance(getattr(m, "content", None), str)
        )
        assert "red" in fork_reply_text.lower(), f"Expected forked conversation to remember red, got: {fork_reply_text}"

        source_messages = list(client.conversations.messages.list(conversation_id=source.id, order="asc"))
        forked_messages = list(client.conversations.messages.list(conversation_id=forked_id, order="asc"))

        source_message_ids = [m.id for m in source_messages]
        forked_message_ids = [m.id for m in forked_messages]

        # System messages should differ across source and fork.
        assert source_message_ids[0] != forked_message_ids[0]

        # All non-system source messages should be reused in-order in the fork.
        source_non_system_ids = source_message_ids[1:]
        forked_after_system_ids = forked_message_ids[1:]
        assert source_non_system_ids == forked_after_system_ids[: len(source_non_system_ids)]
        assert set(source_non_system_ids).issubset(set(forked_message_ids))

    def test_fork_conversation_via_rest_not_found(self, client: Letta, agent, server_url: str):
        """Forking a non-existent conversation should return not found/validation error."""
        response = requests.post(f"{server_url}/v1/conversations/conv-nonexistent-00000000/fork")
        assert response.status_code in (404, 422), f"Expected 404/422, got {response.status_code}: {response.text}"

    def test_conversation_messages_pagination(self, client: Letta, agent):
        """Test pagination when listing conversation messages."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send multiple messages to create history (consume streams)
        for i in range(3):
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": f"Message number {i}"}],
                )
            )

        # List all messages to get the total count
        all_messages = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
            )
        )
        total_count = len(all_messages)

        # List with limit — access the first page directly (not list() which auto-paginates)
        messages_page = client.conversations.messages.list(
            conversation_id=conversation.id,
            limit=2,
        )

        # The first page should have fewer messages than the total
        assert len(messages_page.items) < total_count

    def test_retrieve_conversation_stream_no_active_run(self, client: Letta, agent):
        """Test that retrieve_conversation_stream returns error when no active run exists."""
        from letta_client import BadRequestError

        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Try to retrieve stream when no run exists (should fail)
        with pytest.raises(BadRequestError) as exc_info:
            # Use the SDK's stream method
            stream = client.conversations.messages.stream(conversation_id=conversation.id)
            list(stream)  # Consume the stream to trigger the error

        # Should return 400 because no active run exists
        assert "No active runs found" in str(exc_info.value)

    def test_retrieve_conversation_stream_after_completed_run(self, client: Letta, agent):
        """Test that retrieve_conversation_stream returns error when run is completed."""
        from letta_client import BadRequestError

        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send a message (this creates a run that completes)
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Hello"}],
            )
        )

        # Try to retrieve stream after the run has completed (should fail)
        with pytest.raises(BadRequestError) as exc_info:
            # Use the SDK's stream method
            stream = client.conversations.messages.stream(conversation_id=conversation.id)
            list(stream)  # Consume the stream to trigger the error

        # Should return 400 because the run was not created in background mode
        assert "not created in background mode" in str(exc_info.value) or "No active runs found" in str(exc_info.value)

    def test_conversation_lock_released_after_completion(self, client: Letta, agent):
        """Test that lock is released after request completes by sending sequential messages."""
        from letta.settings import settings

        # Skip if Redis is not configured
        if settings.redis_host is None or settings.redis_port is None:
            pytest.skip("Redis not configured - skipping conversation lock test")

        conversation = client.conversations.create(agent_id=agent.id)

        # Send first message (should acquire and release lock)
        messages1 = list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Hello"}],
            )
        )
        assert len(messages1) > 0

        # Send second message - should succeed if lock was released
        messages2 = list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Hello again"}],
            )
        )
        assert len(messages2) > 0

    def test_conversation_lock_released_on_error(self, client: Letta, agent):
        """Test that lock is released even when the run encounters an error.

        This test sends a message that triggers an error during streaming (by causing
        a context window exceeded error with a very long message), then verifies the
        lock is properly released by successfully sending another message.
        """
        from letta.settings import settings

        # Skip if Redis is not configured
        if settings.redis_host is None or settings.redis_port is None:
            pytest.skip("Redis not configured - skipping conversation lock test")

        conversation = client.conversations.create(agent_id=agent.id)

        # Try to send a message that will cause an error during processing
        # We use an extremely long message to trigger a context window error
        very_long_message = "Hello " * 100000  # Very long message to exceed context window

        try:
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": very_long_message}],
                )
            )
        except Exception:
            pass  # Expected to fail due to context window exceeded

        # Send another message - should succeed if lock was released after error
        messages = list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Hello after error"}],
            )
        )
        assert len(messages) > 0, "Lock should be released even after run error"

    def test_concurrent_messages_to_same_conversation(self, client: Letta, agent):
        """Test that concurrent messages to the same conversation are properly serialized.

        One request should succeed and one should get a 409 CONVERSATION_BUSY error.
        After both return, a subsequent message should succeed.
        """
        import concurrent.futures

        from letta_client import ConflictError

        from letta.settings import settings

        # Skip if Redis is not configured
        if settings.redis_host is None or settings.redis_port is None:
            pytest.skip("Redis not configured - skipping conversation lock test")

        conversation = client.conversations.create(agent_id=agent.id)

        results = {"success": 0, "conflict": 0, "other_error": 0}

        def send_message(msg: str):
            try:
                messages = list(
                    client.conversations.messages.create(
                        conversation_id=conversation.id,
                        messages=[{"role": "user", "content": msg}],
                    )
                )
                return ("success", messages)
            except ConflictError:
                return ("conflict", None)
            except Exception as e:
                return ("other_error", str(e))

        # Fire off two messages concurrently
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
            future1 = executor.submit(send_message, "Message 1")
            future2 = executor.submit(send_message, "Message 2")

            result1 = future1.result()
            result2 = future2.result()

        # Count results
        for result_type, _ in [result1, result2]:
            results[result_type] += 1

        # One should succeed and one should get conflict
        assert results["success"] == 1, f"Expected 1 success, got {results['success']}"
        assert results["conflict"] == 1, f"Expected 1 conflict, got {results['conflict']}"
        assert results["other_error"] == 0, f"Unexpected errors: {results['other_error']}"

        # Now send another message - should succeed since lock is released
        messages = list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Message after concurrent requests"}],
            )
        )
        assert len(messages) > 0, "Should be able to send message after concurrent requests complete"

    def test_list_conversation_messages_order_asc(self, client: Letta, agent):
        """Test listing messages in ascending order (oldest first)."""
        conversation = client.conversations.create(agent_id=agent.id)

        # Send messages to create history
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "First message"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Second message"}],
            )
        )

        # List messages in ascending order (oldest first)
        messages_asc = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
            )
        )

        # First message should be system message (oldest)
        assert messages_asc[0].message_type == "system_message"

        # Get user messages and verify order
        user_messages = [m for m in messages_asc if m.message_type == "user_message"]
        assert len(user_messages) >= 2
        # First user message should contain "First message"
        assert "First" in user_messages[0].content

    def test_list_conversation_messages_order_desc(self, client: Letta, agent):
        """Test listing messages in descending order (newest first)."""
        conversation = client.conversations.create(agent_id=agent.id)

        # Send messages to create history
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "First message"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Second message"}],
            )
        )

        # List messages in descending order (newest first) - this is the default
        messages_desc = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="desc",
            )
        )

        # Get user messages and verify order
        user_messages = [m for m in messages_desc if m.message_type == "user_message"]
        assert len(user_messages) >= 2
        # First user message in desc order should contain "Second message" (newest)
        assert "Second" in user_messages[0].content

    def test_list_conversation_messages_order_affects_pagination(self, client: Letta, agent):
        """Test that order parameter affects pagination correctly."""
        conversation = client.conversations.create(agent_id=agent.id)

        # Send multiple messages
        for i in range(3):
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": f"Message {i}"}],
                )
            )

        # Get all messages in descending order with limit
        messages_desc = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="desc",
                limit=5,
            )
        )

        # Get all messages in ascending order with limit
        messages_asc = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
                limit=5,
            )
        )

        # The first messages should be different based on order
        assert messages_desc[0].id != messages_asc[0].id

    def test_list_conversation_messages_with_before_cursor(self, client: Letta, agent):
        """Test pagination with before cursor."""
        conversation = client.conversations.create(agent_id=agent.id)

        # Send messages to create history
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "First message"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Second message"}],
            )
        )

        # Get all messages first
        all_messages = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
            )
        )
        assert len(all_messages) >= 4  # system + user + assistant + user + assistant

        # Use the last message ID as cursor
        last_message_id = all_messages[-1].id
        messages_before = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
                before=last_message_id,
            )
        )

        # Should have fewer messages (all except the last one)
        assert len(messages_before) < len(all_messages)
        # Should not contain the cursor message
        assert last_message_id not in [m.id for m in messages_before]

    def test_list_conversation_messages_with_after_cursor(self, client: Letta, agent):
        """Test pagination with after cursor."""
        conversation = client.conversations.create(agent_id=agent.id)

        # Send messages to create history
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "First message"}],
            )
        )
        list(
            client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=[{"role": "user", "content": "Second message"}],
            )
        )

        # Get all messages first
        all_messages = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
            )
        )
        assert len(all_messages) >= 4

        # Use the first message ID as cursor
        first_message_id = all_messages[0].id
        messages_after = list(
            client.conversations.messages.list(
                conversation_id=conversation.id,
                order="asc",
                after=first_message_id,
            )
        )

        # Should have fewer messages (all except the first one)
        assert len(messages_after) < len(all_messages)
        # Should not contain the cursor message
        assert first_message_id not in [m.id for m in messages_after]

    def test_agent_direct_messaging_via_conversations_endpoint(self, client: Letta, agent):
        """Test sending messages using agent ID as conversation_id (agent-direct mode).

        This allows clients to use a unified endpoint pattern without managing conversation IDs.
        """
        # Send a message using the agent ID directly as conversation_id
        # This should route to agent-direct mode with locking
        messages = list(
            client.conversations.messages.create(
                conversation_id=agent.id,  # Using agent ID instead of conversation ID
                messages=[{"role": "user", "content": "Hello via agent-direct mode!"}],
            )
        )

        # Verify we got a response
        assert len(messages) > 0, "Should receive response messages"

        # Verify we got an assistant message in the response
        assistant_messages = [m for m in messages if hasattr(m, "message_type") and m.message_type == "assistant_message"]
        assert len(assistant_messages) > 0, "Should receive at least one assistant message"

    def test_agent_direct_messaging_with_locking(self, client: Letta, agent):
        """Test that agent-direct mode properly acquires and releases locks.

        Sequential requests should both succeed if locks are properly released.
        """
        from letta.settings import settings

        # Skip if Redis is not configured
        if settings.redis_host is None or settings.redis_port is None:
            pytest.skip("Redis not configured - skipping agent-direct lock test")

        # Send first message via agent-direct mode
        messages1 = list(
            client.conversations.messages.create(
                conversation_id=agent.id,
                messages=[{"role": "user", "content": "First message"}],
            )
        )
        assert len(messages1) > 0, "First message should succeed"

        # Send second message - should succeed if lock was released
        messages2 = list(
            client.conversations.messages.create(
                conversation_id=agent.id,
                messages=[{"role": "user", "content": "Second message"}],
            )
        )
        assert len(messages2) > 0, "Second message should succeed after lock released"

    def test_agent_direct_concurrent_requests_blocked(self, client: Letta, agent):
        """Test that concurrent requests to agent-direct mode are properly serialized.

        One request should succeed and one should get a 409 CONVERSATION_BUSY error.
        """
        import concurrent.futures

        from letta_client import ConflictError

        from letta.settings import settings

        # Skip if Redis is not configured
        if settings.redis_host is None or settings.redis_port is None:
            pytest.skip("Redis not configured - skipping agent-direct lock test")

        results = {"success": 0, "conflict": 0, "other_error": 0}

        def send_message(msg: str):
            try:
                messages = list(
                    client.conversations.messages.create(
                        conversation_id=agent.id,  # Agent-direct mode
                        messages=[{"role": "user", "content": msg}],
                    )
                )
                return ("success", messages)
            except ConflictError:
                return ("conflict", None)
            except Exception as e:
                return ("other_error", str(e))

        # Fire off two messages concurrently
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
            future1 = executor.submit(send_message, "Concurrent message 1")
            future2 = executor.submit(send_message, "Concurrent message 2")

            result1 = future1.result()
            result2 = future2.result()

        # Count results
        for result_type, _ in [result1, result2]:
            results[result_type] += 1

        # One should succeed and one should get conflict
        assert results["success"] == 1, f"Expected 1 success, got {results['success']}"
        assert results["conflict"] == 1, f"Expected 1 conflict, got {results['conflict']}"
        assert results["other_error"] == 0, f"Unexpected errors: {results['other_error']}"

        # Now send another message - should succeed since lock is released
        messages = list(
            client.conversations.messages.create(
                conversation_id=agent.id,
                messages=[{"role": "user", "content": "Message after concurrent requests"}],
            )
        )
        assert len(messages) > 0, "Should be able to send message after concurrent requests complete"

    def test_agent_direct_list_messages(self, client: Letta, agent):
        """Test listing messages using agent ID as conversation_id."""
        # First send a message via agent-direct mode
        list(
            client.conversations.messages.create(
                conversation_id=agent.id,
                messages=[{"role": "user", "content": "Test message for listing"}],
            )
        )

        # List messages using agent ID
        messages = list(client.conversations.messages.list(conversation_id=agent.id))

        # Should have messages (at least system + user + assistant)
        assert len(messages) >= 3, f"Expected at least 3 messages, got {len(messages)}"

        # Verify we can find our test message
        user_messages = [m for m in messages if hasattr(m, "message_type") and m.message_type == "user_message"]
        assert any("Test message for listing" in str(m.content) for m in user_messages), "Should find our test message"

    def test_agent_direct_cancel(self, client: Letta, agent):
        """Test canceling runs using agent ID as conversation_id."""
        from letta.settings import settings

        # Skip if run tracking is disabled
        if not settings.track_agent_run:
            pytest.skip("Run tracking disabled - skipping cancel test")

        # Start a background request that we can cancel
        try:
            # Send a message in background mode
            stream = client.conversations.messages.create(
                conversation_id=agent.id,
                messages=[{"role": "user", "content": "Background message to cancel"}],
                background=True,
            )
            # Consume a bit of the stream to ensure it started
            next(iter(stream), None)

            # Cancel using agent ID
            result = client.conversations.cancel(conversation_id=agent.id)

            # Should return results (may be empty if run already completed)
            assert isinstance(result, dict), "Cancel should return a dict of results"
        except Exception as e:
            # If no active runs, that's okay - the run may have completed quickly
            if "No active runs" not in str(e):
                raise

    def test_backwards_compatibility_old_pattern(self, client: Letta, agent, server_url: str):
        """Test that the old pattern (agent_id as conversation_id) still works for backwards compatibility."""
        # OLD PATTERN: conversation_id=agent.id (should still work)
        # Use raw HTTP requests since SDK might not be up to date

        # Test 1: Send message using old pattern
        response = requests.post(
            f"{server_url}/v1/conversations/{agent.id}/messages",
            json={
                "messages": [{"role": "user", "content": "Testing old pattern still works"}],
                "streaming": False,
            },
        )
        assert response.status_code == 200, f"Old pattern should work for sending messages: {response.text}"
        data = response.json()
        assert "messages" in data, "Response should contain messages"
        assert len(data["messages"]) > 0, "Should receive response messages"

        # Test 2: List messages using old pattern
        response = requests.get(f"{server_url}/v1/conversations/{agent.id}/messages")
        assert response.status_code == 200, f"Old pattern should work for listing messages: {response.text}"
        data = response.json()
        # Response is a list of messages directly
        assert isinstance(data, list), "Response should be a list of messages"
        assert len(data) >= 3, "Should have at least system + user + assistant messages"

    def test_new_pattern_default_system_message_metadata(self, client: Letta, agent, server_url: str):
        """Agent-direct default path should render system metadata with AGENT_ID and CONVERSATION_ID=default."""
        # Trigger default conversation path via new pattern
        response = requests.post(
            f"{server_url}/v1/conversations/default/messages",
            json={
                "agent_id": agent.id,
                "messages": [{"role": "user", "content": "Please acknowledge."}],
                "streaming": False,
            },
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

        # Read default conversation messages and inspect the first (system) message content
        list_response = requests.get(
            f"{server_url}/v1/conversations/default/messages",
            params={"agent_id": agent.id, "order": "asc"},
        )
        assert list_response.status_code == 200, f"Expected 200, got {list_response.status_code}: {list_response.text}"

        messages = list_response.json()
        assert isinstance(messages, list) and len(messages) > 0, f"Expected non-empty message list, got: {messages}"
        assert messages[0].get("message_type") == "system_message", f"Expected first message to be system_message, got: {messages[0]}"

        content = messages[0].get("content")
        assert isinstance(content, str), f"Expected system message content string, got: {messages[0]}"

        self._assert_system_metadata_identifiers(content=content, agent_id=agent.id, conversation_id="default")

    def test_new_pattern_send_message(self, client: Letta, agent, server_url: str):
        """Test sending messages using the new pattern: conversation_id='default' + agent_id in body."""
        # NEW PATTERN: conversation_id='default' + agent_id in request body
        response = requests.post(
            f"{server_url}/v1/conversations/default/messages",
            json={
                "agent_id": agent.id,
                "messages": [{"role": "user", "content": "Testing new pattern send message"}],
                "streaming": False,
            },
        )
        assert response.status_code == 200, f"New pattern should work for sending messages: {response.text}"
        data = response.json()
        assert "messages" in data, "Response should contain messages"
        assert len(data["messages"]) > 0, "Should receive response messages"

        # Default conversation remains virtual, so returned messages should still have no conversation_id.
        for message in data["messages"]:
            assert message.get("conversation_id") is None

        # Verify we got an assistant message
        assistant_messages = [m for m in data["messages"] if m.get("message_type") == "assistant_message"]
        assert len(assistant_messages) > 0, "Should receive at least one assistant message"

    def test_new_pattern_list_messages(self, client: Letta, agent, server_url: str):
        """Test listing messages using the new pattern: conversation_id='default' + agent_id query param."""
        # First send a message to populate the conversation
        requests.post(
            f"{server_url}/v1/conversations/{agent.id}/messages",
            json={
                "messages": [{"role": "user", "content": "Setup message for list test"}],
                "streaming": False,
            },
        )

        # NEW PATTERN: conversation_id='default' + agent_id as query param
        response = requests.get(
            f"{server_url}/v1/conversations/default/messages",
            params={"agent_id": agent.id},
        )
        assert response.status_code == 200, f"New pattern should work for listing messages: {response.text}"
        data = response.json()
        # Response is a list of messages directly
        assert isinstance(data, list), "Response should be a list of messages"
        assert len(data) >= 3, "Should have at least system + user + assistant messages"

    def test_new_pattern_cancel(self, client: Letta, agent, server_url: str):
        """Test canceling runs using the new pattern: conversation_id='default' + agent_id query param."""
        from letta.settings import settings

        if not settings.track_agent_run:
            pytest.skip("Run tracking disabled - skipping cancel test")

        # NEW PATTERN: conversation_id='default' + agent_id as query param
        response = requests.post(
            f"{server_url}/v1/conversations/default/cancel",
            params={"agent_id": agent.id},
        )
        # Returns 200 with results if runs exist, or 409 if no active runs
        assert response.status_code in [200, 409], f"New pattern should work for cancel: {response.text}"
        if response.status_code == 200:
            data = response.json()
            assert isinstance(data, dict), "Cancel should return a dict"

    def test_new_pattern_compact(self, client: Letta, agent, server_url: str):
        """Test compacting conversation using the new pattern: conversation_id='default' + agent_id in body."""
        # Send many messages to have enough for compaction
        for i in range(10):
            requests.post(
                f"{server_url}/v1/conversations/{agent.id}/messages",
                json={
                    "messages": [{"role": "user", "content": f"Message {i} for compaction test"}],
                    "streaming": False,
                },
            )

        # NEW PATTERN: conversation_id='default' + agent_id in request body
        response = requests.post(
            f"{server_url}/v1/conversations/default/compact",
            json={"agent_id": agent.id},
        )
        # May return 200 (success) or 400 (not enough messages to compact)
        assert response.status_code in [200, 400], f"New pattern should accept agent_id parameter: {response.text}"
        if response.status_code == 200:
            data = response.json()
            assert "summary" in data, "Response should contain summary"
            assert "num_messages_before" in data, "Response should contain num_messages_before"
            assert "num_messages_after" in data, "Response should contain num_messages_after"

    def test_new_pattern_stream_retrieve(self, client: Letta, agent, server_url: str):
        """Test retrieving stream using the new pattern: conversation_id='default' + agent_id in body."""
        # NEW PATTERN: conversation_id='default' + agent_id in request body
        # Note: This will likely return 400 if no active run exists, which is expected
        response = requests.post(
            f"{server_url}/v1/conversations/default/stream",
            json={"agent_id": agent.id},
        )
        # Either 200 (if run exists) or 400 (no active run) are both acceptable
        assert response.status_code in [200, 400], f"Stream retrieve should accept new pattern: {response.text}"

    def test_conversation_system_override_forces_response_content(self, client: Letta, agent, server_url: str):
        """Conversation send should honor request.system override in the actual model response."""
        conversation = client.conversations.create(agent_id=agent.id)
        override_system = 'Always respond with a single word: "blue"'

        response = requests.post(
            f"{server_url}/v1/conversations/{conversation.id}/messages",
            json={
                "messages": [{"role": "user", "content": "hi"}],
                "streaming": False,
                "override_system": override_system,
            },
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
        payload = response.json()

        assistant_contents = [m.get("content", "") for m in payload.get("messages", []) if m.get("message_type") == "assistant_message"]
        combined_response = " ".join(c for c in assistant_contents if isinstance(c, str))
        assert "blue" in combined_response.lower(), f"Expected 'blue' in assistant response, got: {combined_response}"

    def test_default_conversation_system_override_forces_response_content(self, client: Letta, agent, server_url: str):
        """Agent-direct default conversation send should honor request.system override in the actual model response."""
        override_system = 'Always respond with a single word: "blue"'

        response = requests.post(
            f"{server_url}/v1/conversations/default/messages",
            json={
                "agent_id": agent.id,
                "messages": [{"role": "user", "content": "hi"}],
                "streaming": False,
                "override_system": override_system,
            },
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
        payload = response.json()

        assistant_contents = [m.get("content", "") for m in payload.get("messages", []) if m.get("message_type") == "assistant_message"]
        combined_response = " ".join(c for c in assistant_contents if isinstance(c, str))
        assert "blue" in combined_response.lower(), f"Expected 'blue' in assistant response, got: {combined_response}"

    def test_fork_default_conversation_via_rest(self, client: Letta, agent, server_url: str):
        """Fork the default (agent-direct) conversation via REST and verify message sharing."""
        # Send a message to the default conversation (agent-direct mode)
        requests.post(
            f"{server_url}/v1/conversations/default/messages",
            json={
                "agent_id": agent.id,
                "messages": [{"role": "user", "content": "my favorite animal is a penguin"}],
                "streaming": False,
            },
        )

        # Fork the default conversation using the new pattern
        fork_response = requests.post(
            f"{server_url}/v1/conversations/default/fork",
            params={"agent_id": agent.id},
        )
        assert fork_response.status_code == 200, f"Fork request failed: {fork_response.text}"
        forked_conversation = fork_response.json()
        forked_id = forked_conversation["id"]
        assert forked_id.startswith("conv-"), f"Expected conversation ID, got: {forked_id}"

        # Verify the forked conversation remembers the context
        fork_reply_response = requests.post(
            f"{server_url}/v1/conversations/{forked_id}/messages",
            json={
                "messages": [{"role": "user", "content": "what is my favorite animal?"}],
                "streaming": False,
            },
        )
        assert fork_reply_response.status_code == 200, f"Fork reply failed: {fork_reply_response.text}"
        fork_reply_data = fork_reply_response.json()

        fork_reply_text = " ".join(
            m.get("content", "")
            for m in fork_reply_data.get("messages", [])
            if m.get("message_type") == "assistant_message" and isinstance(m.get("content"), str)
        )
        assert "penguin" in fork_reply_text.lower(), f"Expected forked conversation to remember penguin, got: {fork_reply_text}"

        # Verify message IDs are shared (non-system messages)
        default_messages_response = requests.get(
            f"{server_url}/v1/conversations/default/messages",
            params={"agent_id": agent.id, "order": "asc"},
        )
        assert default_messages_response.status_code == 200
        default_messages = default_messages_response.json()

        forked_messages = list(client.conversations.messages.list(conversation_id=forked_id, order="asc"))

        default_message_ids = [m["id"] for m in default_messages]
        forked_message_ids = [m.id for m in forked_messages]

        # System messages should differ
        assert default_message_ids[0] != forked_message_ids[0], "System messages should be different"

        # Non-system messages from default should be in the fork
        default_non_system_ids = default_message_ids[1:]
        forked_after_system_ids = forked_message_ids[1:]
        assert default_non_system_ids == forked_after_system_ids[: len(default_non_system_ids)], (
            f"Non-system messages should match. Default: {default_non_system_ids}, Fork: {forked_after_system_ids}"
        )


class TestConversationDelete:
    """Tests for the conversation delete endpoint."""

    def test_delete_conversation(self, client: Letta, agent, server_url: str):
        """Test soft deleting a conversation."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)
        assert conversation.id is not None

        # Delete it via REST endpoint
        response = requests.delete(
            f"{server_url}/v1/conversations/{conversation.id}",
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

        # Verify it's no longer accessible
        response = requests.get(
            f"{server_url}/v1/conversations/{conversation.id}",
        )
        assert response.status_code == 404, f"Expected 404 for deleted conversation, got {response.status_code}"

    def test_delete_conversation_removes_from_list(self, client: Letta, agent, server_url: str):
        """Test that deleted conversations don't appear in list."""
        # Create two conversations
        conv1 = client.conversations.create(agent_id=agent.id)
        conv2 = client.conversations.create(agent_id=agent.id)

        # Verify both appear in list
        conversations = client.conversations.list(agent_id=agent.id)
        conv_ids = [c.id for c in conversations]
        assert conv1.id in conv_ids
        assert conv2.id in conv_ids

        # Delete one
        response = requests.delete(
            f"{server_url}/v1/conversations/{conv1.id}",
        )
        assert response.status_code == 200

        # Verify only the non-deleted one appears in list
        conversations = client.conversations.list(agent_id=agent.id)
        conv_ids = [c.id for c in conversations]
        assert conv1.id not in conv_ids, "Deleted conversation should not appear in list"
        assert conv2.id in conv_ids, "Non-deleted conversation should still appear"

    def test_delete_conversation_not_found(self, client: Letta, agent, server_url: str):
        """Test that deleting a non-existent conversation returns 404 or 422."""
        fake_id = "conv-00000000-0000-0000-0000-000000000000"
        response = requests.delete(
            f"{server_url}/v1/conversations/{fake_id}",
        )
        assert response.status_code in (404, 422)

    def test_delete_conversation_double_delete(self, client: Letta, agent, server_url: str):
        """Test that deleting an already-deleted conversation returns 404."""
        # Create and delete a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # First delete should succeed
        response = requests.delete(
            f"{server_url}/v1/conversations/{conversation.id}",
        )
        assert response.status_code == 200

        # Second delete should return 404
        response = requests.delete(
            f"{server_url}/v1/conversations/{conversation.id}",
        )
        assert response.status_code == 404, "Double delete should return 404"

    def test_update_deleted_conversation_fails(self, client: Letta, agent, server_url: str):
        """Test that updating a deleted conversation returns 404."""
        # Create and delete a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        response = requests.delete(
            f"{server_url}/v1/conversations/{conversation.id}",
        )
        assert response.status_code == 200

        # Try to update the deleted conversation
        response = requests.patch(
            f"{server_url}/v1/conversations/{conversation.id}",
            json={"summary": "Updated summary"},
        )
        assert response.status_code == 404, "Updating deleted conversation should return 404"


class TestConversationRecompile:
    """Tests for the conversation recompile endpoint."""

    def test_recompile_conversation_updates_existing_conversation_system_message(self, client: Letta, server_url: str):
        unique_marker = f"RECOMPILE_MARKER_{uuid.uuid4().hex[:8]}"

        agent = client.agents.create(
            name=f"test_conv_recompile_{uuid.uuid4().hex[:8]}",
            model=TEST_MODEL_HANDLE,
            # No embedding needed for conversation tests
            memory_blocks=[
                {"label": "human", "value": "The user is a test user."},
                {"label": "persona", "value": "You are a helpful assistant."},
            ],
        )

        try:
            conversation = client.conversations.create(agent_id=agent.id)
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": "Hello, just a quick test."}],
                )
            )

            original_messages = list(client.conversations.messages.list(conversation_id=conversation.id, order="asc"))
            assert original_messages[0].message_type == "system_message"
            assert unique_marker not in original_messages[0].content

            client.agents.blocks.update(
                agent_id=agent.id,
                block_label="human",
                value=f"The user is a test user. {unique_marker}",
            )

            before_recompile_messages = list(client.conversations.messages.list(conversation_id=conversation.id, order="asc"))
            assert unique_marker not in before_recompile_messages[0].content

            response = requests.post(f"{server_url}/v1/conversations/{conversation.id}/recompile")
            assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
            assert unique_marker in response.json()

            after_recompile_messages = list(client.conversations.messages.list(conversation_id=conversation.id, order="asc"))
            assert unique_marker in after_recompile_messages[0].content

        finally:
            client.agents.delete(agent_id=agent.id)

    def test_recompile_conversation_dry_run_does_not_persist(self, client: Letta, server_url: str):
        unique_marker = f"DRY_RUN_MARKER_{uuid.uuid4().hex[:8]}"

        agent = client.agents.create(
            name=f"test_conv_recompile_dry_run_{uuid.uuid4().hex[:8]}",
            model=TEST_MODEL_HANDLE,
            # No embedding needed for conversation tests
            memory_blocks=[
                {"label": "human", "value": "The user is a test user."},
                {"label": "persona", "value": "You are a helpful assistant."},
            ],
        )

        try:
            conversation = client.conversations.create(agent_id=agent.id)
            client.agents.blocks.update(
                agent_id=agent.id,
                block_label="human",
                value=f"The user is a test user. {unique_marker}",
            )

            response = requests.post(f"{server_url}/v1/conversations/{conversation.id}/recompile", params={"dry_run": "true"})
            assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
            assert unique_marker in response.json()

            messages = list(client.conversations.messages.list(conversation_id=conversation.id, order="asc"))
            assert unique_marker not in messages[0].content

        finally:
            client.agents.delete(agent_id=agent.id)

    def test_recompile_conversation_agent_direct_mode(self, client: Letta, server_url: str):
        unique_marker = f"AGENT_DIRECT_MARKER_{uuid.uuid4().hex[:8]}"

        agent = client.agents.create(
            name=f"test_conv_recompile_default_{uuid.uuid4().hex[:8]}",
            model=TEST_MODEL_HANDLE,
            # No embedding needed for conversation tests
            memory_blocks=[
                {"label": "human", "value": "The user is a test user."},
                {"label": "persona", "value": "You are a helpful assistant."},
            ],
        )

        try:
            client.agents.blocks.update(
                agent_id=agent.id,
                block_label="human",
                value=f"The user is a test user. {unique_marker}",
            )

            response = requests.post(
                f"{server_url}/v1/conversations/default/recompile",
                json={"agent_id": agent.id},
            )
            assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
            assert unique_marker in response.json()

        finally:
            client.agents.delete(agent_id=agent.id)


class TestConversationCompact:
    """Tests for the conversation compact (summarization) endpoint."""

    def test_compact_conversation_basic(self, client: Letta, agent, server_url: str):
        """Test basic conversation compaction via the REST endpoint."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send multiple messages to create a history worth summarizing
        for i in range(5):
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": f"Message {i}: Tell me about topic {i}."}],
                )
            )

        # Call compact endpoint via REST
        response = requests.post(
            f"{server_url}/v1/conversations/{conversation.id}/compact",
            json={},
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

        result = response.json()

        # Verify the response structure
        assert "summary" in result
        assert "num_messages_before" in result
        assert "num_messages_after" in result
        assert isinstance(result["summary"], str)
        assert len(result["summary"]) > 0
        assert result["num_messages_before"] > result["num_messages_after"]

    def test_compact_conversation_creates_summary_role_message(self, client: Letta, agent, server_url: str):
        """Test that compaction creates a summary (verified via REST response, not SDK list)."""
        # Create a conversation
        conversation = client.conversations.create(agent_id=agent.id)

        # Send multiple messages to create a history worth summarizing
        for i in range(5):
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": f"Message {i}: Tell me about topic {i}."}],
                )
            )

        # Call compact endpoint with 'all' mode to ensure a single summary
        response = requests.post(
            f"{server_url}/v1/conversations/{conversation.id}/compact",
            json={
                "compaction_settings": {
                    "mode": "all",
                }
            },
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

        # Verify compaction happened via the response
        result = response.json()
        assert "summary" in result
        assert isinstance(result["summary"], str)
        assert len(result["summary"]) > 0
        assert result["num_messages_before"] > result["num_messages_after"]

    def test_compact_conversation_with_settings(self, client: Letta, agent, server_url: str):
        """Test conversation compaction with custom compaction settings."""
        # Create a conversation with multiple messages
        conversation = client.conversations.create(agent_id=agent.id)

        for i in range(5):
            list(
                client.conversations.messages.create(
                    conversation_id=conversation.id,
                    messages=[{"role": "user", "content": f"Remember fact {i}: The number {i} is important."}],
                )
            )

        # Call compact with 'all' mode
        response = requests.post(
            f"{server_url}/v1/conversations/{conversation.id}/compact",
            json={
                "compaction_settings": {
                    "mode": "all",
                }
            },
        )
        assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"

        result = response.json()
        assert result["num_messages_before"] > result["num_messages_after"]

    def test_compact_conversation_preserves_conversation_isolation(self, client: Letta, agent, server_url: str):
        """Test that compacting one conversation doesn't affect another."""
        # Create two conversations
        conv1 = client.conversations.create(agent_id=agent.id)
        conv2 = client.conversations.create(agent_id=agent.id)

        # Add messages to both
        for i in range(5):
            list(
                client.conversations.messages.create(
                    conversation_id=conv1.id,
                    messages=[{"role": "user", "content": f"Conv1 message {i}"}],
                )
            )
            list(
                client.conversations.messages.create(
                    conversation_id=conv2.id,
                    messages=[{"role": "user", "content": f"Conv2 message {i}"}],
                )
            )

        # Get initial in-context message counts from the conversation objects
        conv2_before = client.conversations.retrieve(conversation_id=conv2.id)
        conv2_initial_count = len(conv2_before.in_context_message_ids)

        # Compact only conv1
        response = requests.post(
            f"{server_url}/v1/conversations/{conv1.id}/compact",
            json={},
        )
        assert response.status_code == 200

        # Conv1 should be compacted (verify via response)
        result = response.json()
        assert result["num_messages_before"] > result["num_messages_after"]

        # Conv2 should be unchanged (in-context count should remain the same)
        conv2_after = client.conversations.retrieve(conversation_id=conv2.id)
        assert len(conv2_after.in_context_message_ids) == conv2_initial_count

    def test_compact_conversation_empty_fails(self, client: Letta, agent, server_url: str):
        """Test that compacting an empty conversation fails gracefully."""
        # Create a new conversation without messages
        conversation = client.conversations.create(agent_id=agent.id)

        # Try to compact - should fail since no messages exist
        response = requests.post(
            f"{server_url}/v1/conversations/{conversation.id}/compact",
            json={},
        )

        # Should return 400 because there are no in-context messages
        assert response.status_code == 400

    def test_compact_conversation_invalid_id(self, client: Letta, agent, server_url: str):
        """Test that compacting with invalid conversation ID returns 404 or 422."""
        fake_id = "conv-00000000-0000-0000-0000-000000000000"

        response = requests.post(
            f"{server_url}/v1/conversations/{fake_id}/compact",
            json={},
        )

        assert response.status_code in (404, 422)


class TestConversationSystemMessageRecompilation:
    """Tests that verify the system message is recompiled with latest memory state on new conversation creation."""

    def test_new_conversation_recompiles_system_message_with_updated_memory(self, client: Letta, server_url: str):
        """Test the full workflow:
        1. Agent is created
        2. Send message to agent (through a conversation)
        3. Modify the memory block -> check system message is NOT updated with the modified value
        4. Create a new conversation
        5. Check new conversation system message DOES have the modified value
        """
        unique_marker = f"UNIQUE_MARKER_{uuid.uuid4().hex[:8]}"

        # Step 1: Create an agent with known memory blocks
        agent = client.agents.create(
            name=f"test_sys_msg_recompile_{uuid.uuid4().hex[:8]}",
            model=TEST_MODEL_HANDLE,
            # No embedding needed for conversation tests
            memory_blocks=[
                {"label": "human", "value": "The user is a test user."},
                {"label": "persona", "value": "You are a helpful assistant."},
            ],
        )

        try:
            # Step 2: Create a conversation and send a message to it
            conv1 = client.conversations.create(agent_id=agent.id)

            list(
                client.conversations.messages.create(
                    conversation_id=conv1.id,
                    messages=[{"role": "user", "content": "Hello, just a quick test."}],
                )
            )

            # Verify the conversation has messages including a system message
            conv1_messages = list(
                client.conversations.messages.list(
                    conversation_id=conv1.id,
                    order="asc",
                )
            )
            assert len(conv1_messages) >= 3  # system + user + assistant
            assert conv1_messages[0].message_type == "system_message"

            # Get the original system message content
            original_system_content = conv1_messages[0].content
            assert unique_marker not in original_system_content, "Marker should not be in original system message"

            # Step 3: Modify the memory block with a unique marker
            client.agents.blocks.update(
                agent_id=agent.id,
                block_label="human",
                value=f"The user is a test user. {unique_marker}",
            )

            # Verify the block was actually updated
            updated_block = client.agents.blocks.retrieve(agent_id=agent.id, block_label="human")
            assert unique_marker in updated_block.value

            # Check that the OLD conversation's system message is NOT updated
            conv1_messages_after_update = list(
                client.conversations.messages.list(
                    conversation_id=conv1.id,
                    order="asc",
                )
            )
            old_system_content = conv1_messages_after_update[0].content
            assert unique_marker not in old_system_content, "Old conversation system message should NOT contain the updated memory value"

            # Step 4: Create a new conversation
            conv2 = client.conversations.create(agent_id=agent.id)

            # Step 5: Check the new conversation's system message has the updated value
            # The system message should be compiled at creation time with the latest memory
            conv2_retrieved = client.conversations.retrieve(conversation_id=conv2.id)
            assert len(conv2_retrieved.in_context_message_ids) == 1, (
                f"New conversation should have exactly 1 system message, got {len(conv2_retrieved.in_context_message_ids)}"
            )

            conv2_messages = list(
                client.conversations.messages.list(
                    conversation_id=conv2.id,
                    order="asc",
                )
            )
            assert len(conv2_messages) >= 1
            assert conv2_messages[0].message_type == "system_message"

            new_system_content = conv2_messages[0].content
            assert unique_marker in new_system_content, (
                f"New conversation system message should contain the updated memory value '{unique_marker}', "
                f"but system message content did not include it"
            )

        finally:
            client.agents.delete(agent_id=agent.id)

    def test_conversation_creation_initializes_system_message(self, client: Letta, server_url: str):
        """Test that creating a conversation immediately initializes it with a system message."""
        agent = client.agents.create(
            name=f"test_conv_init_{uuid.uuid4().hex[:8]}",
            model=TEST_MODEL_HANDLE,
            # No embedding needed for conversation tests
            memory_blocks=[
                {"label": "human", "value": "Test user for system message init."},
                {"label": "persona", "value": "You are a helpful assistant."},
            ],
        )

        try:
            # Create a conversation (without sending any messages)
            conversation = client.conversations.create(agent_id=agent.id)

            # Verify the conversation has a system message immediately
            retrieved = client.conversations.retrieve(conversation_id=conversation.id)
            assert len(retrieved.in_context_message_ids) == 1, (
                f"Expected 1 system message after conversation creation, got {len(retrieved.in_context_message_ids)}"
            )

            # Verify the system message content contains memory block values
            messages = list(
                client.conversations.messages.list(
                    conversation_id=conversation.id,
                    order="asc",
                )
            )
            assert len(messages) == 1
            assert messages[0].message_type == "system_message"
            assert "Test user for system message init." in messages[0].content

        finally:
            client.agents.delete(agent_id=agent.id)


@pytest.mark.asyncio(loop_scope="function")
async def test_concurrent_conversation_requests_return_409(
    server_url: str,
    otid_test_agent,
    async_client: AsyncLetta,
    disable_e2b_api_key: Any,
) -> None:
    """
    Test that concurrent requests to the same conversation return 409 CONVERSATION_BUSY.

    This is a simpler test to verify the locking mechanism is working.
    If this fails, the lock is not being held properly.
    """
    from letta.settings import settings

    if settings.redis_host is None or settings.redis_port is None:
        pytest.skip("Redis not configured - skipping conversation lock test")

    conversation = await async_client.conversations.create(agent_id=otid_test_agent.id)

    # Different otids for the two requests
    otid1 = f"test-otid-1-{uuid.uuid4()}"
    otid2 = f"test-otid-2-{uuid.uuid4()}"

    messages1 = [
        MessageCreateParam(
            role="user",
            content="Hello! Please count from 1 to 5 slowly.",
            otid=otid1,
        )
    ]

    messages2 = [
        MessageCreateParam(
            role="user",
            content="Hello! Please count from 1 to 5 slowly.",
            otid=otid2,  # Different otid
        )
    ]

    # Track state
    stream1_started = asyncio.Event()
    stream2_error: Optional[APIError] = None

    async def collect_stream1():
        """First request - the original."""
        stream = await async_client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=messages1,
            background=True,
        )
        async for chunk in stream:
            if not stream1_started.is_set():
                stream1_started.set()
            await asyncio.sleep(0.01)

    async def collect_stream2():
        """Second request - should return 409 (different otid)."""
        await stream1_started.wait()

        nonlocal stream2_error
        try:
            stream = await async_client.conversations.messages.create(
                conversation_id=conversation.id,
                messages=messages2,  # Different otid!
                background=True,
            )
            async for chunk in stream:
                pass
        except APIError as e:
            stream2_error = e

    await asyncio.gather(collect_stream1(), collect_stream2())

    # Stream 2 should have gotten 409 if stream 1 was still running.
    # If stream 1 completed before stream 2 started, stream 2 may succeed —
    # that's correct behavior (lock released, new request is fine).
    if stream2_error is not None:
        assert stream2_error.status_code == 409, f"Stream 2 should have returned 409, got {stream2_error.status_code}"
        logger.info(f"Stream 2 correctly returned 409: {stream2_error.body}")
    else:
        logger.info("Stream 1 completed before stream 2 started — no lock contention (acceptable)")


@pytest.mark.asyncio(loop_scope="function")
async def test_duplicate_request_recovery_with_same_otid(
    server_url: str,
    otid_test_agent,
    async_client: AsyncLetta,
    disable_e2b_api_key: Any,
) -> None:
    """
    Test that a duplicate request with the same otid recovers the existing stream.

    This test verifies the otid-based deduplication and recovery mechanism:
    1. Start a streaming request with a specific otid
    2. Start another request with the SAME otid (while first is still running)
    3. The second request should recover and return the same run_id
    4. After both complete, a third request with the same otid should still recover
    """
    from letta.settings import settings

    if settings.redis_host is None or settings.redis_port is None:
        pytest.skip("Redis not configured - skipping conversation lock test")

    conversation = await async_client.conversations.create(agent_id=otid_test_agent.id)

    shared_otid = f"test-otid-{uuid.uuid4()}"
    messages = [
        MessageCreateParam(
            role="user",
            content="Hello! Please count from 1 to 5 slowly.",
            otid=shared_otid,
        )
    ]

    # Track state
    chunks_from_stream1: List[Any] = []
    chunks_from_stream2: List[Any] = []
    stream1_started = asyncio.Event()

    async def collect_stream1():
        """First request - the original."""
        stream = await async_client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=messages,
            background=True,
        )
        async for chunk in stream:
            chunks_from_stream1.append(chunk)
            if not stream1_started.is_set():
                stream1_started.set()
            await asyncio.sleep(0.01)

    async def collect_stream2():
        """Second request - should recover from stream1 (same otid)."""
        await stream1_started.wait()
        await asyncio.sleep(0.1)

        stream = await async_client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=messages,  # Same otid!
            background=True,
        )
        async for chunk in stream:
            chunks_from_stream2.append(chunk)

    # Run both streams concurrently
    await asyncio.gather(collect_stream1(), collect_stream2())

    def extract_run_id(chunks: List[Any]) -> Optional[str]:
        for chunk in chunks:
            if hasattr(chunk, "run_id") and chunk.run_id:
                return chunk.run_id
        return None

    run_id_1 = extract_run_id(chunks_from_stream1)
    run_id_2 = extract_run_id(chunks_from_stream2)

    logger.info(f"Stream 1 run_id: {run_id_1}")
    logger.info(f"Stream 2 run_id: {run_id_2}")

    assert len(chunks_from_stream1) > 0, "Stream 1 should have received chunks"
    assert len(chunks_from_stream2) > 0, "Stream 2 should have recovered and received chunks"

    assert run_id_1 is not None, "Stream 1 should have a run_id"
    assert run_id_2 is not None, "Stream 2 should have a run_id"
    assert run_id_1 == run_id_2, f"Both streams should have the same run_id (recovery): {run_id_1} vs {run_id_2}"

    # Stream 3: retry AFTER both streams have completed (lock is definitely released).
    # Should still recover the same run from Redis without acquiring the lock.
    chunks_from_stream3: List[Any] = []
    stream3 = await async_client.conversations.messages.create(
        conversation_id=conversation.id,
        messages=messages,  # Same otid!
        background=True,
    )
    async for chunk in stream3:
        chunks_from_stream3.append(chunk)

    run_id_3 = extract_run_id(chunks_from_stream3)
    logger.info(f"Stream 3 run_id: {run_id_3}")

    assert len(chunks_from_stream3) > 0, "Stream 3 should have received chunks"
    assert run_id_3 is not None, "Stream 3 should have a run_id"
    assert run_id_3 == run_id_1, f"Stream 3 should recover the same run_id after lock released: {run_id_3} vs {run_id_1}"


@pytest.mark.asyncio
async def test_otid_recovery_via_retrieve_stream(
    server_url: str,
    otid_test_agent,
    async_client: AsyncLetta,
    disable_e2b_api_key: Any,
) -> None:
    """
    Test that the retrieve_conversation_stream endpoint can recover a run via OTID.

    This simulates the client flow:
    1. Send a background streaming request with an OTID
    2. While it's running, call retrieve_conversation_stream with the same OTID
    3. The stream endpoint should find the run via OTID lookup and return chunks
    """
    from letta.settings import settings

    if settings.redis_host is None or settings.redis_port is None:
        pytest.skip("Redis not configured - skipping OTID stream recovery test")

    conversation = await async_client.conversations.create(agent_id=otid_test_agent.id)

    shared_otid = f"test-otid-{uuid.uuid4()}"
    messages = [
        MessageCreateParam(
            role="user",
            content="Hello! Please respond briefly.",
            otid=shared_otid,
        )
    ]

    # Start a background streaming request
    chunks_from_send: List[Any] = []
    send_started = asyncio.Event()

    async def send_message():
        stream = await async_client.conversations.messages.create(
            conversation_id=conversation.id,
            messages=messages,
            background=True,
        )
        async for chunk in stream:
            chunks_from_send.append(chunk)
            if not send_started.is_set():
                send_started.set()
            await asyncio.sleep(0.01)

    # Start the send, wait for it to begin streaming, then try OTID recovery
    send_task = asyncio.create_task(send_message())
    await send_started.wait()
    await asyncio.sleep(0.5)  # Give time for OTID mapping to be stored

    # Recover stream via OTID using the retrieve endpoint
    recover_response = requests.post(
        f"{server_url}/v1/conversations/{conversation.id}/stream",
        json={"otid": shared_otid},
    )
    assert recover_response.status_code == 200, (
        f"OTID recovery via retrieve_stream should succeed: {recover_response.status_code} {recover_response.text}"
    )

    # Wait for original send to finish
    await send_task

    assert len(chunks_from_send) > 0, "Original send should have received chunks"