Skip to content

Simplify lazy conversation catalog invariants #4122

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
enyst wants to merge 1 commit into from
Closed

Simplify lazy conversation catalog invariants #4122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 83 additions & 63 deletions openhands-agent-server/openhands/agent_server/conversation_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,9 @@
from openhands.agent_server.config import Config, WebhookSpec
from openhands.agent_server.conversation_lease import (
DEFAULT_LEASE_TTL_SECONDS,
ConversationLease,
ConversationLeaseHeldError,
ConversationOwnershipLostError,
)
from openhands.agent_server.event_service import (
LEASE_RENEW_INTERVAL_SECONDS,
Expand Down Expand Up @@ -501,6 +503,7 @@ class ConversationService:
owner_instance_id: str = field(default_factory=lambda: uuid4().hex)
max_concurrent_runs: int = 10
lease_ttl_seconds: float = DEFAULT_LEASE_TTL_SECONDS
# Every live runtime has a catalog record; unloaded records have no runtime.
_event_services: dict[UUID, EventService] | None = field(default=None, init=False)
_conversation_records: dict[UUID, _ConversationRecord] = field(
default_factory=dict, init=False
Expand Down Expand Up @@ -594,25 +597,6 @@ async def _execution_status(
record.execution_status = state.execution_status
return record.execution_status

async def _reconcile_active_records(self) -> None:
"""Fill catalog entries for services injected outside normal startup.

Normal service lifecycle paths maintain the catalog themselves. This
small reconciliation keeps direct embedders and existing test fixtures
that populate ``_event_services`` compatible.
"""
event_services = self._event_services
if event_services is None:
raise ValueError("inactive_service")
for conversation_id, event_service in event_services.items():
if conversation_id in self._conversation_records:
continue
state = await event_service.get_state()
self._conversation_records[conversation_id] = _ConversationRecord(
stored=event_service.stored,
execution_status=state.execution_status,
)

def _prepare_persisted_runtime(self, stored: StoredConversation) -> None:
if stored.tool_module_qualnames:
for tool_name, module_qualname in stored.tool_module_qualnames.items():
Expand Down Expand Up @@ -679,16 +663,7 @@ async def get_conversation(self, conversation_id: UUID) -> ConversationInfo | No
raise ValueError("inactive_service")
record = self._conversation_records.get(conversation_id)
if record is None:
event_service = self._event_services.get(conversation_id)
if event_service is None:
return None
state = await event_service.get_state()
record = _ConversationRecord(
stored=event_service.stored,
execution_status=state.execution_status,
)
self._conversation_records[conversation_id] = record
return _compose_conversation_info(event_service.stored, state)
return None
return await self._conversation_info(conversation_id, record)

async def get_acp_conversation(
Expand Down Expand Up @@ -741,7 +716,6 @@ async def _search_conversations(
) -> tuple[list[ConversationInfo], str | None]:
if self._event_services is None:
raise ValueError("inactive_service")
await self._reconcile_active_records()

records = list(self._conversation_records.items())
if sort_order in (
Expand Down Expand Up @@ -798,7 +772,6 @@ async def _count_conversations(
"""Count conversations matching the given filters."""
if self._event_services is None:
raise ValueError("inactive_service")
await self._reconcile_active_records()

if execution_status is None:
return len(self._conversation_records)
Expand Down Expand Up @@ -887,10 +860,6 @@ async def _start_conversation(
existing_event_service = self._event_services.get(conversation_id)
if existing_event_service is not None and existing_event_service.is_open():
state = await existing_event_service.get_state()
self._conversation_records[conversation_id] = _ConversationRecord(
stored=existing_event_service.stored,
execution_status=state.execution_status,
)
return (
_compose_conversation_info(existing_event_service.stored, state),
False,
Expand Down Expand Up @@ -1081,45 +1050,96 @@ async def delete_conversation(self, conversation_id: UUID) -> bool:
event_services = self._event_services
if event_services is None:
raise ValueError("inactive_service")
event_service = await self._get_or_load_event_service_locked(
conversation_id
)
if event_service is None:
record = self._conversation_records.get(conversation_id)
if record is None:
return False

event_services.pop(conversation_id, None)
self._conversation_records.pop(conversation_id, None)
# Notify conversation webhooks about the stopped conversation before closing
try:
state = await event_service.get_state()
conversation_info = _compose_webhook_conversation_info(
event_service.stored, state
event_service = event_services.get(conversation_id)
if event_service is not None and not event_service.is_open():
event_services.pop(conversation_id, None)
event_service = None
conversation_dir = (
event_service.conversation_dir
if event_service is not None
else self.conversations_dir / conversation_id.hex
)
deletion_lease: ConversationLease | None = None
lease_generation: int | None = None
if event_service is None and self.lease_ttl_seconds > 0:
deletion_lease = ConversationLease(
conversation_dir=conversation_dir,
owner_instance_id=self.owner_instance_id,
ttl_seconds=self.lease_ttl_seconds,
)
conversation_info.execution_status = (
ConversationExecutionStatus.DELETING
try:
claim = await asyncio.to_thread(deletion_lease.claim)
except ConversationLeaseHeldError as exc:
logger.debug(
"Skipping deletion of conversation %s owned by %s until %s",
conversation_id,
exc.owner_instance_id,
exc.expires_at,
)
return False
lease_generation = claim.generation

try:
state = (
await event_service.get_state()
if event_service is not None
else await asyncio.to_thread(
self._load_persisted_state_sync, conversation_id
)
)
await self._notify_conversation_webhooks(conversation_info)
if state is not None:
stored = (
event_service.stored
if event_service is not None
else record.stored
)
conversation_info = _compose_webhook_conversation_info(
stored, state
)
conversation_info.execution_status = (
ConversationExecutionStatus.DELETING
)
await self._notify_conversation_webhooks(conversation_info)
except Exception as e:
logger.warning(
f"Failed to notify webhooks for conversation {conversation_id}: {e}"
)

# Close the event service
try:
await event_service.close()
except Exception as e:
logger.warning(
f"Failed to close event service for conversation "
f"{conversation_id}: {e}"
)
if deletion_lease is not None and lease_generation is not None:
try:
await asyncio.to_thread(deletion_lease.renew, lease_generation)
except ConversationOwnershipLostError:
logger.warning(
"Lost ownership before deleting conversation %s",
conversation_id,
)
return False

event_services.pop(conversation_id, None)
self._conversation_records.pop(conversation_id, None)
if event_service is not None:
try:
await event_service.close()
except Exception as e:
logger.warning(
f"Failed to close event service for conversation "
f"{conversation_id}: {e}"
)

# Safely remove only the conversation directory (workspace is preserved).
# This operation may fail due to permission issues, but we don't want that
# to prevent the conversation from being marked as deleted.
safe_rmtree(
event_service.conversation_dir,
removed = safe_rmtree(
conversation_dir,
f"conversation directory for {conversation_id}",
)
if (
not removed
and deletion_lease is not None
and lease_generation is not None
):
await asyncio.to_thread(deletion_lease.release, lease_generation)

logger.info(f"Successfully deleted conversation {conversation_id}")
return True
Expand Down Expand Up @@ -1483,12 +1503,12 @@ async def _start_event_service(self, stored: StoredConversation) -> EventService
await event_service.close()
raise

event_services[stored.id] = event_service
state = await event_service.get_state()
self._conversation_records[stored.id] = _ConversationRecord(
stored=event_service.stored,
execution_status=state.execution_status,
)
event_services[stored.id] = event_service
return event_service


Expand Down
Loading
Loading