From d6e8c2143adf3c7f1b50b485dae9e2b83ea4bf26 Mon Sep 17 00:00:00 2001 From: Rajat Ahuja Date: Wed, 7 Jan 2026 14:28:21 -0500 Subject: [PATCH] fix: summary creation if start_message_id > end_message_id (#312) * fix: summary creation if start_message_id > end_message_id * fix: change start_id calculation * fix: use resolved configuration * fix: use message_seq_in_session * fix: clean up method * fix: skip if latest summary on session covers through message_id --- src/crud/__init__.py | 2 ++ src/crud/message.py | 50 +++++++++++++++++++++++++++++++++ src/utils/summarizer.py | 61 ++++++++++++++++++++++++++++++----------- 3 files changed, 97 insertions(+), 16 deletions(-) diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 489c0b14..9345905a 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -14,6 +14,7 @@ from .message import ( get_message, get_message_seq_in_session, get_messages, + get_messages_by_seq_range, get_messages_id_range, update_message, ) @@ -72,6 +73,7 @@ __all__ = [ # Message "create_messages", "get_messages", + "get_messages_by_seq_range", "get_messages_id_range", "get_message", "get_message_seq_in_session", diff --git a/src/crud/message.py b/src/crud/message.py index c6d3476d..ed2ccf93 100644 --- a/src/crud/message.py +++ b/src/crud/message.py @@ -282,6 +282,56 @@ async def get_messages_id_range( return list(result.scalars().all()) +async def get_messages_by_seq_range( + db: AsyncSession, + workspace_name: str, + session_name: str, + start_seq: int = 1, + end_seq: int | None = None, +) -> list[models.Message]: + """ + Get messages from a session by seq_in_session range. + + This is useful for getting the last N messages in a session. + + Args: + db: Database session + workspace_name: Name of the workspace + session_name: Name of the session + start_seq: Sequence number of the first message to return (inclusive) + end_seq: Sequence number of the last message to return (inclusive) + + Returns: + List of messages ordered by seq_in_session + """ + if start_seq < 1 or (end_seq is not None and start_seq > end_seq): + return [] + + base_conditions = [ + models.Message.workspace_name == workspace_name, + models.Message.session_name == session_name, + ] + + if end_seq is not None: + base_conditions.append( + and_( + models.Message.seq_in_session >= start_seq, + models.Message.seq_in_session <= end_seq, + ) + ) + else: + base_conditions.append(models.Message.seq_in_session >= start_seq) + + stmt = ( + select(models.Message) + .where(*base_conditions) + .order_by(models.Message.seq_in_session.asc()) + ) + + result = await db.execute(stmt) + return list(result.scalars().all()) + + async def get_message_seq_in_session( db: AsyncSession, workspace_name: str, diff --git a/src/utils/summarizer.py b/src/utils/summarizer.py index 4a93f591..a1552369 100644 --- a/src/utils/summarizer.py +++ b/src/utils/summarizer.py @@ -15,6 +15,7 @@ from src.config import settings from src.crud.session import session_cache_key from src.dependencies import tracked_db from src.exceptions import ResourceNotFoundException +from src.models import Message from src.utils.clients import HonchoLLMCallResponse, honcho_llm_call from src.utils.formatting import utc_now_iso from src.utils.logging import accumulate_metric, conditional_observe @@ -275,9 +276,11 @@ async def summarize_if_needed( db_session, workspace_name, session_name, - message_id, - SummaryType.LONG, - message_public_id, + message_id=message_id, + message_seq_in_session=message_seq_in_session, + message_public_id=message_public_id, + summary_type=SummaryType.LONG, + configuration=configuration, ) accumulate_metric( f"summary_{workspace_name}_{message_id}", @@ -292,9 +295,11 @@ async def summarize_if_needed( db_session, workspace_name, session_name, - message_id, - SummaryType.SHORT, - message_public_id, + message_id=message_id, + message_seq_in_session=message_seq_in_session, + message_public_id=message_public_id, + summary_type=SummaryType.SHORT, + configuration=configuration, ) accumulate_metric( f"summary_{workspace_name}_{message_id}", @@ -316,9 +321,11 @@ async def summarize_if_needed( db, workspace_name, session_name, - message_id, - SummaryType.LONG, - message_public_id, + message_id=message_id, + message_seq_in_session=message_seq_in_session, + message_public_id=message_public_id, + summary_type=SummaryType.LONG, + configuration=configuration, ) accumulate_metric( f"summary_{workspace_name}_{message_id}", @@ -331,9 +338,11 @@ async def summarize_if_needed( db, workspace_name, session_name, - message_id, - SummaryType.SHORT, - message_public_id, + message_id=message_id, + message_seq_in_session=message_seq_in_session, + message_public_id=message_public_id, + summary_type=SummaryType.SHORT, + configuration=configuration, ) accumulate_metric( f"summary_{workspace_name}_{message_id}", @@ -347,9 +356,12 @@ async def _create_and_save_summary( db: AsyncSession, workspace_name: str, session_name: str, + *, message_id: int, - summary_type: SummaryType, + message_seq_in_session: int, message_public_id: str, + summary_type: SummaryType, + configuration: schemas.ResolvedConfiguration, ) -> None: """ Create a new summary and save it to the database. @@ -364,16 +376,33 @@ async def _create_and_save_summary( summary_start = time.perf_counter() latest_summary = await get_summary(db, workspace_name, session_name, summary_type) + if latest_summary: + latest_summary_message_id = latest_summary["message_id"] + # Skip if latest summary already covers message. + if latest_summary_message_id >= message_id: + return previous_summary_text = latest_summary["content"] if latest_summary else None - messages = await crud.get_messages_id_range( + # Calculate the sequence range for messages to summarize + # We want to get the last N messages where N is the configured summary interval + messages_per_summary = ( + configuration.summary.messages_per_long_summary + if summary_type == SummaryType.LONG + else configuration.summary.messages_per_short_summary + ) + start_seq = max(message_seq_in_session - messages_per_summary + 1, 1) + + messages: list[Message] = await crud.get_messages_by_seq_range( db, workspace_name, session_name, - start_id=latest_summary["message_id"] if latest_summary else 0, - end_id=message_id, + start_seq=start_seq, + end_seq=message_seq_in_session, ) + if not messages: + logger.warning("No messages to summarize for message %s", message_id) + return messages_tokens = sum([message.token_count for message in messages]) previous_summary_tokens = latest_summary["token_count"] if latest_summary else 0