Fix closing sessions

openhands/server/listen_socket.py

@@ -30,7 +30,7 @@ async def connect(connection_id: str, environ, auth):
         logger.error('No conversation_id in query params')
         raise ConnectionRefusedError('No conversation_id in query params')
 
-    user_id = -1
+    user_id = None
     if openhands_config.app_mode != AppMode.OSS:
         cookies_str = environ.get('HTTP_COOKIE', '')
         cookies = dict(cookie.split('=', 1) for cookie in cookies_str.split('; '))
@@ -61,7 +61,7 @@ async def connect(connection_id: str, environ, auth):
 
     try:
         event_stream = await session_manager.join_conversation(
-            conversation_id, connection_id, settings
+            conversation_id, connection_id, settings, user_id
         )
     except ConversationDoesNotExistError:
         logger.error(f'Conversation {conversation_id} does not exist')

openhands/server/routes/manage_conversations.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Callable
 
 from fastapi import APIRouter, Body, Request
@@ -42,7 +42,8 @@ async def new_conversation(request: Request, data: InitSessionRequest):
     logger.info('Initializing new conversation')
 
     logger.info('Loading settings')
-    settings_store = await SettingsStoreImpl.get_instance(config, get_user_id(request))
+    user_id = get_user_id(request)
+    settings_store = await SettingsStoreImpl.get_instance(config, user_id)
     settings = await settings_store.load()
     logger.info('Settings loaded')
 
@@ -55,9 +56,7 @@ async def new_conversation(request: Request, data: InitSessionRequest):
     session_init_args['selected_repository'] = data.selected_repository
     conversation_init_data = ConversationInitData(**session_init_args)
     logger.info('Loading conversation store')
-    conversation_store = await ConversationStoreImpl.get_instance(
-        config, get_user_id(request)
-    )
+    conversation_store = await ConversationStoreImpl.get_instance(config, user_id)
     logger.info('Conversation store loaded')
 
     conversation_id = uuid.uuid4().hex
@@ -75,19 +74,19 @@ async def new_conversation(request: Request, data: InitSessionRequest):
         ConversationMetadata(
             conversation_id=conversation_id,
             title=conversation_title,
-            github_user_id=get_user_id(request),
+            github_user_id=user_id,
             selected_repository=data.selected_repository,
         )
     )
 
     logger.info(f'Starting agent loop for conversation {conversation_id}')
     event_stream = await session_manager.maybe_start_agent_loop(
-        conversation_id, conversation_init_data
+        conversation_id, conversation_init_data, user_id
     )
     try:
         event_stream.subscribe(
             EventStreamSubscriber.SERVER,
-            _create_conversation_update_callback(get_user_id(request), conversation_id),
+            _create_conversation_update_callback(user_id, conversation_id),
             UPDATED_AT_CALLBACK_ID,
         )
     except ValueError:
@@ -111,8 +110,8 @@ async def search_conversations(
         for conversation in conversation_metadata_result_set.results
         if hasattr(conversation, 'created_at')
     )
-    running_conversations = await session_manager.get_agent_loop_running(
-        set(conversation_ids)
+    running_conversations = await session_manager.get_running_agent_loops(
+        get_user_id(request), set(conversation_ids)
     )
     result = ConversationInfoResultSet(
         results=await wait_all(
@@ -221,5 +220,5 @@ def callback(*args, **kwargs):
 async def _update_timestamp_for_conversation(user_id: int, conversation_id: str):
     conversation_store = await ConversationStoreImpl.get_instance(config, user_id)
     conversation = await conversation_store.get_metadata(conversation_id)
-    conversation.last_updated_at = datetime.now()
+    conversation.last_updated_at = datetime.now(timezone.utc)
     await conversation_store.save_metadata(conversation)