feat: add sessions endpoint with session management enhancements

pyproject.toml

@@ -232,7 +232,7 @@ ignore-regex = '.*(Stati Uniti|Tense=Pres).*'
 timeout = 120
 timeout_method = "signal"
 minversion = "6.0"
-testpaths = ["tests", "integration"]
+testpaths = ["src/backend/tests"]
 console_output_style = "progress"
 filterwarnings = ["ignore::DeprecationWarning", "ignore::ResourceWarning"]
 log_cli = true

src/backend/base/langflow/api/v1/monitor.py

@@ -40,6 +40,24 @@ async def delete_vertex_builds(flow_id: Annotated[UUID, Query()], session: DbSes
         raise HTTPException(status_code=500, detail=str(e)) from e
 
 
+@router.get("/messages/sessions", dependencies=[Depends(get_current_active_user)])
+async def get_message_sessions(
+    session: DbSession,
+    flow_id: Annotated[UUID | None, Query()] = None,
+) -> list[str]:
+    try:
+        stmt = select(MessageTable.session_id).distinct()
+        stmt = stmt.where(col(MessageTable.session_id).isnot(None))
+
+        if flow_id:
+            stmt = stmt.where(MessageTable.flow_id == flow_id)
+
+        session_ids = await session.exec(stmt)
+        return list(session_ids)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+
 @router.get("/messages")
 async def get_messages(
     session: DbSession,
@@ -54,7 +72,10 @@ async def get_messages(
         if flow_id:
             stmt = stmt.where(MessageTable.flow_id == flow_id)
         if session_id:
-            stmt = stmt.where(MessageTable.session_id == session_id)
+            from urllib.parse import unquote
+
+            decoded_session_id = unquote(session_id)
+            stmt = stmt.where(MessageTable.session_id == decoded_session_id)
         if sender:
             stmt = stmt.where(MessageTable.sender == sender)
         if sender_name:

src/backend/tests/unit/test_messages_endpoints.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timezone
+from urllib.parse import quote
 from uuid import UUID
 
 import pytest
@@ -32,6 +33,20 @@ async def created_messages(session):  # noqa: ARG001
         return await aadd_messagetables(messagetables, _session)
 
 
+@pytest.fixture
+async def messages_with_datetime_session_id(session):  # noqa: ARG001
+    """Create messages with datetime-like session IDs that contain characters requiring URL encoding."""
+    datetime_session_id = "2024-01-15 10:30:45 UTC"  # Contains spaces and colons
+    async with session_scope() as _session:
+        messages = [
+            MessageCreate(text="Datetime message 1", sender="User", sender_name="User", session_id=datetime_session_id),
+            MessageCreate(text="Datetime message 2", sender="AI", sender_name="AI", session_id=datetime_session_id),
+        ]
+        messagetables = [MessageTable.model_validate(message, from_attributes=True) for message in messages]
+        created_messages = await aadd_messagetables(messagetables, _session)
+        return created_messages, datetime_session_id
+
+
 @pytest.mark.api_key_required
 async def test_delete_messages(client: AsyncClient, created_messages, logged_in_headers):
     response = await client.request(
@@ -127,3 +142,96 @@ async def test_no_messages_found_with_given_session_id(client, logged_in_headers
 
     assert response.status_code == 404, response.text
     assert response.json()["detail"] == "Not Found"
+
+
+# Test for URL-encoded datetime session ID
+@pytest.mark.api_key_required
+async def test_get_messages_with_url_encoded_datetime_session_id(
+    client: AsyncClient, messages_with_datetime_session_id, logged_in_headers
+):
+    """Test that URL-encoded datetime session IDs are properly decoded and matched."""
+    created_messages, datetime_session_id = messages_with_datetime_session_id
+
+    # URL encode the datetime session ID (spaces become %20, colons become %3A)
+    encoded_session_id = quote(datetime_session_id)
+
+    # Test with URL-encoded session ID
+    response = await client.get(
+        "api/v1/monitor/messages", params={"session_id": encoded_session_id}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    messages = response.json()
+    assert len(messages) == 2
+
+    # Verify all messages have the correct (decoded) session ID
+    for message in messages:
+        assert message["session_id"] == datetime_session_id
+
+    # Verify message content
+    assert messages[0]["text"] == "Datetime message 1"
+    assert messages[1]["text"] == "Datetime message 2"
+
+
+@pytest.mark.api_key_required
+async def test_get_messages_with_non_encoded_datetime_session_id(
+    client: AsyncClient, messages_with_datetime_session_id, logged_in_headers
+):
+    """Test that non-URL-encoded datetime session IDs also work correctly."""
+    created_messages, datetime_session_id = messages_with_datetime_session_id
+
+    # Test with non-encoded session ID (should still work due to unquote being safe for non-encoded strings)
+    response = await client.get(
+        "api/v1/monitor/messages", params={"session_id": datetime_session_id}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    messages = response.json()
+    assert len(messages) == 2
+
+    # Verify all messages have the correct session ID
+    for message in messages:
+        assert message["session_id"] == datetime_session_id
+
+
+@pytest.mark.api_key_required
+async def test_get_messages_with_various_encoded_characters(client: AsyncClient, logged_in_headers):
+    """Test various URL-encoded characters in session IDs."""
+    # Create a session ID with various special characters
+    special_session_id = "test+session:2024@domain.com"
+
+    async with session_scope() as session:
+        message = MessageCreate(
+            text="Special chars message", sender="User", sender_name="User", session_id=special_session_id
+        )
+        messagetable = MessageTable.model_validate(message, from_attributes=True)
+        await aadd_messagetables([messagetable], session)
+
+    # URL encode the session ID
+    encoded_session_id = quote(special_session_id)
+
+    # Test with URL-encoded session ID
+    response = await client.get(
+        "api/v1/monitor/messages", params={"session_id": encoded_session_id}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    messages = response.json()
+    assert len(messages) == 1
+    assert messages[0]["session_id"] == special_session_id
+    assert messages[0]["text"] == "Special chars message"
+
+
+@pytest.mark.api_key_required
+async def test_get_messages_empty_result_with_encoded_nonexistent_session(client: AsyncClient, logged_in_headers):
+    """Test that URL-encoded non-existent session IDs return empty results."""
+    nonexistent_session_id = "2024-12-31 23:59:59 UTC"
+    encoded_session_id = quote(nonexistent_session_id)
+
+    response = await client.get(
+        "api/v1/monitor/messages", params={"session_id": encoded_session_id}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    messages = response.json()
+    assert len(messages) == 0

src/backend/tests/unit/test_session_endpoint.py

@@ -0,0 +1,142 @@
+from uuid import uuid4
+
+import pytest
+from httpx import AsyncClient
+from langflow.memory import aadd_messagetables
+from langflow.services.database.models.message.model import MessageTable
+from langflow.services.deps import session_scope
+
+
+@pytest.fixture
+async def messages_with_flow_ids(session):  # noqa: ARG001
+    """Create messages with different session_ids and flow_ids for testing sessions endpoint."""
+    async with session_scope() as _session:
+        flow_id_1 = uuid4()
+        flow_id_2 = uuid4()
+
+        # Create MessageTable objects directly since MessageCreate doesn't have flow_id field
+        messagetables = [
+            MessageTable(
+                text="Message 1", sender="User", sender_name="User", session_id="session_A", flow_id=flow_id_1
+            ),
+            MessageTable(text="Message 2", sender="AI", sender_name="AI", session_id="session_A", flow_id=flow_id_1),
+            MessageTable(
+                text="Message 3", sender="User", sender_name="User", session_id="session_B", flow_id=flow_id_1
+            ),
+            MessageTable(
+                text="Message 4", sender="User", sender_name="User", session_id="session_C", flow_id=flow_id_2
+            ),
+            MessageTable(text="Message 5", sender="AI", sender_name="AI", session_id="session_D", flow_id=flow_id_2),
+            MessageTable(
+                text="Message 6",
+                sender="User",
+                sender_name="User",
+                session_id="session_E",
+                flow_id=None,  # No flow_id
+            ),
+        ]
+        created_messages = await aadd_messagetables(messagetables, _session)
+
+        return {
+            "messages": created_messages,
+            "flow_id_1": flow_id_1,
+            "flow_id_2": flow_id_2,
+            "expected_sessions_flow_1": {"session_A", "session_B"},
+            "expected_sessions_flow_2": {"session_C", "session_D"},
+            "expected_all_sessions": {"session_A", "session_B", "session_C", "session_D", "session_E"},
+        }
+
+
+# Tests for /sessions endpoint
+@pytest.mark.api_key_required
+async def test_get_sessions_all(client: AsyncClient, logged_in_headers, messages_with_flow_ids):
+    """Test getting all sessions without any filter."""
+    response = await client.get("api/v1/monitor/messages/sessions", headers=logged_in_headers)
+
+    assert response.status_code == 200, response.text
+    sessions = response.json()
+    assert isinstance(sessions, list)
+
+    # Convert to set for easier comparison since order doesn't matter
+    returned_sessions = set(sessions)
+    expected_sessions = messages_with_flow_ids["expected_all_sessions"]
+
+    assert returned_sessions == expected_sessions
+    assert len(sessions) == len(expected_sessions)
+
+
+@pytest.mark.api_key_required
+async def test_get_sessions_with_flow_id_filter(client: AsyncClient, logged_in_headers, messages_with_flow_ids):
+    """Test getting sessions filtered by flow_id."""
+    flow_id_1 = messages_with_flow_ids["flow_id_1"]
+
+    response = await client.get(
+        "api/v1/monitor/messages/sessions", params={"flow_id": str(flow_id_1)}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    sessions = response.json()
+    assert isinstance(sessions, list)
+
+    returned_sessions = set(sessions)
+    expected_sessions = messages_with_flow_ids["expected_sessions_flow_1"]
+
+    assert returned_sessions == expected_sessions
+    assert len(sessions) == len(expected_sessions)
+
+
+@pytest.mark.api_key_required
+async def test_get_sessions_with_different_flow_id(client: AsyncClient, logged_in_headers, messages_with_flow_ids):
+    """Test getting sessions filtered by a different flow_id."""
+    flow_id_2 = messages_with_flow_ids["flow_id_2"]
+
+    response = await client.get(
+        "api/v1/monitor/messages/sessions", params={"flow_id": str(flow_id_2)}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    sessions = response.json()
+    assert isinstance(sessions, list)
+
+    returned_sessions = set(sessions)
+    expected_sessions = messages_with_flow_ids["expected_sessions_flow_2"]
+
+    assert returned_sessions == expected_sessions
+    assert len(sessions) == len(expected_sessions)
+
+
+@pytest.mark.api_key_required
+async def test_get_sessions_with_non_existent_flow_id(client: AsyncClient, logged_in_headers):
+    """Test getting sessions with a non-existent flow_id returns empty list."""
+    non_existent_flow_id = uuid4()
+
+    response = await client.get(
+        "api/v1/monitor/messages/sessions", params={"flow_id": str(non_existent_flow_id)}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 200, response.text
+    sessions = response.json()
+    assert isinstance(sessions, list)
+    assert len(sessions) == 0
+
+
+@pytest.mark.api_key_required
+async def test_get_sessions_empty_database(client: AsyncClient, logged_in_headers):
+    """Test getting sessions when no messages exist in database."""
+    response = await client.get("api/v1/monitor/messages/sessions", headers=logged_in_headers)
+
+    assert response.status_code == 200, response.text
+    sessions = response.json()
+    assert isinstance(sessions, list)
+    assert len(sessions) == 0
+
+
+@pytest.mark.api_key_required
+async def test_get_sessions_invalid_flow_id_format(client: AsyncClient, logged_in_headers):
+    """Test getting sessions with invalid flow_id format returns 422."""
+    response = await client.get(
+        "api/v1/monitor/messages/sessions", params={"flow_id": "invalid-uuid"}, headers=logged_in_headers
+    )
+
+    assert response.status_code == 422, response.text
+    assert "detail" in response.json()

src/frontend/src/controllers/API/queries/messages/use-delete-messages.ts

@@ -12,7 +12,7 @@ export const useDeleteMessages: useMutationFunctionType<
   undefined,
   DeleteMessagesParams
 > = (options?) => {
-  const { mutate } = UseRequestProcessor();
+  const { mutate, queryClient } = UseRequestProcessor();
 
   const deleteMessage = async ({ ids }: DeleteMessagesParams): Promise<any> => {
     const response = await api.delete(`${getURL("MESSAGES")}`, {
@@ -26,7 +26,15 @@ export const useDeleteMessages: useMutationFunctionType<
     DeleteMessagesParams,
     any,
     DeleteMessagesParams
-  > = mutate(["useDeleteMessages"], deleteMessage, options);
+  > = mutate(["useDeleteMessages"], deleteMessage, {
+    ...options,
+    onSettled: (data, error, variables, context) => {
+      queryClient.invalidateQueries({
+        queryKey: ["useGetSessionsFromFlowQuery"],
+      });
+      options?.onSettled?.(data, error, variables, context);
+    },
+  });
 
   return mutation;
 };

src/frontend/src/controllers/API/queries/messages/use-delete-sessions.ts

@@ -0,0 +1,41 @@
+import { useMutationFunctionType } from "@/types/api";
+import { UseMutationResult } from "@tanstack/react-query";
+import { api } from "../../api";
+import { getURL } from "../../helpers/constants";
+import { UseRequestProcessor } from "../../services/request-processor";
+
+interface DeleteSessionParams {
+  sessionId: string;
+}
+
+export const useDeleteSession: useMutationFunctionType<
+  undefined,
+  DeleteSessionParams
+> = (options?) => {
+  const { mutate, queryClient } = UseRequestProcessor();
+
+  const deleteSession = async ({
+    sessionId,
+  }: DeleteSessionParams): Promise<any> => {
+    const response = await api.delete(
+      `${getURL("MESSAGES")}/session/${sessionId}`,
+    );
+    return response.data;
+  };
+
+  const mutation: UseMutationResult<
+    DeleteSessionParams,
+    any,
+    DeleteSessionParams
+  > = mutate(["useDeleteSession"], deleteSession, {
+    ...options,
+    onSettled: (data, error, variables, context) => {
+      queryClient.invalidateQueries({
+        queryKey: ["useGetSessionsFromFlowQuery"],
+      });
+      options?.onSettled?.(data, error, variables, context);
+    },
+  });
+
+  return mutation;
+};