From c1e62877aec54e4cd51f730a6e5317ea25d32491 Mon Sep 17 00:00:00 2001 From: Edwin Jose Date: Wed, 21 Oct 2026 13:55:17 -0500 Subject: [PATCH 1/4] fix: keep interrupted Langflow streams owned and resumable (#2527) * fix: persist Langflow session ownership before streaming IDs * style: ruff autofix (auto) * fix: persist session aliases and require durable ownership claims * style: ruff autofix (auto) * style: apply CI Ruff fixes for session persistence * fix: remove unused session ownership imports * fix: type SQLModel column expressions for CI mypy --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> (cherry picked from commit 93a3ad15d402937541e2fe0d3525a9f44c2a7904) --- .../versions/0008_session_conversation_id.py | 27 + docs/docs/core-components/chat.mdx | 20 + src/agent.py | 232 +++++--- src/db/migrations_runtime.py | 1 + src/db/models/session_ownership.py | 2 + src/db/repositories/permission_repo.py | 9 +- src/db/repositories/session_ownership_repo.py | 44 +- src/services/chat_service.py | 17 +- .../conversation_persistence_service.py | 57 +- src/services/session_ownership_service.py | 205 ++++++- .../migrations/test_chat_history_migration.py | 139 +++-- .../test_session_ownership_service_modes.py | 218 ++++++- .../unit/test_agent_stream_fence_stripped.py | 3 + .../test_langflow_stream_provider_errors.py | 542 +++++++++++++++++- 14 files changed, 1300 insertions(+), 216 deletions(-) create mode 100644 alembic/versions/0008_session_conversation_id.py diff --git a/alembic/versions/0008_session_conversation_id.py b/alembic/versions/0008_session_conversation_id.py new file mode 100644 index 0000000000..a28a6a34d1 --- /dev/null +++ b/alembic/versions/0008_session_conversation_id.py @@ -0,0 +1,27 @@ +"""Persist canonical conversation IDs for session aliases. + +Revision ID: 0008_session_conversation_id +Revises: 0007_add_knowledge_delete_anonymous +""" + +import sqlalchemy as sa + +from alembic import op + +revision = "0008_session_conversation_id" +down_revision = "0007_add_knowledge_delete_anonymous" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("session_ownership", sa.Column("conversation_id", sa.String(64), nullable=True)) + op.create_index( + "ix_session_ownership_conversation_id", "session_ownership", ["conversation_id"] + ) + op.execute("UPDATE session_ownership SET conversation_id = response_id") + + +def downgrade() -> None: + op.drop_index("ix_session_ownership_conversation_id", table_name="session_ownership") + op.drop_column("session_ownership", "conversation_id") diff --git a/docs/docs/core-components/chat.mdx b/docs/docs/core-components/chat.mdx index 6ee554d9c7..91255de7c4 100644 --- a/docs/docs/core-components/chat.mdx +++ b/docs/docs/core-components/chat.mdx @@ -22,6 +22,26 @@ To add or change the available tools, you must edit the [**OpenRAG OpenSearch Ag Try chatting, uploading documents, and modifying chat settings in the [quickstart](/quickstart). ::: +## Interrupted chat responses + +If you stop a response or the connection closes while an answer is streaming, OpenRAG retains ownership of response IDs already sent to your client. +You can send a follow-up in the same conversation without losing access solely because the previous stream was interrupted. +The backend keeps the partial answer in the active conversation and records the updated conversation metadata when the stream is closed. +An interrupted answer is not automatically regenerated, and ingestion or other tool calls are not automatically replayed. + +For API clients, `conversation_id` identifies the OpenRAG sidebar thread, while `previous_response_id` continues the Langflow session. +These IDs can differ when Langflow returns a new response ID. +Both IDs are checked against the authenticated user's session ownership; interruption recovery does not bypass those checks. + +Response aliases are durably associated with their canonical conversation, including across backend restarts. +Deleting a conversation revokes its canonical ID and all recorded aliases and attempts to remove every associated Langflow session. +A deleted alias cannot be used to resume the conversation. + +IDs are sent only after ownership and its conversation association have been persisted. +In `files` mode this requires a successful file write; in `db` mode it requires a committed database write. +In `hybrid` mode the database is preferred, with a successful durable JSON write as fallback if the database is unavailable. +If neither configured destination can persist the claim, the stream returns an error without exposing a new response ID; an in-memory cache entry is not sufficient. + ## OpenRAG OpenSearch Agent flow {#flow} When you use the OpenRAG