fix: address interrupted-session ownership review follow-ups

alembic/versions/0008_session_conversation_id.py

@@ -23,5 +23,6 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    op.drop_index("ix_session_ownership_conversation_id", table_name="session_ownership")
-    op.drop_column("session_ownership", "conversation_id")
+    with op.batch_alter_table("session_ownership") as batch:
+        batch.drop_index("ix_session_ownership_conversation_id")
+        batch.drop_column("conversation_id")

docs/docs/core-components/chat.mdx

@@ -34,11 +34,19 @@ 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.
+After a backend restart, continuing with an owned response alias reuses the canonical sidebar thread and preserves its saved title and message count.
+Successful message bodies are still retrieved from Langflow; the backend does not reconstruct them from metadata.
 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.
+Bulk deletion also removes the caller's own legacy local conversation records that have no ownership row.
+Those orphan records do not authorize upstream session deletion or bypass the ownership checks on chat requests.
+Orphan cleanup is bulk-only at the API: single-conversation delete endpoints return `404` for records without an ownership row.
 
 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.
+File updates are serialized and persisted on a worker thread, so durable writes do not block the event loop or overwrite concurrent ownership changes.
+File writes sync both the file contents and the parent directory after atomic replacement before ownership is acknowledged.
+If directory sync fails after replacement, file-backed authorization is suspended and further file mutations require a successful reload and directory sync, preventing updates from overwriting ownership changes with stale cached data.
 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.
 

src/agent.py

