feat(agent): keep early chat context with a rolling summary of older messages

src/backend/base/langflow/alembic/versions/f1b3d5a7c9e2_add_session_memory_state.py

@@ -0,0 +1,44 @@
+"""Add session_memory_state for rolling chat-history summaries.
+
+Revision ID: f1b3d5a7c9e2
+Revises: e5a7c9b1d3f2
+Phase: EXPAND
+Safe to rollback: YES (the summaries are a derived cache of the messages)
+Services compatible: Older versions ignore this additional table.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "f1b3d5a7c9e2"  # pragma: allowlist secret - Alembic revision
+down_revision = "e5a7c9b1d3f2"  # pragma: allowlist secret - Alembic revision
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if sa.inspect(op.get_bind()).has_table("session_memory_state"):
+        return
+    op.create_table(
+        "session_memory_state",
+        sa.Column("id", sa.Uuid(), primary_key=True),
+        sa.Column("flow_id", sa.Uuid(), nullable=False),
+        sa.Column("user_id", sa.Uuid(), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("context_id", sa.String(), nullable=False, server_default=""),
+        sa.Column("summary", sa.Text(), nullable=False, server_default=""),
+        sa.Column("cursor_message_id", sa.Uuid(), nullable=True),
+        sa.Column("cursor_timestamp", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("folded_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("summary_model", sa.String(), nullable=True),
+        sa.Column("version", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.UniqueConstraint("flow_id", "user_id", "session_id", "context_id", name="uq_session_memory_state_scope"),
+    )
+    op.create_index("ix_session_memory_state_flow_session", "session_memory_state", ["flow_id", "session_id"])
+
+
+def downgrade() -> None:
+    if sa.inspect(op.get_bind()).has_table("session_memory_state"):
+        op.drop_table("session_memory_state")

src/backend/base/langflow/api/utils/flow_utils.py

@@ -27,6 +27,7 @@
 from langflow.services.database.models.flow.model import Flow
 from langflow.services.database.models.flow_version.model import FlowVersion
 from langflow.services.database.models.message.model import MessageTable
+from langflow.services.database.models.session_memory.model import SessionMemoryStateTable
 from langflow.services.database.models.traces.model import SpanTable, TraceTable
 from langflow.services.database.models.transactions.model import TransactionTable
 from langflow.services.database.models.trigger.model import Trigger
@@ -155,6 +156,7 @@ async def cascade_delete_flow(
         # it might cause unexpected behaviors because the session id could still be
         # used elsewhere to search for these messages.
         await session.exec(delete(MessageTable).where(MessageTable.flow_id == flow_id))
+        await session.exec(delete(SessionMemoryStateTable).where(SessionMemoryStateTable.flow_id == flow_id))
         await session.exec(delete(TransactionTable).where(TransactionTable.flow_id == flow_id))
         await session.exec(delete(VertexBuildTable).where(VertexBuildTable.flow_id == flow_id))
         # Explicit delete despite FK CASCADE -- SQLite doesn't enforce FK cascades

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

@@ -11,6 +11,7 @@
 
 from langflow.api.utils import DbSession, custom_params
 from langflow.api.utils.flow_utils import compute_virtual_flow_id
+from langflow.memory import adelete_session_memory_states
 from langflow.schema.message import MessageResponse
 from langflow.services.auth.utils import get_current_active_superuser, get_current_active_user
 from langflow.services.authorization import FlowAction, ensure_flow_permission
@@ -228,6 +229,12 @@ async def _purge_memory_base_session_data(user_id: UUID, session_ids: list[str],
         await get_memory_base_service().purge_session_data(user_id=user_id, session_ids=session_ids, db=db)
 
 
+async def _delete_session_summaries(db, user_id: UUID, session_ids: list[str]) -> None:
+    """Drop rolling chat summaries for these sessions in flows owned by the user, in the caller's transaction."""
+    owned_flow_ids = list((await db.exec(select(Flow.id).where(Flow.user_id == user_id))).all())
+    await adelete_session_memory_states(flow_ids=owned_flow_ids, session_ids=session_ids, db=db)
+
+
 @router.get("/builds", dependencies=[Depends(get_current_active_user)])
 async def get_vertex_builds(
     flow_id: Annotated[UUID, Query()],
@@ -554,6 +561,7 @@ async def delete_messages_session(
         async with session_storage_operation(user_id=current_user.id, session_ids=owned_sessions):
             await _purge_memory_base_session_data(current_user.id, owned_sessions, db=session)
             await delete_messages_for_user_by_session(session, current_user.id, session_id)
+            await _delete_session_summaries(session, current_user.id, owned_sessions)
             await session.commit()
     except StorageUnavailableError as e:
         await session.rollback()
@@ -638,6 +646,7 @@ async def delete_messages_sessions(
                 .where(col(MessageTable.id).in_(message_ids))
                 .execution_options(synchronize_session="fetch")
             )
+            await _delete_session_summaries(session, current_user.id, affected_session_ids)
             await session.commit()
     except StorageUnavailableError as e:
         await session.rollback()
@@ -743,6 +752,7 @@ async def delete_shared_messages_session(
             .where(MessageTable.session_id == session_id)
         )
         await session.exec(stmt)
+        await adelete_session_memory_states(flow_ids=[virtual_flow_id], session_ids=[session_id], db=session)
     except Exception as e:
         await session.rollback()
         raise HTTPException(status_code=500, detail=str(e)) from e

src/backend/base/langflow/locales/en.json

@@ -73,6 +73,8 @@
   "components.agent.inputs.project_id.info.0d91e5f5": "The project ID associated with the foundation model (IBM watsonx.ai only)",
   "components.agent.inputs.stream.display_name.1eec3071": "Stream",
   "components.agent.inputs.stream.info.f835b3ff": "Stream the response from the model. Streaming works only in Chat.",
+  "components.agent.inputs.summarize_older_messages.display_name.d8deb3df": "Summarize Older Messages",
+  "components.agent.inputs.summarize_older_messages.info.9ec6b4b1": "Keep a running summary of messages older than the chat history window, so early context survives when the window is small.",
   "components.agent.inputs.system_prompt.display_name.55ce85ab": "Agent Instructions",
   "components.agent.inputs.system_prompt.info.a1fa7781": "System Prompt: Initial instructions and context provided to guide the agent's behavior. Supports dynamic placeholders: {current_date}, {model_name}, {optional_user_context}.",
   "components.agent.inputs.tools.display_name.ea93d6a2": "Tools",
@@ -134,6 +136,8 @@
   "components.altkagent.inputs.response_processing_size_threshold.info.d6aeb95b": "Tool output is post-processed only if response exceeds this character threshold.",
   "components.altkagent.inputs.stream.display_name.1eec3071": "Stream",
   "components.altkagent.inputs.stream.info.f835b3ff": "Stream the response from the model. Streaming works only in Chat.",
+  "components.altkagent.inputs.summarize_older_messages.display_name.d8deb3df": "Summarize Older Messages",
+  "components.altkagent.inputs.summarize_older_messages.info.9ec6b4b1": "Keep a running summary of messages older than the chat history window, so early context survives when the window is small.",
   "components.altkagent.inputs.system_prompt.display_name.55ce85ab": "Agent Instructions",
   "components.altkagent.inputs.system_prompt.info.a1fa7781": "System Prompt: Initial instructions and context provided to guide the agent's behavior. Supports dynamic placeholders: {current_date}, {model_name}, {optional_user_context}.",
   "components.altkagent.inputs.tools.display_name.ea93d6a2": "Tools",

src/backend/base/langflow/memory.py

@@ -2,13 +2,16 @@
 import json
 from collections.abc import Sequence
 from contextlib import suppress
+from datetime import datetime, timezone
 from uuid import UUID
 
 from langchain_core.chat_history import BaseChatMessageHistory
 from langchain_core.messages import BaseMessage
 from lfx.log.logger import logger
+from lfx.memory.session_state import SessionMemoryState
 from lfx.utils.async_helpers import run_until_complete
-from sqlalchemy import delete
+from sqlalchemy import and_, delete, or_, update
+from sqlalchemy.exc import IntegrityError
 from sqlmodel import col, select
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -18,6 +21,7 @@
     MessageRead,
     MessageTable,
 )
+from langflow.services.database.models.session_memory.model import SessionMemoryStateTable
 from langflow.services.deps import session_scope
 
 
@@ -127,6 +131,10 @@ def _get_variable_query(
             raise ValueError(msg)
         col = getattr(MessageTable, order_by).desc() if order == "DESC" else getattr(MessageTable, order_by).asc()
         stmt = stmt.order_by(col)
+        if order_by == "timestamp":
+            # Only makes the order of exact-timestamp ties deterministic (it was undefined before).
+            id_col = MessageTable.id.desc() if order == "DESC" else MessageTable.id.asc()
+            stmt = stmt.order_by(id_col)
     if limit:
         stmt = stmt.limit(limit)
     return stmt
@@ -617,3 +625,228 @@ def clear(self) -> None:
     async def aclear(self) -> None:
         flow_id, user_id = self._require_scope()
         await adelete_messages(self.session_id, self.context_id, flow_id=flow_id, user_id=user_id)
+
+
+def _session_memory_scope_filters(flow_id: UUID, user_id: UUID, session_id: str, context_id: str | None):
+    """Predicates identifying one session-memory row; ``None`` context is stored as ``""``."""
+    return (
+        SessionMemoryStateTable.flow_id == flow_id,
+        SessionMemoryStateTable.user_id == user_id,
+        SessionMemoryStateTable.session_id == session_id,
+        SessionMemoryStateTable.context_id == (context_id or ""),
+    )
+
+
+def _to_session_memory_state(row: SessionMemoryStateTable) -> SessionMemoryState:
+    return SessionMemoryState(
+        summary=row.summary,
+        cursor_message_id=row.cursor_message_id,
+        cursor_timestamp=row.cursor_timestamp,
+        folded_count=row.folded_count,
+        version=row.version,
+    )
+
+
+async def aget_session_memory_state(
+    *,
+    flow_id: UUID | str | None,
+    user_id: UUID | str | None,
+    session_id: str,
+    context_id: str | None,
+) -> SessionMemoryState:
+    """Return the stored summary state for a session, or an empty (version 0) state when none exists."""
+    flow_id, user_id = _message_scope(flow_id, user_id)
+    if flow_id is None or user_id is None:
+        return SessionMemoryState()
+    async with session_scope() as session:
+        row = (
+            await session.exec(
+                select(SessionMemoryStateTable).where(
+                    *_session_memory_scope_filters(flow_id, user_id, session_id, context_id)
+                )
+            )
+        ).first()
+        return _to_session_memory_state(row) if row else SessionMemoryState()
+
+
+async def asave_session_memory_state(
+    *,
+    flow_id: UUID | str | None,
+    user_id: UUID | str | None,
+    session_id: str,
+    context_id: str | None,
+    summary: str,
+    cursor_message_id: UUID | str,
+    folded_delta: int,
+    summary_model: str | None,
+    expected_version: int,
+) -> SessionMemoryState | None:
+    """Compare-and-swap save of the summary state; returns None when the save lost a race or the cursor is gone.
+
+    ``expected_version == 0`` inserts the first row; otherwise a single conditional UPDATE only
+    succeeds while the stored version is still ``expected_version``. The cursor timestamp is read from
+    the message row so it keeps the database's full precision.
+    """
+    flow_id, user_id = _message_scope(flow_id, user_id)
+    if flow_id is None or user_id is None:
+        return None
+    cursor_id = UUID(str(cursor_message_id))
+    scope = _session_memory_scope_filters(flow_id, user_id, session_id, context_id)
+    now = datetime.now(timezone.utc)
+    async with session_scope() as session:
+        cursor_timestamp = (
+            await session.exec(
+                select(MessageTable.timestamp).where(
+                    MessageTable.id == cursor_id,
+                    MessageTable.flow_id == flow_id,
+                    MessageTable.user_id == user_id,
+                )
+            )
+        ).first()
+        if cursor_timestamp is None:
+            return None
+        if expected_version == 0:
+            row = SessionMemoryStateTable(
+                flow_id=flow_id,
+                user_id=user_id,
+                session_id=session_id,
+                context_id=context_id or "",
+                summary=summary,
+                cursor_message_id=cursor_id,
+                cursor_timestamp=cursor_timestamp,
+                folded_count=folded_delta,
+                summary_model=summary_model,
+                version=1,
+                created_at=now,
+                updated_at=now,
+            )
+            session.add(row)
+            try:
+                await session.flush()
+            except IntegrityError:
+                # Another run created the row first; leave the session usable and report the lost race.
+                await session.rollback()
+                return None
+            return _to_session_memory_state(row)
+        result = await session.exec(
+            update(SessionMemoryStateTable)
+            .where(*scope, SessionMemoryStateTable.version == expected_version)
+            .values(
+                version=expected_version + 1,
+                summary=summary,
+                cursor_message_id=cursor_id,
+                cursor_timestamp=cursor_timestamp,
+                folded_count=SessionMemoryStateTable.folded_count + folded_delta,
+                summary_model=summary_model,
+                updated_at=now,
+            )
+            .execution_options(synchronize_session=False)
+        )
+        if result.rowcount == 0:
+            return None
+        row = (await session.exec(select(SessionMemoryStateTable).where(*scope))).one()
+        return _to_session_memory_state(row)
+
+
+async def aget_messages_between(
+    *,
+    flow_id: UUID | str | None,
+    user_id: UUID | str | None,
+    session_id: str,
+    context_id: str | None,
+    after_message_id: UUID | str | None,
+    after_timestamp: datetime | None,
+    before_message_id: UUID | str,
+    limit: int,
+) -> list[Message]:
+    """Return non-error messages strictly between two cursors, oldest first.
+
+    Bounds are compared as ``(timestamp, id)`` tuples using the database's own values (a message's
+    string timestamp may have lost sub-second precision). When the lower-bound message no longer
+    exists, ``after_timestamp`` is used instead; with neither the range is open below.
+    """
+    flow_id, user_id = _message_scope(flow_id, user_id)
+    if flow_id is None or user_id is None:
+        return []
+    async with session_scope() as session:
+        boundary = (
+            await session.exec(
+                select(MessageTable.timestamp, MessageTable.id).where(MessageTable.id == UUID(str(before_message_id)))
+            )
+        ).first()
+        if boundary is None:
+            return []
+        before_ts, before_id = boundary
+        stmt = (
+            select(MessageTable)
+            .where(
+                MessageTable.flow_id == flow_id,
+                MessageTable.user_id == user_id,
+                MessageTable.session_id == session_id,
+                MessageTable.error == False,  # noqa: E712
+                MessageTable.category != "error",
+                or_(
+                    col(MessageTable.timestamp) < before_ts,
+                    and_(col(MessageTable.timestamp) == before_ts, col(MessageTable.id) < before_id),
+                ),
+            )
+            .order_by(col(MessageTable.timestamp).asc(), col(MessageTable.id).asc())
+            .limit(limit)
+        )
+        if context_id:
+            stmt = stmt.where(MessageTable.context_id == context_id)
+        after = None
+        if after_message_id is not None:
+            after = (
+                await session.exec(
+                    select(MessageTable.timestamp, MessageTable.id).where(
+                        MessageTable.id == UUID(str(after_message_id))
+                    )
+                )
+            ).first()
+        if after is not None:
+            after_ts, after_id = after
+            stmt = stmt.where(
+                or_(
+                    col(MessageTable.timestamp) > after_ts,
+                    and_(col(MessageTable.timestamp) == after_ts, col(MessageTable.id) > after_id),
+                )
+            )
+        elif after_timestamp is not None:
+            if after_message_id is not None:
+                # Cursor row is gone: keep its (timestamp, id) position so tied rows are not dropped.
+                stmt = stmt.where(
+                    or_(
+                        col(MessageTable.timestamp) > after_timestamp,
+                        and_(
+                            col(MessageTable.timestamp) == after_timestamp,
+                            col(MessageTable.id) > UUID(str(after_message_id)),
+                        ),
+                    )
+                )
+            else:
+                stmt = stmt.where(col(MessageTable.timestamp) > after_timestamp)
+        messages = await session.exec(stmt)
+        return [await Message.create(**d.model_dump()) for d in messages]
+
+
+async def adelete_session_memory_states(
+    *,
+    flow_ids: list[UUID],
+    session_ids: list[str] | None = None,
+    db: AsyncSession | None = None,
+) -> int:
+    """Delete summary state for flows (optionally only some sessions); returns the rows removed.
+
+    Not scoped to a graph run: callers (session/flow delete endpoints) have already checked ownership.
+    Pass ``db`` to delete inside the caller's transaction; otherwise a session of its own is used.
+    """
+    if not flow_ids:
+        return 0
+    stmt = delete(SessionMemoryStateTable).where(col(SessionMemoryStateTable.flow_id).in_(flow_ids))
+    if session_ids is not None:
+        stmt = stmt.where(col(SessionMemoryStateTable.session_id).in_(session_ids))
+    if db is not None:
+        return (await db.exec(stmt)).rowcount
+    async with session_scope() as session:
+        return (await session.exec(stmt)).rowcount