github/OpenHands
1
0
Fork 0
mirror of https://github.com/OpenHands/OpenHands.git synced 2026-03-22 13:47:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Use the same event stream instance for conversations as sessions (#9545)

Browse Source 
Co-authored-by: openhands <openhands@all-hands.dev>
This commit is contained in:
Tim O'Farrell
2025-07-07 14:37:17 -06:00
committed by GitHub
parent 7cfecb6e52
commit 517a72fd0d
4 changed files with 68 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
openhands/server/conversation_manager/standalone_conversation_manager.py
View File

@@ -105,9 +105,22 @@ class StandaloneConversationManager(ConversationManager):
)
return conversation
# Get the event stream for the conversation - required to keep the cur_id up to date
event_stream = None
runtime = None
session = self._local_agent_loops_by_sid.get(sid)
if session:
event_stream = session.agent_session.event_stream
runtime = session.agent_session.runtime
# Create new conversation if none exists
c = ServerConversation(
sid, file_store=self.file_store, config=self.config, user_id=user_id
sid,
file_store=self.file_store,
config=self.config,
user_id=user_id,
event_stream=event_stream,
runtime=runtime,
)
try:
await c.connect()

34
openhands/server/session/conversation.py
View File

@@ -15,6 +15,7 @@ class ServerConversation:
event_stream: EventStream
runtime: Runtime
user_id: str | None
_attach_to_existing: bool = False
def __init__(
self,
@@ -22,30 +23,43 @@ class ServerConversation:
file_store: FileStore,
config: OpenHandsConfig,
user_id: str | None,
event_stream: EventStream | None = None,
runtime: Runtime | None = None,
):
self.sid = sid
self.config = config
self.file_store = file_store
self.user_id = user_id
self.event_stream = EventStream(sid, file_store, user_id)
if event_stream is None:
event_stream = EventStream(sid, file_store, user_id)
self.event_stream = event_stream
if config.security.security_analyzer:
self.security_analyzer = options.SecurityAnalyzers.get(
config.security.security_analyzer, SecurityAnalyzer
)(self.event_stream)
runtime_cls = get_runtime_cls(self.config.runtime)
self.runtime = runtime_cls(
config=config,
event_stream=self.event_stream,
sid=self.sid,
attach_to_existing=True,
headless_mode=False,
)
if runtime:
self._attach_to_existing = True
else:
runtime_cls = get_runtime_cls(self.config.runtime)
runtime = runtime_cls(
config=config,
event_stream=self.event_stream,
sid=self.sid,
attach_to_existing=True,
headless_mode=False,
)
self.runtime = runtime
async def connect(self) -> None:
await self.runtime.connect()
if not self._attach_to_existing:
await self.runtime.connect()
async def disconnect(self) -> None:
if self._attach_to_existing:
return
if self.event_stream:
self.event_stream.close()
asyncio.create_task(call_sync_from_async(self.runtime.close))