@@ -161,6 +161,8 @@ async def store_conversation_thread(user_id: str, response_id: str, conversation
 
     # 2. Store only essential metadata to disk (simplified JSON)
     messages = conversation_state.get("messages", [])
+    # A restarted thread has durable metadata, but not successful message bodies.
+    persisted_metadata = conversation_state.get("_persisted_metadata", {})
     first_user_msg = next((msg for msg in messages if msg.get("role") == "user"), None)
     title = "New Chat"
     if first_user_msg:
@@ -170,17 +172,21 @@ async def store_conversation_thread(user_id: str, response_id: str, conversation
     user_assistant_messages = [msg for msg in messages if msg.get("role") in ["user", "assistant"]]
     metadata_only = {
         "response_id": response_id,
-        "title": title,
+        "title": persisted_metadata.get("title") or title,
         "endpoint": "langflow",
         "created_at": conversation_state.get("created_at"),
         "last_activity": conversation_state.get("last_activity"),
-        "previous_response_id": conversation_state.get("previous_response_id"),
+        "previous_response_id": persisted_metadata.get(
+            "previous_response_id", conversation_state.get("previous_response_id")
+        ),
         "filter_id": conversation_state.get("filter_id"),
-        "total_messages": len(user_assistant_messages),
+        "total_messages": (persisted_metadata.get("total_messages") or 0)
+        + len(user_assistant_messages),
     }
     # Snapshot error-thread turns: Langflow history is incomplete after fresh-session retries.
-    if any(msg.get("error") for msg in user_assistant_messages):
-        metadata_only["messages"] = [
+    persisted_messages = persisted_metadata.get("messages", [])
+    if persisted_messages or any(msg.get("error") for msg in user_assistant_messages):
+        metadata_only["messages"] = persisted_messages + [
             {
                 "role": msg.get("role"),
                 "content": msg.get("content"),
@@ -910,10 +916,37 @@ async def async_langflow_chat_stream(
             thread_id,
         )
     previous_loaded_from_memory = _conversation_thread_in_memory(user_id, thread_id)
-    conversation_state = get_conversation_thread(user_id, thread_id)
+    owned_session_ids = []
+    persisted_metadata = {}
+    if thread_id and not previous_loaded_from_memory:
+        from services.session_ownership_service import session_ownership_service
 
+        owned_session_ids = await session_ownership_service.get_conversation_sessions(
+            user_id, thread_id
+        )
+        if owned_session_ids:
+            thread_id = owned_session_ids[0]
+            previous_loaded_from_memory = _conversation_thread_in_memory(user_id, thread_id)
+            if not previous_loaded_from_memory:
+                persisted_metadata = await conversation_persistence.get_conversation_thread(
+                    user_id, thread_id
+                )
     from datetime import datetime
 
+    conversation_state = get_conversation_thread(user_id, thread_id)
+    if persisted_metadata:
+        conversation_state["_persisted_metadata"] = persisted_metadata
+        restored_created_at = persisted_metadata.get("created_at")
+        if isinstance(restored_created_at, str):
+            try:
+                restored_created_at = datetime.fromisoformat(restored_created_at)
+            except ValueError:
+                restored_created_at = None
+        if isinstance(restored_created_at, datetime):
+            conversation_state["created_at"] = restored_created_at
+        conversation_state["filter_id"] = persisted_metadata.get("filter_id")
+        conversation_state["response_ids"] = list(owned_session_ids)
+
     conversation_state["messages"].append(
         {"role": "user", "content": prompt, "timestamp": datetime.now()}
     )
@@ -930,7 +963,8 @@ async def async_langflow_chat_stream(
     usage_data = None
     collected_chunks = []  # Store all chunks for function call data
     error_occurred = False
-    store_id = None
+    # A durable owned group can safely retain its canonical sidebar ID on restart.
+    store_id = thread_id if owned_session_ids else None
     claimed_ids: set[str] = set()
     turn_finished = False
 
@@ -1127,9 +1161,14 @@ async def persist_stream_state(*session_ids, mint_if_missing=False):
             await upstream.aclose()
 
 
-async def delete_user_conversation(user_id: str, response_id: str) -> bool:
+async def delete_user_conversation(
+    user_id: str, response_id: str, *, session_ids: list[str] | None = None
+) -> bool:
     """Delete a conversation for a user from both memory and persistent storage.
 
+    ``session_ids`` is the caller's already-authorized ownership snapshot, when
+    provided. An empty group permits only user-scoped local orphan cleanup.
+
     Returns:
         True  — conversation was found and deleted from at least one store.
         False — conversation did not exist in any store (confirmed not-found).
@@ -1140,13 +1179,15 @@ async def delete_user_conversation(user_id: str, response_id: str) -> bool:
     """
     from services.session_ownership_service import session_ownership_service
 
-    session_ids = await session_ownership_service.get_conversation_sessions(user_id, response_id)
-    if not session_ids:
-        return False
-    canonical_id = session_ids[0]
-    # Revoke ALL durable aliases before removing the local index. Failures must
-    # reach the caller, not turn into a successful delete with reusable IDs.
-    await session_ownership_service.release_conversation_sessions(user_id, session_ids)
+    if session_ids is None:
+        session_ids = await session_ownership_service.get_conversation_sessions(
+            user_id, response_id
+        )
+    canonical_id = session_ids[0] if session_ids else response_id
+    if session_ids:
+        # Revoke only the snapshot used for upstream cleanup. Never revoke a late
+        # alias whose upstream deletion was not attempted.
+        await session_ownership_service.release_conversation_sessions(user_id, session_ids)
     deleted = await conversation_persistence.delete_conversation_thread(user_id, canonical_id)
     if canonical_id in active_conversations.get(user_id, {}):
         del active_conversations[user_id][canonical_id]

src/db/repositories/session_ownership_repo.py

@@ -3,6 +3,7 @@
 from datetime import UTC, datetime
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlmodel import col
 
@@ -27,11 +28,6 @@ async def claim(
             canonical = await self.get(canonical_id)
             if canonical is None or canonical.user_id != user_id:
                 raise PermissionError("Conversation is not owned by this user")
-        if existing is not None and (
-            existing.user_id != user_id
-            or (existing.conversation_id is not None and existing.conversation_id != canonical_id)
-        ):
-            raise PermissionError("Session ownership or conversation conflict")
         if existing is None:
             row = SessionOwnership(
                 response_id=response_id,
@@ -40,9 +36,40 @@ async def claim(
                 created_at=now,
                 last_accessed=now,
             )
-            self.session.add(row)
-            await self.session.flush()
-            return row
+            try:
+                # begin_nested flushes pending caller work before establishing
+                # its savepoint, so add only this claim inside the savepoint.
+                async with self.session.begin_nested():
+                    self.session.add(row)
+                    await self.session.flush()
+            except IntegrityError as exc:
+                # Only primary-key/unique violations can be an INSERT race.
+                # Never hide another constraint failure behind a matching row.
+                if (
+                    getattr(exc.orig, "sqlstate", None) != "23505"
+                    and getattr(exc.orig, "sqlite_errorname", None)
+                    != "SQLITE_CONSTRAINT_PRIMARYKEY"
+                ):
+                    raise
+                # The savepoint has rolled back only this insert; preserve the
+                # caller's transaction while reading the winning claim afresh.
+                existing = await self.session.get(
+                    SessionOwnership, response_id, populate_existing=True
+                )
+                if existing is None:
+                    raise
+                if canonical_id != response_id:
+                    canonical = await self.session.get(
+                        SessionOwnership, canonical_id, populate_existing=True
+                    )
+                    if canonical is None or canonical.user_id != user_id:
+                        raise PermissionError("Conversation is not owned by this user") from None
+            else:
+                return row
+        if existing.user_id != user_id or (
+            (existing.conversation_id or response_id) != canonical_id
+        ):
+            raise PermissionError("Session ownership or conversation conflict")
         existing.conversation_id = canonical_id
         existing.last_accessed = now
         self.session.add(existing)

src/services/chat_service.py

@@ -688,11 +688,12 @@ async def get_chat_history(self, user_id: str):
                     if first_user_msg
                     else "New chat"
                 )
+                persisted_metadata = conversation_state.get("_persisted_metadata", {})
 
                 conversations.append(
                     {
                         "response_id": response_id,
-                        "title": title,
+                        "title": persisted_metadata.get("title") or title,
                         "endpoint": "chat",
                         "messages": messages,
                         "created_at": conversation_state.get("created_at").isoformat()
@@ -703,7 +704,8 @@ async def get_chat_history(self, user_id: str):
                         else None,
                         "previous_response_id": conversation_state.get("previous_response_id"),
                         "filter_id": conversation_state.get("filter_id"),
-                        "total_messages": len(messages),
+                        "total_messages": (persisted_metadata.get("total_messages") or 0)
+                        + len(messages),
                         "source": "in_memory",
                     }
                 )
@@ -948,7 +950,9 @@ async def delete_session(self, user_id: str, session_id: str):
             ]
             langflow_deleted = bool(upstream_results) and all(upstream_results)
 
-            local_deleted = await delete_user_conversation(user_id, session_id)
+            local_deleted = await delete_user_conversation(
+                user_id, session_id, session_ids=session_ids
+            )
 
             if not local_deleted:
                 return {
@@ -981,7 +985,7 @@ async def delete_sessions(self, user_id: str, session_ids: list[str]) -> dict[st
         for session_id in dict.fromkeys(session_ids):
             try:
                 owner = await session_ownership_service.get_session_owner(session_id)
-                if owner != user_id:
+                if owner is not None and owner != user_id:
                     logger.warning("bulk_delete: %s not owned by %s", session_id, user_id)
                     failed.append(session_id)
                     continue