From e9b67372b9de7b22178e5e89b2eb96cc9f202f8d Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 15:20:38 +0000 Subject: [PATCH 01/10] feat(agent-server): add explicit conversation archiving Co-authored-by: openhands --- .../src/__tests__/api-clients.test.ts | 12 +++ .../src/client/conversation-client.ts | 16 ++++ clients/typescript/src/models/conversation.ts | 2 + .../openhands/agent_server/README.md | 11 +++ .../agent_server/conversation_router.py | 35 +++++++- .../agent_server/conversation_service.py | 50 +++++++++++- .../agent_server/docker_runtime/routers.py | 46 +++++++++++ .../openhands/agent_server/models.py | 8 ++ .../docker_runtime/test_routes.py | 80 ++++++++++++++++++- .../agent_server/test_conversation_router.py | 46 ++++++++++- .../agent_server/test_conversation_service.py | 51 ++++++++++++ 11 files changed, 351 insertions(+), 6 deletions(-) diff --git a/clients/typescript/src/__tests__/api-clients.test.ts b/clients/typescript/src/__tests__/api-clients.test.ts index 7c6f8af3093..5abdaf69140 100644 --- a/clients/typescript/src/__tests__/api-clients.test.ts +++ b/clients/typescript/src/__tests__/api-clients.test.ts @@ -2288,6 +2288,8 @@ describe('Auxiliary API clients', () => { await client.updateConversation('c1', { title: 'New title' }); await client.getRuntime('c1'); await client.reprovisionRuntime('c1'); + await client.archiveConversation('c1'); + await client.unarchiveConversation('c1'); expect(global.fetch).toHaveBeenNthCalledWith( 1, @@ -2327,6 +2329,16 @@ describe('Auxiliary API clients', () => { 'http://example.com/api/conversations/c1/runtime/reprovision', expect.objectContaining({ method: 'POST', body: JSON.stringify({}) }) ); + expect(global.fetch).toHaveBeenNthCalledWith( + 11, + 'http://example.com/api/conversations/c1/archive', + expect.objectContaining({ method: 'POST', body: JSON.stringify({}) }) + ); + expect(global.fetch).toHaveBeenNthCalledWith( + 12, + 'http://example.com/api/conversations/c1/unarchive', + expect.objectContaining({ method: 'POST', body: JSON.stringify({}) }) + ); }); it('ConversationClient wraps SDK v1.23.0 conversation endpoints', async () => { diff --git a/clients/typescript/src/client/conversation-client.ts b/clients/typescript/src/client/conversation-client.ts index 6b001064f87..f1029656af1 100644 --- a/clients/typescript/src/client/conversation-client.ts +++ b/clients/typescript/src/client/conversation-client.ts @@ -124,6 +124,22 @@ export class ConversationClient { return response.data; } + async archiveConversation(conversationId: string): Promise { + const response = await this.client.post( + `/api/conversations/${conversationId}/archive`, + {} + ); + return response.data; + } + + async unarchiveConversation(conversationId: string): Promise { + const response = await this.client.post( + `/api/conversations/${conversationId}/unarchive`, + {} + ); + return response.data; + } + async searchEvents( conversationId: string, options: ConversationEventSearchOptions = {} diff --git a/clients/typescript/src/models/conversation.ts b/clients/typescript/src/models/conversation.ts index bfd1b6eec14..edf6c5af284 100644 --- a/clients/typescript/src/models/conversation.ts +++ b/clients/typescript/src/models/conversation.ts @@ -67,6 +67,7 @@ export interface ConversationInfo { title?: string; created_at?: string; updated_at?: string; + archived_at?: string | null; tags?: Record; /** * HEAD of the conversation tree: the parent of the next appended event. @@ -171,6 +172,7 @@ export interface ConversationSearchRequest { limit?: number; status?: ConversationExecutionStatus; sort_order?: ConversationSortOrder; + archived?: boolean; tag?: string[]; } diff --git a/openhands-agent-server/openhands/agent_server/README.md b/openhands-agent-server/openhands/agent_server/README.md index 85c1f8b44b1..1e074077734 100644 --- a/openhands-agent-server/openhands/agent_server/README.md +++ b/openhands-agent-server/openhands/agent_server/README.md @@ -12,6 +12,17 @@ The OpenHands Agent Server is a minimal REST API and WebSocket server that provi - **Webhooks**: Configurable webhook notifications for events - **Auto-reload**: Development mode with automatic code reloading + +## Conversation lifecycle + +Conversation metadata separates retained history from runtime availability: + +- `runtime_status` reports whether execution infrastructure is available; it does not imply archive state. +- `POST /api/conversations/{id}/archive` sets `archived_at`, hides the conversation from the default search, and stops only runtime infrastructure owned by this server. It preserves history and workspace files. +- `POST /api/conversations/{id}/unarchive` clears `archived_at` without provisioning infrastructure or starting agent execution. +- `GET /api/conversations/search?archived=true` returns archived conversations; the default `archived=false` returns active conversations. +- `DELETE /api/conversations/{id}` permanently removes the conversation rather than retaining it as archived. + ## Quick Start ### Prerequisites diff --git a/openhands-agent-server/openhands/agent_server/conversation_router.py b/openhands-agent-server/openhands/agent_server/conversation_router.py index 3782873a79d..880845a6b3d 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_router.py +++ b/openhands-agent-server/openhands/agent_server/conversation_router.py @@ -121,12 +121,15 @@ async def search_conversations( ConversationSortOrder, Query(title="Sort order for conversations"), ] = ConversationSortOrder.CREATED_AT_DESC, + archived: Annotated[ + bool, Query(title="Return archived rather than active conversations") + ] = False, include_skills: Annotated[bool, Query(title=INCLUDE_SKILLS_PARAM_TITLE)] = False, conversation_service: ConversationService = Depends(get_conversation_service), ) -> ConversationPage: """Search / List conversations""" page = await conversation_service.search_conversations( - page_id, limit, status, sort_order + page_id, limit, status, sort_order, archived ) if not include_skills: # ``model_copy`` rather than in-place mutation so we never @@ -207,6 +210,36 @@ async def reprovision_local_conversation_runtime( ) +@conversation_router.post("/{conversation_id}/archive") +async def archive_conversation( + conversation_id: UUID, + request: Request, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> ConversationInfo: + """Archive retained history and stop only its currently owned runtime.""" + conversation = await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + if conversation is None: + raise HTTPException(status.HTTP_404_NOT_FOUND) + return _with_runtime_info(request, conversation) + + +@conversation_router.post("/{conversation_id}/unarchive") +async def unarchive_conversation( + conversation_id: UUID, + request: Request, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> ConversationInfo: + """Restore catalog visibility without provisioning or starting execution.""" + conversation = await conversation_service.set_conversation_archived( + conversation_id, archived=False + ) + if conversation is None: + raise HTTPException(status.HTTP_404_NOT_FOUND) + return _with_runtime_info(request, conversation) + + @conversation_router.get( "/{conversation_id}/agent_final_response", responses={404: {"description": "Conversation not found"}}, diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 032da8e9117..f3ff3ebf7ee 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -530,6 +530,7 @@ def _compose_conversation_info( metrics=stored.metrics, created_at=stored.created_at, updated_at=stored.updated_at, + archived_at=stored.archived_at, forked_from_conversation_id=stored.forked_from_conversation_id, forked_from_event_id=stored.forked_from_event_id, parent_conversation_id=stored.parent_conversation_id, @@ -1239,7 +1240,7 @@ async def _get_or_load_event_service_locked( return event_service record = self._conversation_records.get(conversation_id) - if record is None: + if record is None or record.stored.archived_at is not None: return None pending_bindings = self._credential_bindings.get(conversation_id, {}) @@ -1293,12 +1294,14 @@ async def search_conversations( limit: int = 100, execution_status: ConversationExecutionStatus | None = None, sort_order: ConversationSortOrder = ConversationSortOrder.CREATED_AT_DESC, + archived: bool = False, ) -> ConversationPage: items, next_page_id = await self._search_conversations( page_id=page_id, limit=limit, execution_status=execution_status, sort_order=sort_order, + archived=archived, ) return ConversationPage( items=items, @@ -1317,6 +1320,7 @@ async def search_acp_conversations( limit=limit, execution_status=execution_status, sort_order=sort_order, + archived=False, ) return ConversationPage( items=items, @@ -1329,6 +1333,7 @@ async def _search_conversations( limit: int, execution_status: ConversationExecutionStatus | None, sort_order: ConversationSortOrder, + archived: bool, ) -> tuple[list[ConversationInfo], str | None]: if self._event_services is None: raise ValueError("inactive_service") @@ -1344,7 +1349,10 @@ async def _search_conversations( records = [ (conversation_id, record) for conversation_id, record in self._conversation_records.items() - if execution_status is None or record.execution_status == execution_status + if (record.stored.archived_at is not None) == archived + and ( + execution_status is None or record.execution_status == execution_status + ) ] if sort_order in ( ConversationSortOrder.CREATED_AT, @@ -2032,6 +2040,44 @@ async def update_conversation( ) return True + async def set_conversation_archived( + self, conversation_id: UUID, *, archived: bool + ) -> ConversationInfo | None: + """Persist archive state without loading or starting a conversation runtime.""" + async with self._conversation_lifecycle(conversation_id): + await self._reconcile_active_records() + record = self._conversation_records.get(conversation_id) + if record is None: + return None + + now = utc_now() + archived_at = (record.stored.archived_at or now) if archived else None + record.stored = record.stored.model_copy( + update={"archived_at": archived_at, "updated_at": now} + ) + record.cached_info = None + + event_service = ( + self._event_services.get(conversation_id) + if self._event_services is not None + else None + ) + if event_service is not None and event_service.is_open(): + event_service.stored = record.stored + + metadata = record.stored.model_dump_json( + context={"cipher": self._cipher_for(conversation_id)} + ) + meta_file = self.conversations_dir / conversation_id.hex / "meta.json" + await asyncio.to_thread(meta_file.write_text, metadata) + + if archived and event_service is not None and event_service.is_open(): + await event_service.close() + if self._event_services is not None: + self._event_services.pop(conversation_id, None) + + return await self._conversation_info(conversation_id, record) + async def get_event_service(self, conversation_id: UUID) -> EventService | None: return await self._get_or_load_event_service(conversation_id) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py index 2929b969448..fffcf7a8a61 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -28,6 +28,7 @@ DockerConversationRegistry, ) from openhands.agent_server.models import ( + ConversationInfo, ConversationRuntimeInfo, ConversationRuntimeStatus, UpdateSecretsRequest, @@ -47,11 +48,22 @@ def get_registry(request: Request) -> DockerConversationRegistry: return registry +def _is_archived(registry: DockerConversationRegistry, conversation_id: UUID) -> bool: + metadata_path = registry.conversation_dir(conversation_id) / "meta.json" + if not metadata_path.is_file(): + return False + return json.loads(metadata_path.read_text()).get("archived_at") is not None + + async def _container( registry: DockerConversationRegistry, conversation_id: UUID ) -> ConversationContainer: if not registry.provisioning.manifest_path(conversation_id).is_file(): raise HTTPException(404, "Conversation not found") + if _is_archived(registry, conversation_id): + raise HTTPException( + 409, "Conversation is archived; unarchive it before using its runtime" + ) try: return await registry.get_or_create(conversation_id) except Exception as exc: @@ -197,6 +209,40 @@ async def reprovision_runtime( ) +async def _set_archive_state( + conversation_id: UUID, request: Request, *, archived: bool +) -> ConversationInfo: + registry = get_registry(request) + conversation = await get_conversation_service(request).set_conversation_archived( + conversation_id, archived=archived + ) + if conversation is None: + raise HTTPException(404, "Conversation not found") + if archived: + await registry.stop(conversation_id) + return conversation.model_copy( + update={"runtime_info": registry.runtime_info(conversation_id)} + ) + + +@docker_conversation_router.post( + "/{conversation_id}/archive", response_model=ConversationInfo +) +async def archive_conversation( + conversation_id: UUID, request: Request +) -> ConversationInfo: + return await _set_archive_state(conversation_id, request, archived=True) + + +@docker_conversation_router.post( + "/{conversation_id}/unarchive", response_model=ConversationInfo +) +async def unarchive_conversation( + conversation_id: UUID, request: Request +) -> ConversationInfo: + return await _set_archive_state(conversation_id, request, archived=False) + + @docker_conversation_router.delete("/{conversation_id}") async def delete_conversation(conversation_id: UUID, request: Request) -> Response: registry = get_registry(request) diff --git a/openhands-agent-server/openhands/agent_server/models.py b/openhands-agent-server/openhands/agent_server/models.py index c07eb6f467e..9a6a897d72f 100644 --- a/openhands-agent-server/openhands/agent_server/models.py +++ b/openhands-agent-server/openhands/agent_server/models.py @@ -119,6 +119,10 @@ class StoredConversation(ConversationConfig): metrics: MetricsSnapshot | None = None created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) + archived_at: datetime | None = Field( + default=None, + description="When this conversation was archived by the control plane.", + ) forked_from_conversation_id: OpenHandsUUID | None = Field( default=None, description=( @@ -245,6 +249,10 @@ class _ConversationInfoBase(BaseModel): metrics: MetricsSnapshot | None = None created_at: datetime = Field(default_factory=utc_now) updated_at: datetime = Field(default_factory=utc_now) + archived_at: datetime | None = Field( + default=None, + description="UTC datetime the conversation was archived, if applicable.", + ) # Plain ``UUID`` (not ``OpenHandsUUID``) so it JSON-serializes dashed, exactly # like the ``id`` field above — clients must be able to correlate the two. forked_from_conversation_id: UUID | None = Field( diff --git a/tests/agent_server/docker_runtime/test_routes.py b/tests/agent_server/docker_runtime/test_routes.py index 9a532a5224b..e2d9bc35e29 100644 --- a/tests/agent_server/docker_runtime/test_routes.py +++ b/tests/agent_server/docker_runtime/test_routes.py @@ -1,7 +1,7 @@ import asyncio import json from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, call from uuid import uuid4 import pytest @@ -22,9 +22,12 @@ proxy_conversation, ) from openhands.agent_server.event_router import event_read_router -from openhands.agent_server.models import UpdateSecretsRequest +from openhands.agent_server.models import ConversationInfo, UpdateSecretsRequest +from openhands.sdk import LLM, Agent from openhands.sdk.profiles.agent_profile import LaunchedAgentProfile from openhands.sdk.secret import LookupSecret +from openhands.sdk.utils import utc_now +from openhands.sdk.workspace import LocalWorkspace def test_docker_mode_replaces_local_conversation_execution_routes(tmp_path): @@ -217,6 +220,79 @@ def test_runtime_info_marks_legacy_local_conversation_non_resumable( } +def test_archive_stops_runtime_and_unarchive_stays_cold(tmp_path, monkeypatch): + monkeypatch.setenv("OH_PERSISTENCE_DIR", str(tmp_path / "persistence")) + config = Config( + conversations_path=tmp_path / "conversations", + secret_key=SecretStr("outer-key"), + ) + conversation_id = uuid4() + registry = DockerConversationRegistry(config) + registry.provisioning.create(conversation_id) + registry.stop = AsyncMock() + archived_at = utc_now() + conversation = ConversationInfo( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/workspace"), + agent=Agent(llm=LLM(model="test-model"), tools=[]), + ) + service = AsyncMock() + service.set_conversation_archived.side_effect = [ + conversation.model_copy(update={"archived_at": archived_at}), + conversation, + ] + + app = FastAPI() + app.state.conversation_registry = registry + app.state.conversation_service = service + app.include_router(docker_conversation_router, prefix="/api") + with TestClient(app) as client: + archived = client.post(f"/api/conversations/{conversation_id}/archive") + unarchived = client.post(f"/api/conversations/{conversation_id}/unarchive") + + assert archived.status_code == 200 + assert archived.json()["archived_at"] == archived_at.isoformat().replace( + "+00:00", "Z" + ) + assert unarchived.status_code == 200 + assert unarchived.json()["archived_at"] is None + registry.stop.assert_awaited_once_with(conversation_id) + service.set_conversation_archived.assert_has_awaits( + [ + call(conversation_id, archived=True), + call(conversation_id, archived=False), + ] + ) + + +def test_archived_conversation_cannot_start_a_runtime(tmp_path, monkeypatch): + monkeypatch.setenv("OH_PERSISTENCE_DIR", str(tmp_path / "persistence")) + config = Config( + conversations_path=tmp_path / "conversations", + secret_key=SecretStr("outer-key"), + ) + conversation_id = uuid4() + registry = DockerConversationRegistry(config) + registry.provisioning.create(conversation_id) + directory = registry.conversation_dir(conversation_id) + directory.mkdir(parents=True) + (directory / "meta.json").write_text('{"archived_at":"2026-08-14T00:00:00+00:00"}') + + app = FastAPI() + app.state.conversation_registry = registry + app.state.conversation_service = AsyncMock() + app.include_router(docker_conversation_router, prefix="/api") + with TestClient(app) as client: + reprovision = client.post( + f"/api/conversations/{conversation_id}/runtime/reprovision" + ) + proxy = client.get(f"/api/conversations/{conversation_id}/run") + + assert reprovision.status_code == 409 + assert proxy.status_code == 409 + assert registry.get(conversation_id) is None + + def test_docker_event_history_reads_persistence_without_starting_a_container( tmp_path, monkeypatch ): diff --git a/tests/agent_server/test_conversation_router.py b/tests/agent_server/test_conversation_router.py index 61fbfc69f15..8f12b655dce 100644 --- a/tests/agent_server/test_conversation_router.py +++ b/tests/agent_server/test_conversation_router.py @@ -120,6 +120,48 @@ def sample_start_conversation_request(): ) +@pytest.mark.parametrize( + ("action", "archived"), (("archive", True), ("unarchive", False)) +) +def test_set_conversation_archive_state( + client, mock_conversation_service, sample_conversation_info, action, archived +): + mock_conversation_service.set_conversation_archived.return_value = ( + sample_conversation_info.model_copy( + update={"archived_at": utc_now() if archived else None} + ) + ) + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + + try: + response = client.post( + f"/api/conversations/{sample_conversation_info.id}/{action}" + ) + assert response.status_code == 200 + assert (response.json()["archived_at"] is not None) == archived + mock_conversation_service.set_conversation_archived.assert_awaited_once_with( + sample_conversation_info.id, archived=archived + ) + finally: + client.app.dependency_overrides.clear() + + +def test_archive_conversation_not_found(client, mock_conversation_service): + conversation_id = uuid4() + mock_conversation_service.set_conversation_archived.return_value = None + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + + try: + response = client.post(f"/api/conversations/{conversation_id}/archive") + assert response.status_code == 404 + finally: + client.app.dependency_overrides.clear() + + def test_search_conversations_default_params( client, mock_conversation_service, sample_conversation_info ): @@ -146,7 +188,7 @@ def test_search_conversations_default_params( # Verify service was called with default parameters mock_conversation_service.search_conversations.assert_called_once_with( - None, 100, None, ConversationSortOrder.CREATED_AT_DESC + None, 100, None, ConversationSortOrder.CREATED_AT_DESC, False ) finally: client.app.dependency_overrides.clear() @@ -210,6 +252,7 @@ def test_search_conversations_with_all_params( "limit": 50, "status": ConversationExecutionStatus.IDLE.value, "sort_order": ConversationSortOrder.UPDATED_AT_DESC.value, + "archived": True, }, ) @@ -224,6 +267,7 @@ def test_search_conversations_with_all_params( 50, ConversationExecutionStatus.IDLE, ConversationSortOrder.UPDATED_AT_DESC, + True, ) finally: client.app.dependency_overrides.clear() diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 6b66e9d407a..8864abc587a 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -1223,6 +1223,57 @@ async def test_search_conversations_basic( assert result.items[0].execution_status == ConversationExecutionStatus.IDLE assert result.next_page_id is None + @pytest.mark.asyncio + async def test_archive_filter_is_independent_from_runtime( + self, conversation_service, sample_stored_conversation + ): + conversation_id = sample_stored_conversation.id + state = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=sample_stored_conversation.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=sample_stored_conversation.confirmation_policy, + ) + directory = conversation_service.conversations_dir / conversation_id.hex + directory.mkdir(parents=True) + (directory / "meta.json").write_text( + sample_stored_conversation.model_dump_json() + ) + (directory / "base_state.json").write_text(state.model_dump_json()) + + mock_service = AsyncMock(spec=EventService) + mock_service.stored = sample_stored_conversation + mock_service.get_state.return_value = state + conversation_service._event_services[conversation_id] = mock_service + await conversation_service._reconcile_active_records() + conversation_service._event_services.pop(conversation_id) + + archived = await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + assert archived is not None + assert archived.archived_at is not None + assert conversation_id not in conversation_service._event_services + assert (await conversation_service.search_conversations()).items == [] + assert [ + item.id + for item in ( + await conversation_service.search_conversations(archived=True) + ).items + ] == [conversation_id] + + restored = await conversation_service.set_conversation_archived( + conversation_id, archived=False + ) + assert restored is not None + assert restored.archived_at is None + assert conversation_id not in conversation_service._event_services + assert [ + item.id + for item in (await conversation_service.search_conversations()).items + ] == [conversation_id] + @pytest.mark.asyncio async def test_search_conversations_with_critic_redacts_api_key( self, conversation_service From 6edc0fe5d34455022d544cb7c6a7ebf1d106a4dc Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 21 Sep 2026 11:35:48 +0000 Subject: [PATCH 02/10] fix(agent-server): preserve archived runtime state --- .../agent_server/conversation_service.py | 12 +++++++++ .../agent_server/docker_runtime/routers.py | 11 ++++++-- .../docker_runtime/test_routes.py | 22 ++++++++++++---- tests/agent_server/test_credential_binding.py | 25 +++++++++++++++++++ 4 files changed, 63 insertions(+), 7 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index f3ff3ebf7ee..452045a8087 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1244,6 +1244,18 @@ async def _get_or_load_event_service_locked( return None pending_bindings = self._credential_bindings.get(conversation_id, {}) + if require_runtime_bindings and ( + record.stored.required_runtime_credential_bindings - pending_bindings.keys() + ): + # A cold local runtime can reconstruct canonical bindings (for + # example CODEX_AUTH_JSON) from its injected secrets store. Restore + # those bindings before enforcing admission; _start_event_service + # consumes the same pending map below. + restored = await self._resolve_credential_bindings( + record.stored, agent=agent + ) + pending_bindings = self._credential_bindings.setdefault(conversation_id, {}) + pending_bindings.update(restored) missing_bindings = ( record.stored.required_runtime_credential_bindings - pending_bindings.keys() ) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py index fffcf7a8a61..3f92746e7a3 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -112,6 +112,10 @@ async def start_conversation( body["workspace"] = {"kind": "LocalWorkspace", "working_dir": "/workspace"} registry = get_registry(request) + if _is_archived(registry, conversation_id): + raise HTTPException( + 409, "Conversation is archived; unarchive it before using its runtime" + ) try: prepared, launched = await prepare_start(body, registry.config) identity = registry.provisioning.create(conversation_id, host_workspace) @@ -213,13 +217,16 @@ async def _set_archive_state( conversation_id: UUID, request: Request, *, archived: bool ) -> ConversationInfo: registry = get_registry(request) + if archived: + # The inner service may still hold an older copy of meta.json and write it + # during graceful shutdown. Stop it before persisting the archive marker + # so that stale runtime state cannot undo the transition. + await registry.stop(conversation_id) conversation = await get_conversation_service(request).set_conversation_archived( conversation_id, archived=archived ) if conversation is None: raise HTTPException(404, "Conversation not found") - if archived: - await registry.stop(conversation_id) return conversation.model_copy( update={"runtime_info": registry.runtime_info(conversation_id)} ) diff --git a/tests/agent_server/docker_runtime/test_routes.py b/tests/agent_server/docker_runtime/test_routes.py index e2d9bc35e29..b41eebb03c2 100644 --- a/tests/agent_server/docker_runtime/test_routes.py +++ b/tests/agent_server/docker_runtime/test_routes.py @@ -229,7 +229,8 @@ def test_archive_stops_runtime_and_unarchive_stays_cold(tmp_path, monkeypatch): conversation_id = uuid4() registry = DockerConversationRegistry(config) registry.provisioning.create(conversation_id) - registry.stop = AsyncMock() + transitions = [] + registry.stop = AsyncMock(side_effect=lambda _: transitions.append("stop")) archived_at = utc_now() conversation = ConversationInfo( id=conversation_id, @@ -237,10 +238,16 @@ def test_archive_stops_runtime_and_unarchive_stays_cold(tmp_path, monkeypatch): agent=Agent(llm=LLM(model="test-model"), tools=[]), ) service = AsyncMock() - service.set_conversation_archived.side_effect = [ - conversation.model_copy(update={"archived_at": archived_at}), - conversation, - ] + + async def set_archived(_, *, archived): + transitions.append("archive" if archived else "unarchive") + return ( + conversation.model_copy(update={"archived_at": archived_at}) + if archived + else conversation + ) + + service.set_conversation_archived.side_effect = set_archived app = FastAPI() app.state.conversation_registry = registry @@ -263,6 +270,7 @@ def test_archive_stops_runtime_and_unarchive_stays_cold(tmp_path, monkeypatch): call(conversation_id, archived=False), ] ) + assert transitions == ["stop", "archive", "unarchive"] def test_archived_conversation_cannot_start_a_runtime(tmp_path, monkeypatch): @@ -283,11 +291,15 @@ def test_archived_conversation_cannot_start_a_runtime(tmp_path, monkeypatch): app.state.conversation_service = AsyncMock() app.include_router(docker_conversation_router, prefix="/api") with TestClient(app) as client: + start = client.post( + "/api/conversations", json={"conversation_id": str(conversation_id)} + ) reprovision = client.post( f"/api/conversations/{conversation_id}/runtime/reprovision" ) proxy = client.get(f"/api/conversations/{conversation_id}/run") + assert start.status_code == 409 assert reprovision.status_code == 409 assert proxy.status_code == 409 assert registry.get(conversation_id) is None diff --git a/tests/agent_server/test_credential_binding.py b/tests/agent_server/test_credential_binding.py index b026fe78bd0..fa9fd347a39 100644 --- a/tests/agent_server/test_credential_binding.py +++ b/tests/agent_server/test_credential_binding.py @@ -401,6 +401,31 @@ async def test_direct_start_strips_reserved_conversation_secret(tmp_path) -> Non assert not event_service.stored.required_runtime_credential_bindings +@pytest.mark.asyncio +async def test_local_codex_binding_is_restored_before_cold_admission(tmp_path) -> None: + store = FileSecretsStore(tmp_path / "settings") + store.set_secret("CODEX_AUTH_JSON", "canonical") + conversations_dir = tmp_path / "conversations" + request = StartConversationRequest( + agent=ACPAgent(acp_command=["codex-acp"], acp_server="codex"), + workspace=LocalWorkspace(working_dir=tmp_path / "workspace"), + ) + + async with ConversationService( + conversations_dir=conversations_dir, + secrets_store=store, + ) as service: + info, _ = await service.start_conversation(request) + + async with ConversationService( + conversations_dir=conversations_dir, + secrets_store=store, + ) as restarted: + event_service = await restarted.get_event_service(info.id) + assert event_service is not None + assert "CODEX_AUTH_JSON" in event_service.credential_bindings + + @pytest.mark.asyncio async def test_callback_binding_blocks_cold_hydration_until_reactivation( tmp_path, From 511a277714d1df9c8962982432b7c8fbd52a5c9e Mon Sep 17 00:00:00 2001 From: openhands Date: Mon, 28 Sep 2026 17:04:31 +0000 Subject: [PATCH 03/10] fix(agent-server): keep archived conversations deletable The unconditional archived short-circuit in _get_or_load_event_service_locked() also caught delete_conversation(), which loads through the same helper with require_runtime_bindings=False. Deleting an archived conversation therefore returned False and the local router answered 400, contradicting the README added by this PR ('DELETE /api/conversations/{id} permanently removes the conversation rather than retaining it as archived'). Docker mode hid the bug because its DELETE route bypasses this service method. Add an explicit allow_archived escape hatch, defaulted off so every other caller keeps the archived-is-not-runnable guarantee, and pass it from delete_conversation(). The test fails without the fix (assert False is True) and the new case is added alongside the existing archive test. Co-authored-by: openhands --- .../agent_server/conversation_service.py | 12 +++++- .../agent_server/test_conversation_service.py | 38 +++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 452045a8087..6aae9d68010 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1228,6 +1228,7 @@ async def _get_or_load_event_service_locked( *, require_runtime_bindings: bool = True, agent: AgentBase | None = None, + allow_archived: bool = False, ) -> EventService | None: event_services = self._event_services if event_services is None: @@ -1240,7 +1241,12 @@ async def _get_or_load_event_service_locked( return event_service record = self._conversation_records.get(conversation_id) - if record is None or record.stored.archived_at is not None: + if record is None: + return None + if record.stored.archived_at is not None and not allow_archived: + # An archived conversation retains its history but has no runnable + # runtime. Deletion is the exception: it must still reach the + # archived conversation so it can be permanently removed. return None pending_bindings = self._credential_bindings.get(conversation_id, {}) @@ -1951,6 +1957,10 @@ async def delete_conversation(self, conversation_id: UUID) -> bool: event_service = await self._get_or_load_event_service_locked( conversation_id, require_runtime_bindings=False, + # Deletion is independent of archive state: an archived + # conversation must still be permanently removable rather + # than retained as archived. + allow_archived=True, ) if event_service is None: return False diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 8864abc587a..44e3d3acce1 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -1274,6 +1274,44 @@ async def test_archive_filter_is_independent_from_runtime( for item in (await conversation_service.search_conversations()).items ] == [conversation_id] + @pytest.mark.asyncio + async def test_archived_conversation_is_still_deletable( + self, conversation_service, sample_stored_conversation + ): + """``DELETE`` must reach archived conversations, not 400 on them.""" + conversation_id = sample_stored_conversation.id + state = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=sample_stored_conversation.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=sample_stored_conversation.confirmation_policy, + ) + directory = conversation_service.conversations_dir / conversation_id.hex + directory.mkdir(parents=True) + (directory / "meta.json").write_text( + sample_stored_conversation.model_dump_json() + ) + (directory / "base_state.json").write_text(state.model_dump_json()) + + mock_service = AsyncMock(spec=EventService) + mock_service.stored = sample_stored_conversation + mock_service.get_state.return_value = state + mock_service.conversation_dir = directory + mock_service.is_open.return_value = False + conversation_service._event_services[conversation_id] = mock_service + await conversation_service._reconcile_active_records() + conversation_service._event_services.pop(conversation_id) + + archived = await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + assert archived is not None and archived.archived_at is not None + + assert await conversation_service.delete_conversation(conversation_id) is True + assert conversation_id not in conversation_service._conversation_records + assert not directory.exists() + @pytest.mark.asyncio async def test_search_conversations_with_critic_redacts_api_key( self, conversation_service From c88e9c96b02c6f62fd68058d241618bfe56a1fcd Mon Sep 17 00:00:00 2001 From: openhands Date: Tue, 29 Sep 2026 14:55:32 +0000 Subject: [PATCH 04/10] fix(agent-server): reject archived resume and keep archived history readable Local-runtime archiving diverged from the documented contract in two ways; both now match the Docker runtime. 1. Resume silently succeeded. start_conversation() fell through to _conversation_info() for an archived id and returned 200 while leaving _event_services empty, so a client got success for an operation that did nothing. It now raises ConversationArchivedError, mapped to 409 to match Docker's rejection of the same request. 2. Archived history became unreadable. Read routes depended on get_event_service, whose archived guard returned None and surfaced 404. A new get_read_event_service seam falls back to the persisted event log for archived conversations, mirroring Docker's serves_persisted_event_reads=True. The four read endpoints (search/count/get/batch-get) use it; runtime and write routes keep the strict seam, so archived-is-not-runnable is preserved. Tests: service-level regression tests for the 409 guard, the readable persisted history, and is_conversation_archived for unknown ids; router-level 409 test; dependency-seam tests for archived/live/unknown resolution; read endpoint tests retargeted to the new seam. Co-authored-by: openhands --- .../openhands/agent_server/api.py | 11 ++ .../agent_server/conversation_router.py | 5 + .../agent_server/conversation_service.py | 29 +++++ .../openhands/agent_server/dependencies.py | 37 ++++++ .../openhands/agent_server/event_router.py | 13 ++- .../agent_server/test_conversation_router.py | 41 +++++++ .../agent_server/test_conversation_service.py | 107 ++++++++++++++++++ tests/agent_server/test_event_router.py | 96 ++++++++++++++-- .../test_search_limit_validation.py | 6 +- 9 files changed, 325 insertions(+), 20 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/api.py b/openhands-agent-server/openhands/agent_server/api.py index b0a7ab37ea2..3e3de1d72e1 100644 --- a/openhands-agent-server/openhands/agent_server/api.py +++ b/openhands-agent-server/openhands/agent_server/api.py @@ -34,6 +34,7 @@ conversation_router, ) from openhands.agent_server.conversation_service import ( + ConversationArchivedError, CredentialBindingActivationRequired, get_default_conversation_service, ) @@ -559,6 +560,16 @@ async def _credential_binding_activation_required_handler( }, ) + @api.exception_handler(ConversationArchivedError) + async def _conversation_archived_handler( + _request: Request, + exc: ConversationArchivedError, + ) -> JSONResponse: + # 409 matches the Docker runtime's rejection of the same operation, so a + # client sees one status whether the conversation is host-local or + # container-backed. + return JSONResponse(status_code=409, content={"detail": str(exc)}) + @api.exception_handler(RequestValidationError) async def _validation_exception_handler( request: Request, exc: RequestValidationError diff --git a/openhands-agent-server/openhands/agent_server/conversation_router.py b/openhands-agent-server/openhands/agent_server/conversation_router.py index 880845a6b3d..d57dc034df8 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_router.py +++ b/openhands-agent-server/openhands/agent_server/conversation_router.py @@ -21,6 +21,7 @@ ) from openhands.agent_server.conversation_registry import ConversationRegistry from openhands.agent_server.conversation_service import ( + ConversationArchivedError, ConversationService, InvalidParentConversation, ) @@ -299,6 +300,10 @@ async def start_conversation( """Start a conversation in the local environment.""" try: info, is_new = await conversation_service.start_conversation(request) + except ConversationArchivedError as e: + # Matches the Docker runtime, which rejects start/resume of an archived + # conversation with 409 rather than silently returning a stale record. + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) from e except ProfileNotFound as e: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e except DanglingMcpServerRef as e: diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index 6aae9d68010..afbc113da47 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -306,6 +306,15 @@ class InvalidParentConversation(ValueError): a different workspace.""" +class ConversationArchivedError(ValueError): + """An archived conversation cannot be started or resumed. + + Archiving retains history but releases the runtime; the caller must + unarchive before execution can be restarted. Mirrors the Docker runtime's + 409 rejection so both runtimes reject the same operation identically. + """ + + def _same_workspace(a: LocalWorkspace, b: LocalWorkspace) -> bool: return Path(a.working_dir).resolve() == Path(b.working_dir).resolve() @@ -1514,6 +1523,19 @@ async def _start_conversation( ): async with self._conversation_lifecycle(conversation_id): existing_event_service = self._event_services.get(conversation_id) + current_record = self._conversation_records.get(conversation_id) + if ( + current_record is not None + and current_record.stored.archived_at is not None + ): + # Archiving releases the runtime and retains history. A + # start/resume on an archived id must fail loudly instead of + # returning 200 for a conversation that never came up; the + # caller unarchives first, exactly as Docker requires. + raise ConversationArchivedError( + "Conversation is archived; unarchive it before using " + "its runtime" + ) stored = ( existing_event_service.stored if existing_event_service is not None @@ -2114,6 +2136,13 @@ async def get_persisted_event_service( return None return EventService.for_persisted_events(record.stored, self.conversations_dir) + async def is_conversation_archived(self, conversation_id: UUID) -> bool: + """Whether the catalog marks this conversation as archived.""" + if self._event_services is None: + raise ValueError("inactive_service") + record = self._conversation_records.get(conversation_id) + return record is not None and record.stored.archived_at is not None + async def generate_conversation_title( self, conversation_id: UUID, max_length: int = 50, llm: LLM | None = None ) -> str | None: diff --git a/openhands-agent-server/openhands/agent_server/dependencies.py b/openhands-agent-server/openhands/agent_server/dependencies.py index d1d2a6aece2..5602714ddc2 100644 --- a/openhands-agent-server/openhands/agent_server/dependencies.py +++ b/openhands-agent-server/openhands/agent_server/dependencies.py @@ -110,3 +110,40 @@ async def get_event_service( detail=f"Conversation not found: {conversation_id}", ) return event_service + + +async def get_read_event_service( + conversation_id: UUID, + request: Request, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> EventService: + """Resolve a conversation's readable event history. + + Runtime and write routes use :func:`get_event_service`, which 404s for an + archived conversation because archiving released its runtime. History reads + must survive archiving, so this seam mirrors the Docker runtime's + ``serves_persisted_event_reads`` behaviour: archived conversations are read + from the shared persisted event log rather than 404ing. + """ + registry = getattr(request.app.state, "conversation_registry", None) + if ( + isinstance(registry, ConversationRegistry) + and registry.serves_persisted_event_reads + ): + event_service = await conversation_service.get_persisted_event_service( + conversation_id + ) + else: + event_service = await conversation_service.get_event_service(conversation_id) + if event_service is None and ( + await conversation_service.is_conversation_archived(conversation_id) + ): + event_service = await conversation_service.get_persisted_event_service( + conversation_id + ) + if event_service is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Conversation not found: {conversation_id}", + ) + return event_service diff --git a/openhands-agent-server/openhands/agent_server/event_router.py b/openhands-agent-server/openhands/agent_server/event_router.py index 91d8179c000..5a4311181d2 100644 --- a/openhands-agent-server/openhands/agent_server/event_router.py +++ b/openhands-agent-server/openhands/agent_server/event_router.py @@ -15,7 +15,10 @@ ) from starlette.responses import JSONResponse -from openhands.agent_server.dependencies import get_event_service +from openhands.agent_server.dependencies import ( + get_event_service, + get_read_event_service, +) from openhands.agent_server.event_service import EventService from openhands.agent_server.models import ( ConfirmationResponseRequest, @@ -103,7 +106,7 @@ async def search_conversation_events( datetime | None, Query(title="Filter: event timestamp < this datetime"), ] = None, - event_service: EventService = Depends(get_event_service), + event_service: EventService = Depends(get_read_event_service), ) -> JSONResponse: """Search / List local events""" @@ -165,7 +168,7 @@ async def count_conversation_events( datetime | None, Query(title="Filter: event timestamp < this datetime"), ] = None, - event_service: EventService = Depends(get_event_service), + event_service: EventService = Depends(get_read_event_service), ) -> int: """Count local events matching the given filters""" # Normalize timezone-aware datetimes to server timezone @@ -190,7 +193,7 @@ async def count_conversation_events( ) async def get_conversation_event( event_id: str, - event_service: EventService = Depends(get_event_service), + event_service: EventService = Depends(get_read_event_service), ) -> Event: """Get a local event given an id""" event = await event_service.get_event(event_id) @@ -202,7 +205,7 @@ async def get_conversation_event( @event_read_router.get("") async def batch_get_conversation_events( event_ids: list[str], - event_service: EventService = Depends(get_event_service), + event_service: EventService = Depends(get_read_event_service), ) -> list[Event | None]: """Get a batch of local events given their ids, returning null for any missing item.""" diff --git a/tests/agent_server/test_conversation_router.py b/tests/agent_server/test_conversation_router.py index 8f12b655dce..7fc93419c5b 100644 --- a/tests/agent_server/test_conversation_router.py +++ b/tests/agent_server/test_conversation_router.py @@ -2786,3 +2786,44 @@ def test_runtime_requires_existing_conversation( ) response = client.request(method, f"/api/conversations/{uuid4()}/runtime{suffix}") assert response.status_code == 404 + + +def test_start_conversation_on_archived_id_returns_conflict( + client, mock_conversation_service +): + """Archived start/resume is a 409, matching the Docker runtime. + + Previously the archived record was re-composed into a 200 response for a + conversation that was never started. + """ + from openhands.agent_server.conversation_service import ConversationArchivedError + + mock_conversation_service.start_conversation.side_effect = ( + ConversationArchivedError( + "Conversation is archived; unarchive it before using its runtime" + ) + ) + + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + + request = StartConversationRequest( + agent=Agent( + llm=LLM(model="gpt-4o", api_key=SecretStr("test-key"), usage_id="t"), + tools=[], + ), + workspace=LocalWorkspace(working_dir="/tmp/test"), + conversation_id=uuid4(), + ) + + try: + response = client.post( + "/api/conversations", + json=request.model_dump(mode="json"), + ) + + assert response.status_code == 409 + assert "archived" in response.json()["detail"] + finally: + client.app.dependency_overrides.clear() diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 44e3d3acce1..2728799278e 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -20,6 +20,7 @@ ) from openhands.agent_server.conversation_service import ( AutoTitleSubscriber, + ConversationArchivedError, ConversationService, _compose_conversation_info, _ConversationRecord, @@ -1312,6 +1313,112 @@ async def test_archived_conversation_is_still_deletable( assert conversation_id not in conversation_service._conversation_records assert not directory.exists() + @pytest.mark.asyncio + async def test_start_on_archived_conversation_raises_instead_of_succeeding( + self, conversation_service, sample_stored_conversation, tmp_path + ): + """Resuming an archived id must fail, not return 200 for a dead runtime. + + Before this guard, ``_start_conversation`` composed the record into a + ``ConversationInfo`` and reported success while leaving the archived + conversation unstarted. + """ + conversation_id = sample_stored_conversation.id + state = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=sample_stored_conversation.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=sample_stored_conversation.confirmation_policy, + ) + directory = conversation_service.conversations_dir / conversation_id.hex + directory.mkdir(parents=True) + (directory / "meta.json").write_text( + sample_stored_conversation.model_dump_json() + ) + (directory / "base_state.json").write_text(state.model_dump_json()) + + mock_service = AsyncMock(spec=EventService) + mock_service.stored = sample_stored_conversation + mock_service.get_state.return_value = state + mock_service.is_open.return_value = False + conversation_service._event_services[conversation_id] = mock_service + await conversation_service._reconcile_active_records() + conversation_service._event_services.pop(conversation_id) + + assert ( + await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + is not None + ) + + workspace_dir = tmp_path / "resume-workspace" + workspace_dir.mkdir() + request = StartConversationRequest( + agent=_sample_agent(), + workspace=LocalWorkspace(working_dir=str(workspace_dir)), + confirmation_policy=NeverConfirm(), + conversation_id=conversation_id, + ) + + with pytest.raises(ConversationArchivedError): + await conversation_service.start_conversation(request) + + # The failed start must not leave a runtime behind. + assert conversation_id not in conversation_service._event_services + assert len(conversation_service._conversation_records) == 1 + + @pytest.mark.asyncio + async def test_archived_history_remains_readable( + self, conversation_service, sample_stored_conversation, tmp_path + ): + """Archiving releases the runtime but must not hide event history.""" + conversation_id = sample_stored_conversation.id + state = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=sample_stored_conversation.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=sample_stored_conversation.confirmation_policy, + ) + directory = conversation_service.conversations_dir / conversation_id.hex + directory.mkdir(parents=True) + (directory / "meta.json").write_text( + sample_stored_conversation.model_dump_json() + ) + (directory / "base_state.json").write_text(state.model_dump_json()) + + mock_service = AsyncMock(spec=EventService) + mock_service.stored = sample_stored_conversation + mock_service.get_state.return_value = state + mock_service.is_open.return_value = False + conversation_service._event_services[conversation_id] = mock_service + await conversation_service._reconcile_active_records() + conversation_service._event_services.pop(conversation_id) + + assert ( + await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + is not None + ) + + assert await conversation_service.is_conversation_archived(conversation_id) + # No runtime is handed out for an archived conversation... + assert await conversation_service.get_event_service(conversation_id) is None + # ...but the persisted log is still openable for reads. + persisted = await conversation_service.get_persisted_event_service( + conversation_id + ) + assert persisted is not None + + @pytest.mark.asyncio + async def test_is_conversation_archived_is_false_for_unknown_id( + self, conversation_service + ): + assert not await conversation_service.is_conversation_archived(uuid4()) + @pytest.mark.asyncio async def test_search_conversations_with_critic_redacts_api_key( self, conversation_service diff --git a/tests/agent_server/test_event_router.py b/tests/agent_server/test_event_router.py index 503947a9dc0..0a64f0c29d6 100644 --- a/tests/agent_server/test_event_router.py +++ b/tests/agent_server/test_event_router.py @@ -7,10 +7,13 @@ from uuid import uuid4 import pytest -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient -from openhands.agent_server.dependencies import get_event_service +from openhands.agent_server.dependencies import ( + get_event_service, + get_read_event_service, +) from openhands.agent_server.event_router import ( event_router, normalize_datetime_to_server_timezone, @@ -363,7 +366,9 @@ async def test_search_events_with_naive_datetime( ): """Test search events with naive datetime (no timezone).""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -400,7 +405,9 @@ async def test_search_events_with_timezone_aware_datetime( ): """Test search events with timezone-aware datetime.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -437,7 +444,9 @@ async def test_search_events_with_timezone_range( """Test search events with both timestamp filters using timezone-aware datetimes.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -474,7 +483,9 @@ async def test_count_events_with_timezone_aware_datetime( ): """Test count events with timezone-aware datetime.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the count_events method to return a sample result @@ -507,7 +518,9 @@ async def test_count_events_with_source_filter( ): """Test count events with source filter.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the count_events method to return a sample result @@ -541,7 +554,9 @@ async def test_search_events_timezone_normalization_consistency( """Test that different timezone representations of the same moment normalize consistently.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -596,7 +611,9 @@ async def test_search_events_with_source_filter( ): """Test search events with source filter.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -631,7 +648,9 @@ async def test_search_events_with_multiple_filters( ): """Test search events with multiple filters including source.""" # Override the dependency to return our mock - client.app.dependency_overrides[get_event_service] = lambda: mock_event_service + client.app.dependency_overrides[get_read_event_service] = ( + lambda: mock_event_service + ) try: # Mock the search_events method to return a sample result @@ -717,7 +736,7 @@ async def test_search_events_with_source_filter_real_events( conversation._state = state event_service._conversation = conversation - client.app.dependency_overrides[get_event_service] = lambda: event_service + client.app.dependency_overrides[get_read_event_service] = lambda: event_service try: # Test filtering by source="user" - should return 2 events @@ -790,7 +809,7 @@ async def test_search_events_with_body_filter_real_events( conversation._state = state event_service._conversation = conversation - client.app.dependency_overrides[get_event_service] = lambda: event_service + client.app.dependency_overrides[get_read_event_service] = lambda: event_service try: # Test filtering by body="hello" (case-insensitive) - should return 2 events @@ -808,3 +827,56 @@ async def test_search_events_with_body_filter_real_events( finally: client.app.dependency_overrides.clear() + + +class TestReadEventServiceSeam: + """The read seam must serve archived history without handing out a runtime. + + ``get_event_service`` 404s an archived conversation because archiving + releases its runtime; ``get_read_event_service`` falls back to the persisted + event log so history stays readable, mirroring the Docker runtime's + ``serves_persisted_event_reads`` behaviour. + """ + + @pytest.mark.asyncio + async def test_archived_conversation_reads_from_persisted_log(self): + conversation_id = uuid4() + persisted = MagicMock(spec=EventService) + service = AsyncMock() + service.get_event_service.return_value = None + service.is_conversation_archived.return_value = True + service.get_persisted_event_service.return_value = persisted + request = MagicMock() + request.app.state = MagicMock(spec=[]) + + resolved = await get_read_event_service(conversation_id, request, service) + + assert resolved is persisted + service.get_persisted_event_service.assert_awaited_once_with(conversation_id) + + @pytest.mark.asyncio + async def test_live_conversation_reads_from_runtime(self): + conversation_id = uuid4() + live = MagicMock(spec=EventService) + service = AsyncMock() + service.get_event_service.return_value = live + request = MagicMock() + request.app.state = MagicMock(spec=[]) + + resolved = await get_read_event_service(conversation_id, request, service) + + assert resolved is live + service.get_persisted_event_service.assert_not_awaited() + + @pytest.mark.asyncio + async def test_unknown_conversation_still_404s(self): + service = AsyncMock() + service.get_event_service.return_value = None + service.is_conversation_archived.return_value = False + request = MagicMock() + request.app.state = MagicMock(spec=[]) + + with pytest.raises(HTTPException) as exc_info: + await get_read_event_service(uuid4(), request, service) + + assert exc_info.value.status_code == 404 diff --git a/tests/agent_server/test_search_limit_validation.py b/tests/agent_server/test_search_limit_validation.py index a857341c6d5..400adc38755 100644 --- a/tests/agent_server/test_search_limit_validation.py +++ b/tests/agent_server/test_search_limit_validation.py @@ -16,7 +16,7 @@ from openhands.agent_server.bash_router import bash_router from openhands.agent_server.conversation_router import conversation_router -from openhands.agent_server.dependencies import get_event_service +from openhands.agent_server.dependencies import get_read_event_service from openhands.agent_server.event_router import event_router from openhands.agent_server.event_service import EventService from openhands.agent_server.file_router import file_discovery_router @@ -58,7 +58,7 @@ def test_search_events_rejects_out_of_range_limit(limit): app.include_router(event_router, prefix="/api") service = AsyncMock(spec=EventService) service.search_events = AsyncMock(return_value={"items": [], "next_page_id": None}) - app.dependency_overrides[get_event_service] = lambda: service + app.dependency_overrides[get_read_event_service] = lambda: service with TestClient(app) as client: response = client.get( @@ -74,7 +74,7 @@ def test_search_events_accepts_the_maximum_limit(): app.include_router(event_router, prefix="/api") service = AsyncMock(spec=EventService) service.search_events = AsyncMock(return_value={"items": [], "next_page_id": None}) - app.dependency_overrides[get_event_service] = lambda: service + app.dependency_overrides[get_read_event_service] = lambda: service with TestClient(app) as client: response = client.get( From 768b6e7e827fdd47e893ea55f43f497f356d441f Mon Sep 17 00:00:00 2001 From: openhands Date: Wed, 30 Sep 2026 10:49:16 +0000 Subject: [PATCH 05/10] fix(agent-server): keep runtime metadata when archiving; filter count by archive Archiving persisted the outer catalog's copy of StoredConversation. In Docker mode that copy can predate what the inner runtime wrote (for example a title set on the runtime, or its final shutdown write), and in local mode an open EventService may have replaced its stored metadata without syncing the catalog record. Either way the archive marker overwrote newer metadata. Reload persisted metadata after stopping the Docker runtime, and base the local update on the live service's copy. GET /api/conversations/count now accepts the same archived filter and default as search, so a count shown next to the default listing no longer includes archived conversations. Co-authored-by: openhands --- .../src/__tests__/api-clients.test.ts | 4 +- .../src/client/conversation-client.ts | 2 +- .../src/conversation/conversation-manager.ts | 8 ++- .../openhands/agent_server/README.md | 2 +- .../agent_server/conversation_router.py | 5 +- .../agent_server/conversation_service.py | 47 ++++++++----- .../agent_server/docker_runtime/routers.py | 7 +- .../docker_runtime/test_routes.py | 66 ++++++++++++++++++- .../agent_server/test_conversation_router.py | 24 ++++++- .../agent_server/test_conversation_service.py | 48 ++++++++++++++ 10 files changed, 183 insertions(+), 30 deletions(-) diff --git a/clients/typescript/src/__tests__/api-clients.test.ts b/clients/typescript/src/__tests__/api-clients.test.ts index 2641843502d..c4f2dbabdc4 100644 --- a/clients/typescript/src/__tests__/api-clients.test.ts +++ b/clients/typescript/src/__tests__/api-clients.test.ts @@ -2369,7 +2369,7 @@ describe('Auxiliary API clients', () => { const client = new ConversationClient({ host: 'http://example.com' }); await expect( - client.countConversations({ status: ConversationExecutionStatus.IDLE }) + client.countConversations({ status: ConversationExecutionStatus.IDLE, archived: true }) ).resolves.toBe(2); await expect(client.searchEvents('c1', { kind: 'MessageEvent', limit: 5 })).resolves.toEqual({ items: [event], @@ -2388,7 +2388,7 @@ describe('Auxiliary API clients', () => { expect(global.fetch).toHaveBeenNthCalledWith( 1, - 'http://example.com/api/conversations/count?status=idle', + 'http://example.com/api/conversations/count?status=idle&archived=true', expect.objectContaining({ method: 'GET' }) ); expect(global.fetch).toHaveBeenNthCalledWith( diff --git a/clients/typescript/src/client/conversation-client.ts b/clients/typescript/src/client/conversation-client.ts index f1029656af1..a56af61b8eb 100644 --- a/clients/typescript/src/client/conversation-client.ts +++ b/clients/typescript/src/client/conversation-client.ts @@ -85,7 +85,7 @@ export class ConversationClient { } async countConversations( - options: { status?: ConversationExecutionStatus } = {} + options: { status?: ConversationExecutionStatus; archived?: boolean } = {} ): Promise { const response = await this.client.get('/api/conversations/count', { params: options as Record, diff --git a/clients/typescript/src/conversation/conversation-manager.ts b/clients/typescript/src/conversation/conversation-manager.ts index 8ce95d6eae7..b270e775892 100644 --- a/clients/typescript/src/conversation/conversation-manager.ts +++ b/clients/typescript/src/conversation/conversation-manager.ts @@ -41,7 +41,9 @@ export class ACPConversationNamespace { return this.manager.searchACPConversations(options); } - countConversations(options: { status?: ConversationExecutionStatus } = {}): Promise { + countConversations( + options: { status?: ConversationExecutionStatus; archived?: boolean } = {} + ): Promise { return this.manager.countACPConversations(options); } @@ -148,7 +150,7 @@ export class ConversationManager { * Count conversations matching the provided filters. */ async countConversations( - options: { status?: ConversationExecutionStatus } = {} + options: { status?: ConversationExecutionStatus; archived?: boolean } = {} ): Promise { const response = await this.client.get('/api/conversations/count', { params: options as Record, @@ -275,7 +277,7 @@ export class ConversationManager { * Count ACP-capable conversations. */ async countACPConversations( - options: { status?: ConversationExecutionStatus } = {} + options: { status?: ConversationExecutionStatus; archived?: boolean } = {} ): Promise { const response = await this.client.get('/api/conversations/count', { params: options, diff --git a/openhands-agent-server/openhands/agent_server/README.md b/openhands-agent-server/openhands/agent_server/README.md index 1e074077734..dcb6840fdef 100644 --- a/openhands-agent-server/openhands/agent_server/README.md +++ b/openhands-agent-server/openhands/agent_server/README.md @@ -20,7 +20,7 @@ Conversation metadata separates retained history from runtime availability: - `runtime_status` reports whether execution infrastructure is available; it does not imply archive state. - `POST /api/conversations/{id}/archive` sets `archived_at`, hides the conversation from the default search, and stops only runtime infrastructure owned by this server. It preserves history and workspace files. - `POST /api/conversations/{id}/unarchive` clears `archived_at` without provisioning infrastructure or starting agent execution. -- `GET /api/conversations/search?archived=true` returns archived conversations; the default `archived=false` returns active conversations. +- `GET /api/conversations/search?archived=true` returns archived conversations; the default `archived=false` returns active conversations. `GET /api/conversations/count` accepts the same `archived` filter and default. - `DELETE /api/conversations/{id}` permanently removes the conversation rather than retaining it as archived. ## Quick Start diff --git a/openhands-agent-server/openhands/agent_server/conversation_router.py b/openhands-agent-server/openhands/agent_server/conversation_router.py index d57dc034df8..511fae4bc01 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_router.py +++ b/openhands-agent-server/openhands/agent_server/conversation_router.py @@ -157,10 +157,13 @@ async def count_conversations( ConversationExecutionStatus | None, Query(title="Optional filter by conversation execution status"), ] = None, + archived: Annotated[ + bool, Query(title="Count archived rather than active conversations") + ] = False, conversation_service: ConversationService = Depends(get_conversation_service), ) -> int: """Count conversations matching the given filters""" - count = await conversation_service.count_conversations(status) + count = await conversation_service.count_conversations(status, archived) return count diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index afbc113da47..527dc0fd0cc 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1420,26 +1420,31 @@ async def _search_conversations( async def count_conversations( self, execution_status: ConversationExecutionStatus | None = None, + archived: bool = False, ) -> int: - return await self._count_conversations(execution_status=execution_status) + return await self._count_conversations( + execution_status=execution_status, archived=archived + ) async def _count_conversations( self, execution_status: ConversationExecutionStatus | None, + archived: bool, ) -> int: """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) - - await self._refresh_execution_statuses() + if execution_status is not None: + await self._refresh_execution_statuses() return sum( 1 for record in self._conversation_records.values() - if record.execution_status == execution_status + if (record.stored.archived_at is not None) == archived + and ( + execution_status is None or record.execution_status == execution_status + ) ) async def batch_get_conversations( @@ -2094,20 +2099,28 @@ async def set_conversation_archived( if record is None: return None - now = utc_now() - archived_at = (record.stored.archived_at or now) if archived else None - record.stored = record.stored.model_copy( - update={"archived_at": archived_at, "updated_at": now} - ) - record.cached_info = None - event_service = ( self._event_services.get(conversation_id) if self._event_services is not None else None ) - if event_service is not None and event_service.is_open(): - event_service.stored = record.stored + live_service = ( + event_service + if event_service is not None and event_service.is_open() + else None + ) + # A live service may have replaced its stored metadata without + # syncing the catalog record; it is authoritative while open. + current = live_service.stored if live_service is not None else record.stored + now = utc_now() + archived_at = (current.archived_at or now) if archived else None + record.stored = current.model_copy( + update={"archived_at": archived_at, "updated_at": now} + ) + record.cached_info = None + + if live_service is not None: + live_service.stored = record.stored metadata = record.stored.model_dump_json( context={"cipher": self._cipher_for(conversation_id)} @@ -2115,8 +2128,8 @@ async def set_conversation_archived( meta_file = self.conversations_dir / conversation_id.hex / "meta.json" await asyncio.to_thread(meta_file.write_text, metadata) - if archived and event_service is not None and event_service.is_open(): - await event_service.close() + if archived and live_service is not None: + await live_service.close() if self._event_services is not None: self._event_services.pop(conversation_id, None) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py index 953a7e6b72e..e1b10066d9b 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -250,12 +250,15 @@ async def _set_archive_state( conversation_id: UUID, request: Request, *, archived: bool ) -> ConversationInfo: registry = get_registry(request) + conversation_service = get_conversation_service(request) if archived: # The inner service may still hold an older copy of meta.json and write it # during graceful shutdown. Stop it before persisting the archive marker - # so that stale runtime state cannot undo the transition. + # so that stale runtime state cannot undo the transition, then reload + # what it wrote so the marker does not overwrite its final metadata. await registry.stop(conversation_id) - conversation = await get_conversation_service(request).set_conversation_archived( + await conversation_service.refresh_persisted_conversation(conversation_id) + conversation = await conversation_service.set_conversation_archived( conversation_id, archived=archived ) if conversation is None: diff --git a/tests/agent_server/docker_runtime/test_routes.py b/tests/agent_server/docker_runtime/test_routes.py index b41eebb03c2..a7f21e21cfa 100644 --- a/tests/agent_server/docker_runtime/test_routes.py +++ b/tests/agent_server/docker_runtime/test_routes.py @@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, call from uuid import uuid4 +import httpx import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -14,6 +15,7 @@ from openhands.agent_server.api import create_app from openhands.agent_server.config import Config +from openhands.agent_server.conversation_service import ConversationService from openhands.agent_server.docker_runtime.provisioning import RuntimeProvisioningStore from openhands.agent_server.docker_runtime.registry import DockerConversationRegistry from openhands.agent_server.docker_runtime.routers import ( @@ -22,10 +24,16 @@ proxy_conversation, ) from openhands.agent_server.event_router import event_read_router -from openhands.agent_server.models import ConversationInfo, UpdateSecretsRequest +from openhands.agent_server.models import ( + ConversationInfo, + StoredConversation, + UpdateSecretsRequest, +) from openhands.sdk import LLM, Agent +from openhands.sdk.conversation.state import ConversationState from openhands.sdk.profiles.agent_profile import LaunchedAgentProfile from openhands.sdk.secret import LookupSecret +from openhands.sdk.security.confirmation_policy import NeverConfirm from openhands.sdk.utils import utc_now from openhands.sdk.workspace import LocalWorkspace @@ -273,6 +281,62 @@ async def set_archived(_, *, archived): assert transitions == ["stop", "archive", "unarchive"] +@pytest.mark.asyncio +async def test_archive_keeps_metadata_written_by_stopping_runtime( + tmp_path, monkeypatch +): + monkeypatch.setenv("OH_PERSISTENCE_DIR", str(tmp_path / "persistence")) + config = Config( + conversations_path=tmp_path / "conversations", + secret_key=SecretStr("outer-key"), + ) + conversation_id = uuid4() + registry = DockerConversationRegistry(config) + registry.provisioning.create(conversation_id) + directory = registry.conversation_dir(conversation_id) + directory.mkdir(parents=True) + stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir="/workspace"), + confirmation_policy=NeverConfirm(), + ) + (directory / "meta.json").write_text(stored.model_dump_json()) + (directory / "base_state.json").write_text( + ConversationState( + id=conversation_id, + agent=Agent(llm=LLM(model="test-model"), tools=[]), + workspace=stored.workspace, + confirmation_policy=stored.confirmation_policy, + ).model_dump_json() + ) + + async def inner_runtime_shutdown(_): + runtime_stored = stored.model_copy(update={"title": "Title set by runtime"}) + (directory / "meta.json").write_text(runtime_stored.model_dump_json()) + + registry.stop = AsyncMock(side_effect=inner_runtime_shutdown) + + app = FastAPI() + app.state.conversation_registry = registry + app.include_router(docker_conversation_router, prefix="/api") + async with ConversationService( + conversations_dir=config.conversations_path + ) as service: + app.state.conversation_service = service + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + f"/api/conversations/{conversation_id}/archive" + ) + + assert response.status_code == 200 + assert response.json()["title"] == "Title set by runtime" + persisted = json.loads((directory / "meta.json").read_text()) + assert persisted["title"] == "Title set by runtime" + assert persisted["archived_at"] is not None + + def test_archived_conversation_cannot_start_a_runtime(tmp_path, monkeypatch): monkeypatch.setenv("OH_PERSISTENCE_DIR", str(tmp_path / "persistence")) config = Config( diff --git a/tests/agent_server/test_conversation_router.py b/tests/agent_server/test_conversation_router.py index 7fc93419c5b..9519048efd6 100644 --- a/tests/agent_server/test_conversation_router.py +++ b/tests/agent_server/test_conversation_router.py @@ -338,7 +338,27 @@ def test_count_conversations_no_filter(client, mock_conversation_service): assert response.json() == 5 # Verify service was called with no status filter - mock_conversation_service.count_conversations.assert_called_once_with(None) + mock_conversation_service.count_conversations.assert_called_once_with( + None, False + ) + finally: + client.app.dependency_overrides.clear() + + +def test_count_conversations_with_archived_filter(client, mock_conversation_service): + mock_conversation_service.count_conversations.return_value = 2 + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + + try: + response = client.get("/api/conversations/count", params={"archived": "true"}) + + assert response.status_code == 200 + assert response.json() == 2 + mock_conversation_service.count_conversations.assert_called_once_with( + None, True + ) finally: client.app.dependency_overrides.clear() @@ -364,7 +384,7 @@ def test_count_conversations_with_status_filter(client, mock_conversation_servic # Verify service was called with status filter mock_conversation_service.count_conversations.assert_called_once_with( - ConversationExecutionStatus.RUNNING + ConversationExecutionStatus.RUNNING, False ) finally: client.app.dependency_overrides.clear() diff --git a/tests/agent_server/test_conversation_service.py b/tests/agent_server/test_conversation_service.py index 2728799278e..55a4e1f3775 100644 --- a/tests/agent_server/test_conversation_service.py +++ b/tests/agent_server/test_conversation_service.py @@ -1263,6 +1263,14 @@ async def test_archive_filter_is_independent_from_runtime( await conversation_service.search_conversations(archived=True) ).items ] == [conversation_id] + assert await conversation_service.count_conversations() == 0 + assert await conversation_service.count_conversations(archived=True) == 1 + assert ( + await conversation_service.count_conversations( + ConversationExecutionStatus.IDLE, archived=True + ) + == 1 + ) restored = await conversation_service.set_conversation_archived( conversation_id, archived=False @@ -1274,6 +1282,46 @@ async def test_archive_filter_is_independent_from_runtime( item.id for item in (await conversation_service.search_conversations()).items ] == [conversation_id] + assert await conversation_service.count_conversations() == 1 + assert await conversation_service.count_conversations(archived=True) == 0 + + @pytest.mark.asyncio + async def test_archive_keeps_metadata_replaced_by_live_service( + self, conversation_service, sample_stored_conversation + ): + conversation_id = sample_stored_conversation.id + state = ConversationState( + id=conversation_id, + agent=_sample_agent(), + workspace=sample_stored_conversation.workspace, + execution_status=ConversationExecutionStatus.IDLE, + confirmation_policy=sample_stored_conversation.confirmation_policy, + ) + directory = conversation_service.conversations_dir / conversation_id.hex + directory.mkdir(parents=True) + (directory / "base_state.json").write_text(state.model_dump_json()) + live_service = AsyncMock(spec=EventService) + live_service.stored = sample_stored_conversation + live_service.get_state.return_value = state + live_service.is_open.return_value = True + conversation_service._event_services[conversation_id] = live_service + await conversation_service._reconcile_active_records() + # e.g. switch_acp_model or credential scrubbing replace the live copy + # without syncing the catalog record. + live_service.stored = sample_stored_conversation.model_copy( + update={"title": "Live title"} + ) + + archived = await conversation_service.set_conversation_archived( + conversation_id, archived=True + ) + + assert archived is not None and archived.title == "Live title" + assert live_service.stored.title == "Live title" + assert live_service.stored.archived_at is not None + live_service.close.assert_awaited_once() + persisted = json.loads((directory / "meta.json").read_text()) + assert persisted["title"] == "Live title" @pytest.mark.asyncio async def test_archived_conversation_is_still_deletable( From 753dfa96115107bb1f1213f74c2daac91488cea9 Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 3 Oct 2026 20:26:09 +0000 Subject: [PATCH 06/10] fix(agent-server): resolve symlinked mounts before the archive check The merge of main into this branch placed the archived-conversation check ahead of main's conversation_dir() guard, so a symlinked runtime mount raised ValueError out of the request handler and returned 500 instead of the 422 that test_a_symlinked_conversation_dir_is_rejected asserts. Resolve the conversation directory (mapping the guard violation to 422) first, and make _is_archived treat an unresolvable directory as not archived. Co-authored-by: openhands --- .../agent_server/docker_runtime/routers.py | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py index 22adce0114f..4bbbc7ed4ec 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -58,7 +58,12 @@ def get_registry(request: Request) -> DockerConversationRegistry: def _is_archived(registry: DockerConversationRegistry, conversation_id: UUID) -> bool: - metadata_path = registry.conversation_dir(conversation_id) / "meta.json" + try: + metadata_path = registry.conversation_dir(conversation_id) / "meta.json" + except ValueError: + # Symlinked runtime mount: not archived, and the caller reports the + # guard violation itself. + return False if not metadata_path.is_file(): return False return json.loads(metadata_path.read_text()).get("archived_at") is not None @@ -154,18 +159,19 @@ async def start_conversation( body["workspace"] = {"kind": "LocalWorkspace", "working_dir": "/workspace"} registry = get_registry(request) - if _is_archived(registry, conversation_id): - raise HTTPException( - 409, "Conversation is archived; unarchive it before using its runtime" - ) - # Only a conversation the container already created is left alone on - # failure; a manifest from a failed earlier start is retried like a new one. + # Resolve the directory before the archive check: a symlinked mount must + # surface as the 422 the guard in conversation_dir raises, not a 500 from + # reading metadata through it. try: existing = ( registry.conversation_dir(conversation_id).joinpath("meta.json").is_file() ) except ValueError as exc: raise HTTPException(422, str(exc)) from exc + if _is_archived(registry, conversation_id): + raise HTTPException( + 409, "Conversation is archived; unarchive it before using its runtime" + ) try: payload, launched = await _prepare_forward(body, registry, existing=existing) identity = registry.provisioning.create(conversation_id, host_workspace) From 968936198370aea051934d4cc810495db3ca844f Mon Sep 17 00:00:00 2001 From: allhands-bot Date: Sat, 3 Oct 2026 21:29:36 +0000 Subject: [PATCH 07/10] chore: Remove PR-only artifacts [automated] --- .pr/design.html | 111 ------------------------------------------------ 1 file changed, 111 deletions(-) delete mode 100644 .pr/design.html diff --git a/.pr/design.html b/.pr/design.html deleted file mode 100644 index c16589e9d27..00000000000 --- a/.pr/design.html +++ /dev/null @@ -1,111 +0,0 @@ - - - - -Automation Observability Context Propagation - - - -
-

Automation observability context propagation

-

This PR makes SDK and agent-server conversation creation carry automation metadata and Laminar parent span context, including direct API-created conversations.

-
-
- -
-
-

What changed

-

Decision: observability context is now product-neutral data that can be supplied through generic OpenHands env variables, merged with explicit SDK metadata, serialized into HTTP headers, and rehydrated by the agent server before a conversation starts.

-
★ The important behavior change is that POST /api/conversations can now join an automation trace even when the custom automation bypasses the SDK and calls the agent-server API directly. See openhands-agent-server/openhands/agent_server/conversation_router.py:83 and :331.
-

Compatibility: additive. Existing callers may ignore the new field and headers. Existing explicit request body values continue to win over propagated header defaults.

-
- -
-

API before → after

-
-
-

Before

-
StartConversationRequest:
-  observability_metadata
-  observability_tags
-  observability_span_name
-
-POST /api/conversations:
-  body metadata only
-
-Remote workspace calls:
-  X-Session-API-Key only
-

Parent span context and searchable metadata stopped at the process boundary unless each script manually copied them into SDK arguments.

-
-
-

After

-
StartConversationRequest:
-  + observability_parent_span_context
-
-POST /api/conversations headers:
-  X-OpenHands-Observability-Metadata
-  X-OpenHands-Observability-Tags
-  X-OpenHands-Observability-Span-Name
-  X-OpenHands-Observability-Parent-Span-Context
-
-Remote workspace calls:
-  + generic OpenHands observability headers from env
-

The new request field is defined at openhands-sdk/openhands/sdk/conversation/request.py:261; generic headers are emitted from observability_headers_from_env(). Automation services can still make automation.id searchable by placing it in OPENHANDS_OBSERVABILITY_METADATA.

-
-
-
- -
-

Flow before → after

- - -Before: trace fragments -automation.run.dispatchLaminar trace A -sandbox envIDs, no parent usage -Conversation / LLMLaminar trace B - -Manual-dispatch metadata exists, but spans are split by trace_id. - -After: one propagated parent context -automation.run.dispatchserializes parent -laminar.py env helpersmerge metadata + parent -SDK / direct APIheaders or request field -agent-serverconversation child - -Searchable metadata such as automation IDs, trigger type, manual dispatch source, conversation IDs, and parent context can travel together without the SDK knowing automation-specific env names. - -
- -
-

Changed file groups

-
    -
  • core helpers openhands-sdk/openhands/sdk/observability/laminar.py adds generic OpenHands metadata parsing, parent context lookup, header generation, and parent-aware root span startup.
    • -
  • conversation API conversation/request.py, conversation/base.py, local_conversation.py, remote_conversation.py, and conversation.py thread observability_parent_span_context through construction and span creation.
    • -
  • remote calls workspace/remote/remote_workspace_mixin.py:77 merges generic OpenHands observability headers into workspace/API requests.
    • -
  • agent server conversation_router.py:83 validates and merges observability headers; event_service.py forwards stored parent context into server-side local conversations.
    • -
  • OpenAI routes openai/router.py:82 accepts the same parent context header for OpenAI-compatible entrypoints.
    • -
  • tests Router and remote workspace tests cover direct API header handling and env-driven header propagation.
    • -
-
- -
-

Risks and validation

-
    -
  • Header trust: the agent server validates span names, tags, and metadata with existing typed adapters, and request body metadata wins over header defaults.
    • -
  • Compatibility: all new fields are optional. Existing SDK and API callers retain current behavior.
    • -
  • Validation run: targeted pytest and ruff checks passed; see PR description for exact commands.
    • -
-

This page was created as a temporary PR artifact under .pr/ and is expected to be removed by the repository's PR artifact cleanup workflow.

-
-
-
- - From d52774e52fa46e90282782c82df55e7804fc68dd Mon Sep 17 00:00:00 2001 From: Graham Neubig Date: Sat, 3 Oct 2026 18:49:49 -0400 Subject: [PATCH 08/10] feat(ts-client): normalize conversation lifecycle (#5003) Co-authored-by: openhands --- .../__tests__/conversation-lifecycle.test.ts | 228 ++++++++++++++++++ clients/typescript/src/client/cloud-client.ts | 4 + clients/typescript/src/clients.ts | 9 + clients/typescript/src/index.ts | 11 + .../src/models/conversation-lifecycle.ts | 113 +++++++++ 5 files changed, 365 insertions(+) create mode 100644 clients/typescript/src/__tests__/conversation-lifecycle.test.ts create mode 100644 clients/typescript/src/models/conversation-lifecycle.ts diff --git a/clients/typescript/src/__tests__/conversation-lifecycle.test.ts b/clients/typescript/src/__tests__/conversation-lifecycle.test.ts new file mode 100644 index 00000000000..a90e88c381f --- /dev/null +++ b/clients/typescript/src/__tests__/conversation-lifecycle.test.ts @@ -0,0 +1,228 @@ +import { + normalizeCloudConversationLifecycle, + normalizeConversationLifecycle, +} from '../models/conversation-lifecycle'; +import { ConversationExecutionStatus } from '../types/base'; +import type { ConversationInfo } from '../models/conversation'; + +describe('normalizeConversationLifecycle', () => { + it('maps explicit archived Agent Server lifecycle fields directly', () => { + expect( + normalizeConversationLifecycle({ + archived_at: '2026-09-12T15:00:00Z', + runtime_info: { + runtime_status: 'missing', + can_resume: true, + runtime_error: null, + }, + execution_status: ConversationExecutionStatus.PAUSED, + }) + ).toEqual({ + isArchived: true, + archivedAt: '2026-09-12T15:00:00Z', + runtimeStatus: 'missing', + executionStatus: ConversationExecutionStatus.PAUSED, + canResume: true, + }); + }); + + it('keeps missing runtimes independent from explicit unarchived state', () => { + expect( + normalizeConversationLifecycle({ + archived_at: null, + runtime_info: { + runtime_status: 'missing', + can_resume: false, + runtime_error: null, + }, + execution_status: ConversationExecutionStatus.ERROR, + }) + ).toEqual({ + isArchived: false, + archivedAt: null, + runtimeStatus: 'missing', + executionStatus: ConversationExecutionStatus.ERROR, + canResume: false, + }); + }); + + it('accepts the public conversation response type', () => { + const conversation = { + archived_at: null, + runtime_info: { + runtime_status: 'available', + can_resume: false, + runtime_error: null, + }, + execution_status: ConversationExecutionStatus.FINISHED, + } as ConversationInfo; + + expect(normalizeConversationLifecycle(conversation)).toMatchObject({ + isArchived: false, + runtimeStatus: 'available', + executionStatus: ConversationExecutionStatus.FINISHED, + canResume: false, + }); + }); + + it('defaults to local runtime semantics when runtime_info is omitted', () => { + // `create`, `fork`, and `navigate` return a bare `ConversationInfo`, so + // `runtime_info` is absent even on a fully current Agent Server. Callers + // must not need an availability check to read these responses. + expect( + normalizeConversationLifecycle({ + archived_at: null, + execution_status: ConversationExecutionStatus.FINISHED, + }) + ).toEqual({ + isArchived: false, + archivedAt: null, + runtimeStatus: 'available', + executionStatus: ConversationExecutionStatus.FINISHED, + canResume: true, + }); + }); + + it('defaults to local runtime semantics when runtime_info is null', () => { + expect( + normalizeConversationLifecycle({ + archived_at: null, + runtime_info: null, + execution_status: ConversationExecutionStatus.IDLE, + }) + ).toEqual({ + isArchived: false, + archivedAt: null, + runtimeStatus: 'available', + executionStatus: ConversationExecutionStatus.IDLE, + canResume: true, + }); + }); + + it('preserves an archived marker on a response without runtime_info', () => { + expect( + normalizeConversationLifecycle({ + archived_at: '2026-09-12T15:00:00Z', + execution_status: ConversationExecutionStatus.PAUSED, + }) + ).toMatchObject({ + isArchived: true, + archivedAt: '2026-09-12T15:00:00Z', + canResume: true, + }); + }); + + it('rejects responses from servers without the canonical lifecycle contract', () => { + expect(() => + normalizeConversationLifecycle({ + execution_status: ConversationExecutionStatus.FINISHED, + }) + ).toThrow('Conversation response does not include canonical lifecycle fields'); + }); +}); + +describe('normalizeCloudConversationLifecycle', () => { + it('prefers an explicit archive timestamp over the sandbox status', () => { + expect( + normalizeCloudConversationLifecycle({ + archived_at: '2026-09-12T15:00:00Z', + sandbox_status: 'RUNNING', + execution_status: ConversationExecutionStatus.FINISHED, + }) + ).toEqual({ + isArchived: true, + archivedAt: '2026-09-12T15:00:00Z', + runtimeStatus: 'available', + executionStatus: ConversationExecutionStatus.FINISHED, + canResume: false, + }); + }); + + it('prefers explicit unarchived metadata over a missing sandbox', () => { + expect( + normalizeCloudConversationLifecycle({ + archived_at: null, + sandbox_status: 'MISSING', + execution_status: ConversationExecutionStatus.ERROR, + can_resume: true, + }) + ).toEqual({ + isArchived: false, + archivedAt: null, + runtimeStatus: 'missing', + executionStatus: ConversationExecutionStatus.ERROR, + canResume: true, + }); + }); + + it('temporarily infers archive state from a legacy missing sandbox', () => { + expect( + normalizeCloudConversationLifecycle({ + sandbox_status: 'MISSING', + execution_status: ConversationExecutionStatus.PAUSED, + }) + ).toEqual({ + isArchived: true, + archivedAt: null, + runtimeStatus: 'missing', + executionStatus: ConversationExecutionStatus.PAUSED, + canResume: false, + }); + }); + + it.each([ + ['RUNNING', 'available'], + ['PAUSED', 'available'], + ['STARTING', 'starting'], + ['ERROR', 'error'], + [null, 'missing'], + ] as const)( + 'maps Cloud sandbox status %s to runtime status %s', + (sandboxStatus, runtimeStatus) => { + expect( + normalizeCloudConversationLifecycle({ + archived_at: null, + sandbox_status: sandboxStatus, + execution_status: null, + }) + ).toMatchObject({ + runtimeStatus, + executionStatus: null, + }); + } + ); + + it.each([ + ['RUNNING', true], + ['PAUSED', true], + ['STARTING', false], + ['ERROR', false], + ['MISSING', false], + ] as const)( + 'infers legacy Cloud resumability for sandbox status %s as %s', + (sandboxStatus, canResume) => { + expect( + normalizeCloudConversationLifecycle({ + sandbox_status: sandboxStatus, + execution_status: ConversationExecutionStatus.RUNNING, + }) + ).toMatchObject({ canResume }); + } + ); + + it('uses explicit resumability for unarchived conversations', () => { + expect( + normalizeCloudConversationLifecycle({ + archived_at: null, + sandbox_status: 'ERROR', + execution_status: ConversationExecutionStatus.ERROR, + can_resume: false, + }) + ).toMatchObject({ + isArchived: false, + runtimeStatus: 'error', + executionStatus: ConversationExecutionStatus.ERROR, + canResume: false, + }); + }); +}); diff --git a/clients/typescript/src/client/cloud-client.ts b/clients/typescript/src/client/cloud-client.ts index 966710111df..a0afd477427 100644 --- a/clients/typescript/src/client/cloud-client.ts +++ b/clients/typescript/src/client/cloud-client.ts @@ -20,6 +20,7 @@ import type { SaveProfileRequest, SettingsSchema, } from '../models/api'; +import type { ConversationRuntimeStatus } from '../models/conversation'; export interface CloudProxyOptions { /** Agent-server or ingress host exposing `/api/cloud-proxy`. */ @@ -218,7 +219,10 @@ export interface CloudAppConversation { metrics: unknown; created_at: string; updated_at: string; + archived_at?: string | null; execution_status: string | null; + runtime_status?: ConversationRuntimeStatus | null; + can_resume?: boolean | null; sandbox_status?: string | null; conversation_url: string | null; session_api_key: string | null; diff --git a/clients/typescript/src/clients.ts b/clients/typescript/src/clients.ts index 2d999817cb9..2c31f3dac92 100644 --- a/clients/typescript/src/clients.ts +++ b/clients/typescript/src/clients.ts @@ -34,6 +34,15 @@ export { pollForToken, startDeviceFlow, } from './client/device-flow-client'; +export { + normalizeCloudConversationLifecycle, + normalizeConversationLifecycle, +} from './models/conversation-lifecycle'; +export type { + ConversationLifecycle, + ConversationLifecycleFields, + LegacyCloudConversationLifecycleFields, +} from './models/conversation-lifecycle'; export { AGENT_SERVER_VERSION_ERROR_CODE, AgentServerFeatureRequirements, diff --git a/clients/typescript/src/index.ts b/clients/typescript/src/index.ts index 43f43f8919c..0e07fcc75e5 100644 --- a/clients/typescript/src/index.ts +++ b/clients/typescript/src/index.ts @@ -254,6 +254,17 @@ export type { export { deriveSwitchPlan } from './profiles/derive-switch-plan'; export type { SwitchPlan } from './profiles/derive-switch-plan'; +// Conversation lifecycle +export { + normalizeCloudConversationLifecycle, + normalizeConversationLifecycle, +} from './models/conversation-lifecycle'; +export type { + ConversationLifecycle, + ConversationLifecycleFields, + LegacyCloudConversationLifecycleFields, +} from './models/conversation-lifecycle'; + // Conversation models export type { ConversationInfo, diff --git a/clients/typescript/src/models/conversation-lifecycle.ts b/clients/typescript/src/models/conversation-lifecycle.ts new file mode 100644 index 00000000000..b8ee7146114 --- /dev/null +++ b/clients/typescript/src/models/conversation-lifecycle.ts @@ -0,0 +1,113 @@ +import { ConversationExecutionStatus } from '../types/base'; +import type { ConversationInfo, ConversationRuntimeStatus } from './conversation'; + +export interface ConversationLifecycle { + isArchived: boolean; + archivedAt: string | null; + runtimeStatus: ConversationRuntimeStatus; + executionStatus: ConversationExecutionStatus | null; + canResume: boolean; +} + +export type ConversationLifecycleFields = Pick< + ConversationInfo, + 'archived_at' | 'runtime_info' | 'execution_status' +>; + +/** + * Runtime semantics for a local (in-process) conversation, mirroring + * `ConversationRegistry.runtime_info` on the Agent Server. Used when a response + * carries `archived_at` but omits the optional `runtime_info` — e.g. the + * `create`/`fork`/`navigate` responses, which return a bare `ConversationInfo`. + */ +const LOCAL_RUNTIME_STATUS: ConversationRuntimeStatus = 'available'; +const LOCAL_CAN_RESUME = true; + +export interface LegacyCloudConversationLifecycleFields { + archived_at?: string | null; + runtime_status?: ConversationRuntimeStatus | null; + execution_status?: string | null; + can_resume?: boolean | null; + sandbox_status?: string | null; +} + +export function normalizeConversationLifecycle( + conversation: ConversationLifecycleFields +): ConversationLifecycle { + const { archived_at: archivedAt, runtime_info: runtimeInfo } = conversation; + if (archivedAt === undefined) { + throw new Error('Conversation response does not include canonical lifecycle fields'); + } + + // `runtime_info` is optional on the public `ConversationInfo`: the catalog + // routes add it, but the create/fork/navigate handlers return a bare + // `ConversationInfo`. A response that carries `archived_at` but no + // `runtime_info` is therefore a perfectly canonical local conversation, not a + // pre-contract server, and callers must not need a version check to read it. + // Fall back to the local runtime semantics the Agent Server itself reports + // (`ConversationRegistry.runtime_info`: always available and resumable). + return { + isArchived: archivedAt !== null, + archivedAt, + runtimeStatus: runtimeInfo?.runtime_status ?? LOCAL_RUNTIME_STATUS, + executionStatus: conversation.execution_status, + canResume: runtimeInfo?.can_resume ?? LOCAL_CAN_RESUME, + }; +} + +export function normalizeCloudConversationLifecycle( + conversation: LegacyCloudConversationLifecycleFields +): ConversationLifecycle { + const archivedAt = conversation.archived_at ?? null; + const isArchived = + conversation.archived_at !== undefined + ? archivedAt !== null + : inferLegacyCloudArchiveState(conversation.sandbox_status); + + const runtimeStatus = + conversation.runtime_status ?? cloudRuntimeStatus(conversation.sandbox_status); + + return { + isArchived, + archivedAt, + runtimeStatus, + executionStatus: cloudExecutionStatus(conversation.execution_status), + // Matches the Agent Server, which reports an available runtime as resumable. + canResume: conversation.can_resume ?? (!isArchived && runtimeStatus === 'available'), + }; +} + +/** + * @deprecated Replaced by explicit Cloud `archived_at`. Remove when every + * supported Cloud deployment returns that field. See the linked removal issue. + */ +function inferLegacyCloudArchiveState(sandboxStatus: string | null | undefined): boolean { + return sandboxStatus === 'MISSING'; +} + +function cloudRuntimeStatus(sandboxStatus: string | null | undefined): ConversationRuntimeStatus { + switch (sandboxStatus) { + case 'RUNNING': + case 'PAUSED': + return 'available'; + case 'STARTING': + return 'starting'; + case 'ERROR': + return 'error'; + case 'MISSING': + case null: + case undefined: + default: + return 'missing'; + } +} + +function cloudExecutionStatus( + executionStatus: string | null | undefined +): ConversationExecutionStatus | null { + return Object.values(ConversationExecutionStatus).includes( + executionStatus as ConversationExecutionStatus + ) + ? (executionStatus as ConversationExecutionStatus) + : null; +} From 18b936d52c837255cb80d5fc1232fd466d6e8e48 Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 23 Oct 2026 18:45:59 +0000 Subject: [PATCH 09/10] fix(agent-server): distinguish runtime expiration from conversation archive Expose runtime_expired through the existing runtime error contract, preserve legacy retention markers, and verify archive/unarchive cannot revive expired storage. Co-authored-by: openhands --- .../openhands/agent_server/README.md | 24 +++- .../openhands/agent_server/config.py | 6 +- .../agent_server/docker_runtime/registry.py | 32 ++++-- .../agent_server/docker_runtime/routers.py | 18 +-- .../agent_server/docker_runtime/storage.py | 6 +- .../agent_server/storage/reclaimer.py | 40 +++---- .../docker_runtime/STORAGE_SCENARIO.md | 2 +- .../docker_runtime/test_retention.py | 40 +++++-- .../docker_runtime/test_routes.py | 104 ++++++++++++++++++ .../docker_runtime/test_storage_scenario.py | 8 +- tests/agent_server/storage/test_reclaimer.py | 4 +- 11 files changed, 222 insertions(+), 62 deletions(-) diff --git a/openhands-agent-server/openhands/agent_server/README.md b/openhands-agent-server/openhands/agent_server/README.md index 96afe3a752f..0aff1aaead4 100644 --- a/openhands-agent-server/openhands/agent_server/README.md +++ b/openhands-agent-server/openhands/agent_server/README.md @@ -18,11 +18,33 @@ The OpenHands Agent Server is a minimal REST API and WebSocket server that provi Conversation metadata separates retained history from runtime availability: - `runtime_status` reports whether execution infrastructure is available; it does not imply archive state. -- `POST /api/conversations/{id}/archive` sets `archived_at`, hides the conversation from the default search, and stops only runtime infrastructure owned by this server. It preserves history and workspace files. +- `POST /api/conversations/{id}/archive` sets `archived_at`, hides the conversation from the default search, and stops only runtime infrastructure owned by this server. It does not delete history or workspace files; separately configured runtime retention still applies. - `POST /api/conversations/{id}/unarchive` clears `archived_at` without provisioning infrastructure or starting agent execution. - `GET /api/conversations/search?archived=true` returns archived conversations; the default `archived=false` returns active conversations. `GET /api/conversations/count` accepts the same `archived` filter and default. - `DELETE /api/conversations/{id}` permanently removes the conversation rather than retaining it as archived. +Runtime retention is **expiration**, not conversation archive. When +`conversation_storage_retention_days` is configured, an inactive Docker runtime +can expire: its server-owned home and workspace are deleted, while conversation +history and its encryption identity remain readable. Caller-owned workspaces are +not deleted. Expiration does not change `archived_at` or search membership. +The runtime reports `runtime_status: "missing"`, `can_resume: false`, and +`runtime_error.code: "runtime_expired"`; unarchive cannot restore expired storage, +and resume/reprovision returns HTTP 410. The legacy `.archived.json` runtime marker +is retained on disk for rollback compatibility; it is not the conversation's +`meta.json.archived_at` field. Retention is disabled by default. + +`can_resume` describes runtime recovery capability, not permission to execute now: +an explicitly archived conversation must first be unarchived even when +`can_resume` is true. Unarchive only restores catalog visibility; it does not +start compute or guarantee that runtime storage is still available. + +Search and count select disjoint archive partitions. A UI's "Show archived", a +backup inventory, or a bulk-cleanup client that needs **all** conversations must +query both `archived=false` and `archived=true` and paginate each independently. +Do not infer archive intent from a missing runtime or substitute a provider's +destructive sandbox-delete operation for a history-preserving archive. + ## Quick Start ### Prerequisites diff --git a/openhands-agent-server/openhands/agent_server/config.py b/openhands-agent-server/openhands/agent_server/config.py index f461e9fc3d1..c5f05201e8a 100644 --- a/openhands-agent-server/openhands/agent_server/config.py +++ b/openhands-agent-server/openhands/agent_server/config.py @@ -451,13 +451,13 @@ class Config(BaseModel): gt=0, description=( "Days a stopped conversation may stay inactive before its runtime " - "(sandbox home and workspace) is deleted and the conversation is " - "archived: its history and encryption identity are kept, so it " + "(sandbox home and workspace) expires and is deleted. Its history " + "and encryption identity are kept, so the conversation " "stays readable but can no longer be resumed. A workspace the " "server did not create is never deleted. Unset (the default) keeps " "runtimes until the conversation is deleted. Leave it unset when " "the server runs in a sandbox that its provider pauses: no pass " - "runs while paused, so the first pass after a resume archives " + "runs while paused, so the first pass after a resume expires " "every conversation that crossed the limit in the meantime." ), ) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py b/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py index b17b1a0bda4..4fad69e82a2 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py @@ -22,6 +22,7 @@ from openhands.agent_server.docker_runtime.storage import DockerRuntimeStorage from openhands.agent_server.launch import container_browser_enabled from openhands.agent_server.models import ( + ConversationRuntimeError, ConversationRuntimeInfo, ConversationRuntimeStatus, ) @@ -75,7 +76,7 @@ ) -class RuntimeArchivedError(RuntimeError): +class RuntimeExpiredError(RuntimeError): """Retention deleted the runtime; the conversation is read-only.""" @@ -141,17 +142,30 @@ def resolve_persisted_cipher(self, conversation_id: UUID) -> Cipher: identity = self.provisioning.load_optional(conversation_id) return identity.cipher if identity is not None else self.provisioning.cipher - def archived_marker(self, conversation_id: UUID) -> Path: - # Beside the manifest, not in it: the manifest must stay loadable by - # older builds, and it holds the key to the conversation's history. + def expired_marker(self, conversation_id: UUID) -> Path: + # Keep the legacy filename so older builds also refuse to revive + # an expired runtime with an empty workspace. return self.provisioning.control_root / f"{conversation_id.hex}.archived.json" - def is_archived(self, conversation_id: UUID) -> bool: - return self.archived_marker(conversation_id).exists() + def is_expired(self, conversation_id: UUID) -> bool: + return self.expired_marker(conversation_id).exists() def runtime_info(self, conversation_id: UUID) -> ConversationRuntimeInfo: identity = self.provisioning.load_optional(conversation_id) - if identity is None or self.is_archived(conversation_id): + if self.is_expired(conversation_id): + return ConversationRuntimeInfo( + runtime_status=ConversationRuntimeStatus.MISSING, + can_resume=False, + runtime_error=ConversationRuntimeError( + code="runtime_expired", + message=( + "Runtime retention expired; this runtime cannot be resumed. " + "Unarchiving does not restore deleted workspace files; " + "conversation history remains readable." + ), + ), + ) + if identity is None: return ConversationRuntimeInfo( runtime_status=ConversationRuntimeStatus.MISSING, can_resume=False, @@ -281,8 +295,8 @@ async def get_or_create(self, conversation_id: UUID) -> ConversationContainer: async with self._lock: if conversation_id in self._deleting: raise RuntimeError("Conversation is being deleted") - if self.is_archived(conversation_id): - raise RuntimeArchivedError("Conversation runtime was archived") + if self.is_expired(conversation_id): + raise RuntimeExpiredError("Conversation runtime expired") self._last_access[conversation_id] = time.monotonic() container = self._containers.get(conversation_id) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py index f1a10c0019e..1b9784de336 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -23,7 +23,7 @@ from openhands.agent_server.docker_runtime.registry import ( ConversationContainer, DockerConversationRegistry, - RuntimeArchivedError, + RuntimeExpiredError, ) from openhands.agent_server.launch import ( forward_to_runtime, @@ -59,7 +59,9 @@ def get_registry(request: Request) -> DockerConversationRegistry: return registry -def _is_archived(registry: DockerConversationRegistry, conversation_id: UUID) -> bool: +def _is_conversation_archived( + registry: DockerConversationRegistry, conversation_id: UUID +) -> bool: try: metadata_path = registry.conversation_dir(conversation_id) / "meta.json" except ValueError: @@ -76,15 +78,17 @@ async def _container( ) -> ConversationContainer: if not registry.provisioning.manifest_path(conversation_id).is_file(): raise HTTPException(404, "Conversation not found") - if _is_archived(registry, conversation_id): + if _is_conversation_archived(registry, conversation_id): raise HTTPException( 409, "Conversation is archived; unarchive it before using its runtime" ) try: return await registry.get_or_create(conversation_id) - except RuntimeArchivedError as exc: + except RuntimeExpiredError as exc: raise HTTPException( - 410, "Conversation runtime was archived; its history is read-only" + 410, + "Conversation runtime expired; unarchiving cannot restore its workspace; " + "history is read-only", ) from exc except Exception as exc: logger.exception("Could not start conversation container %s", conversation_id) @@ -176,7 +180,7 @@ async def start_conversation( ) except ValueError as exc: raise HTTPException(422, str(exc)) from exc - if _is_archived(registry, conversation_id): + if _is_conversation_archived(registry, conversation_id): raise HTTPException( 409, "Conversation is archived; unarchive it before using its runtime" ) @@ -384,7 +388,7 @@ async def delete_conversation(conversation_id: UUID, request: Request) -> Respon try: await registry.stop(conversation_id) registry.provisioning.manifest_path(conversation_id).unlink(missing_ok=True) - registry.archived_marker(conversation_id).unlink(missing_ok=True) + registry.expired_marker(conversation_id).unlink(missing_ok=True) await asyncio.to_thread( safe_rmtree, registry.provisioning.runtime_dir(conversation_id) ) diff --git a/openhands-agent-server/openhands/agent_server/docker_runtime/storage.py b/openhands-agent-server/openhands/agent_server/docker_runtime/storage.py index a2eb055b651..fe66c1e0c45 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/storage.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/storage.py @@ -79,14 +79,14 @@ def runtimes(self) -> list[StoredRuntime]: def idle(self, conversation_id: UUID) -> AbstractAsyncContextManager[bool]: return self.registry.runtime_idle(conversation_id) - def archive(self, conversation_id: UUID) -> list[Path]: + def expire(self, conversation_id: UUID) -> list[Path]: # Marker first: if the move never happens, the next pass retries it. - self.registry.archived_marker(conversation_id).write_text( + self.registry.expired_marker(conversation_id).write_text( json.dumps({"archived_at": datetime.now(UTC).isoformat()}) ) return [self.provisioning.runtime_dir(conversation_id)] - async def on_archived(self, conversation_id: UUID) -> None: + async def on_expired(self, conversation_id: UUID) -> None: await self.registry.refresh_conversation(conversation_id) diff --git a/openhands-agent-server/openhands/agent_server/storage/reclaimer.py b/openhands-agent-server/openhands/agent_server/storage/reclaimer.py index bdee6100414..3a4163782a8 100644 --- a/openhands-agent-server/openhands/agent_server/storage/reclaimer.py +++ b/openhands-agent-server/openhands/agent_server/storage/reclaimer.py @@ -40,11 +40,11 @@ def idle(self, conversation_id: UUID) -> AbstractAsyncContextManager[bool]: """Hold the runtime still; True if nothing runs or starts in it.""" ... - def archive(self, conversation_id: UUID) -> list[Path]: - """Record the runtime as archived; return what to discard. Called idle.""" + def expire(self, conversation_id: UUID) -> list[Path]: + """Record the runtime as expired; return what to discard. Called idle.""" ... - async def on_archived(self, conversation_id: UUID) -> None: ... + async def on_expired(self, conversation_id: UUID) -> None: ... class Reclaimer: @@ -87,18 +87,18 @@ async def shutdown(self) -> None: async def run_pass(self) -> None: """Apply the configured policies once; log only if something was freed.""" free_before = _free_bytes(self.storage.root) - # Retention first: what it archives no longer counts against the budget. - archived = await self._enforce_retention() + # Retention first: what it expires no longer counts against the budget. + expired = await self._enforce_retention() shed = await self._enforce_disk_budget() - if not (archived or shed): + if not (expired or shed): return await self.trash.drain() logger.info( - "Conversation storage: freed %.1f GB (archived %d runtimes, shed " + "Conversation storage: freed %.1f GB (expired %d runtimes, shed " "dependencies of %d), %s at %.0f%%", # Other writes on the filesystem can outweigh a small reclaim. max(0, _free_bytes(self.storage.root) - free_before) / 1e9, - archived, + expired, shed, self.storage.root, disk_usage(self.storage.root) * 100, @@ -118,17 +118,17 @@ async def _maintenance_loop(self) -> None: logger.exception("error_reclaiming_conversation_storage") async def _enforce_retention(self) -> int: - """Archive runtimes inactive for longer than the retention period.""" + """Expire runtimes inactive for longer than the retention period.""" if not self.retention_days: return 0 cutoff = time.time() - self.retention_days * 86400 runtimes = await asyncio.to_thread(self.storage.runtimes) # Least recently active first, so the stale ones are a prefix. stale = itertools.takewhile(lambda r: r.last_active <= cutoff, runtimes) - archived = sum([await self.reclaim(r, Tier.RUNTIME) for r in stale]) - if archived: + expired = sum([await self.reclaim(r, Tier.RUNTIME) for r in stale]) + if expired: self.trash.empty_soon() - return archived + return expired async def _enforce_disk_budget(self) -> int: """Shed dependencies of stopped runtimes, oldest first, until under budget.""" @@ -159,20 +159,20 @@ async def _enforce_disk_budget(self) -> int: async def reclaim(self, runtime: StoredRuntime, tier: Tier) -> bool: """Discard what ``tier`` allows, unless the runtime is in use.""" - archiving = tier >= Tier.RUNTIME - # Archiving takes the whole runtime, so there is nothing to select. - paths = [] if archiving else await asyncio.to_thread(_select, runtime, tier) - if not (paths or archiving): + expiring = tier >= Tier.RUNTIME + # Expiring takes the whole runtime, so there is nothing to select. + paths = [] if expiring else await asyncio.to_thread(_select, runtime, tier) + if not (paths or expiring): return False async with self.storage.idle(runtime.id) as idle: if not idle: return False - if archiving: - paths = self.storage.archive(runtime.id) + if expiring: + paths = self.storage.expire(runtime.id) # Only renames here: the lock is held for microseconds. moved = [path for path in paths if self.trash.discard(path)] - if archiving and moved: - await self.storage.on_archived(runtime.id) + if expiring and moved: + await self.storage.on_expired(runtime.id) return bool(moved) diff --git a/tests/agent_server/docker_runtime/STORAGE_SCENARIO.md b/tests/agent_server/docker_runtime/STORAGE_SCENARIO.md index dda8bd57a83..992f33baa79 100644 --- a/tests/agent_server/docker_runtime/STORAGE_SCENARIO.md +++ b/tests/agent_server/docker_runtime/STORAGE_SCENARIO.md @@ -37,7 +37,7 @@ to what must go: 4. **A reclaimed conversation resumes and works**: reading a workspace file through the API restarts it, its commit is there, and it can write caches and dependencies again. -5. **An archived conversation is read-only until deleted** (retention 1 day, +5. **An expired runtime stays read-only until deleted** (retention 1 day, one conversation aged 2 days): its runtime is gone, its history still reads, the runtime routes answer 410, the recent one is kept, and delete still removes everything. diff --git a/tests/agent_server/docker_runtime/test_retention.py b/tests/agent_server/docker_runtime/test_retention.py index 901345412cc..eb2520a6d94 100644 --- a/tests/agent_server/docker_runtime/test_retention.py +++ b/tests/agent_server/docker_runtime/test_retention.py @@ -12,7 +12,7 @@ from openhands.agent_server.docker_runtime.registry import ( ConversationContainer, DockerConversationRegistry, - RuntimeArchivedError, + RuntimeExpiredError, ) from openhands.agent_server.models import ConversationRuntimeStatus @@ -55,7 +55,7 @@ def provision( @pytest.mark.asyncio -async def test_inactive_runtime_is_archived_and_history_kept(tmp_path, monkeypatch): +async def test_inactive_runtime_is_expired_and_history_kept(tmp_path, monkeypatch): runtime = registry(tmp_path, monkeypatch) conversation_id, workspace = provision(runtime, days_inactive=8) persisted = runtime.resolve_persisted_cipher(conversation_id).encrypt( @@ -74,6 +74,8 @@ async def test_inactive_runtime_is_archived_and_history_kept(tmp_path, monkeypat info = runtime.runtime_info(conversation_id) assert info.runtime_status == ConversationRuntimeStatus.MISSING assert info.can_resume is False + assert info.runtime_error is not None + assert info.runtime_error.code == "runtime_expired" assert list(runtime.reclaimer.trash.dir.iterdir()) == [] @@ -99,11 +101,11 @@ async def test_live_runtime_is_kept_however_old(tmp_path, monkeypatch): await runtime.reclaimer.run_pass() assert (workspace / "main.py").is_file() - assert not runtime.is_archived(conversation_id) + assert not runtime.is_expired(conversation_id) @pytest.mark.asyncio -async def test_caller_supplied_workspace_survives_archiving(tmp_path, monkeypatch): +async def test_caller_supplied_workspace_survives_expiration(tmp_path, monkeypatch): runtime = registry(tmp_path, monkeypatch) checkout = tmp_path / "user-checkout" checkout.mkdir() @@ -111,22 +113,22 @@ async def test_caller_supplied_workspace_survives_archiving(tmp_path, monkeypatc await runtime.reclaimer.run_pass() - assert runtime.is_archived(conversation_id) + assert runtime.is_expired(conversation_id) assert (checkout / "main.py").is_file() @pytest.mark.asyncio -async def test_archived_runtime_cannot_be_resumed(tmp_path, monkeypatch): +async def test_expired_runtime_cannot_be_resumed(tmp_path, monkeypatch): runtime = registry(tmp_path, monkeypatch) conversation_id, _ = provision(runtime, days_inactive=8) await runtime.reclaimer.run_pass() def no_docker(_conversation_id): - pytest.fail("an archived runtime must not start a container") + pytest.fail("an expired runtime must not start a container") # Without this, a regression would start a real container and leak it. monkeypatch.setattr(runtime, "_build_container", no_docker) - with pytest.raises(RuntimeArchivedError): + with pytest.raises(RuntimeExpiredError): await runtime.get_or_create(conversation_id) with pytest.raises(HTTPException) as raised: await routers._container(runtime, conversation_id) @@ -134,13 +136,13 @@ def no_docker(_conversation_id): @pytest.mark.asyncio -async def test_interrupted_archiving_is_finished_by_the_next_pass( +async def test_interrupted_expiration_is_finished_by_the_next_pass( tmp_path, monkeypatch ): runtime = registry(tmp_path, monkeypatch) conversation_id, _ = provision(runtime, days_inactive=8) # A crash after the marker, before the runtime dir was moved. - runtime.archived_marker(conversation_id).touch() + runtime.expired_marker(conversation_id).touch() await runtime.reclaimer.run_pass() @@ -148,7 +150,7 @@ async def test_interrupted_archiving_is_finished_by_the_next_pass( @pytest.mark.asyncio -async def test_archiving_leaves_the_manifest_untouched(tmp_path, monkeypatch): +async def test_expiration_leaves_the_manifest_untouched(tmp_path, monkeypatch): # Older builds forbid unknown manifest fields; a rollback must still load it. runtime = registry(tmp_path, monkeypatch) conversation_id, _ = provision(runtime, days_inactive=8) @@ -157,7 +159,7 @@ async def test_archiving_leaves_the_manifest_untouched(tmp_path, monkeypatch): await runtime.reclaimer.run_pass() - assert runtime.is_archived(conversation_id) + assert runtime.is_expired(conversation_id) assert manifest.read_bytes() == before assert runtime.provisioning.load(conversation_id).conversation_id == conversation_id @@ -179,3 +181,17 @@ async def test_retention_loop_runs_only_when_configured(tmp_path, monkeypatch): assert (runtime.reclaimer._maintenance is not None) is expected await runtime.shutdown() assert runtime.reclaimer._maintenance is None + + +def test_legacy_expiration_marker_remains_non_resumable(tmp_path, monkeypatch): + runtime = registry(tmp_path, monkeypatch) + conversation_id, workspace = provision(runtime, days_inactive=8) + marker = runtime.provisioning.control_root / f"{conversation_id.hex}.archived.json" + marker.write_text('{"archived_at":"2026-09-01T00:00:00Z"}') + + info = runtime.runtime_info(conversation_id) + + assert info.can_resume is False + assert info.runtime_error is not None + assert info.runtime_error.code == "runtime_expired" + assert workspace.exists() diff --git a/tests/agent_server/docker_runtime/test_routes.py b/tests/agent_server/docker_runtime/test_routes.py index 4f5d4764c91..fa4ba643124 100644 --- a/tests/agent_server/docker_runtime/test_routes.py +++ b/tests/agent_server/docker_runtime/test_routes.py @@ -1,5 +1,7 @@ import asyncio import json +import os +import time from types import SimpleNamespace from unittest.mock import AsyncMock, call from uuid import uuid4 @@ -15,6 +17,10 @@ from openhands.agent_server.api import create_app from openhands.agent_server.config import Config +from openhands.agent_server.conversation_router import ( + conversation_catalog_router, + get_conversation, +) from openhands.agent_server.conversation_service import ConversationService from openhands.agent_server.docker_runtime.provisioning import RuntimeProvisioningStore from openhands.agent_server.docker_runtime.registry import ( @@ -745,3 +751,101 @@ def test_invalid_container_options_are_rejected(tmp_path, monkeypatch): assert response.status_code == 422 assert registry.provisioning.load_optional(conversation_id) is None + + +@pytest.mark.asyncio +async def test_expiration_is_independent_of_archive_and_unarchive( + tmp_path, monkeypatch +): + monkeypatch.setenv("OH_PERSISTENCE_DIR", str(tmp_path / "persistence")) + config = Config( + conversations_path=tmp_path / "conversations", + workspace_path=tmp_path / "workspaces", + secret_key=SecretStr("outer-key"), + conversation_storage_retention_days=1, + ) + registry = DockerConversationRegistry(config) + conversation_id = uuid4() + identity = registry.provisioning.create(conversation_id) + workspace = identity.workspace_path + (workspace / "main.py").write_text("print('retained until expiry')") + directory = registry.conversation_dir(conversation_id) + directory.mkdir(parents=True) + stored = StoredConversation( + id=conversation_id, + workspace=LocalWorkspace(working_dir=str(workspace)), + confirmation_policy=NeverConfirm(), + ) + (directory / "meta.json").write_text(stored.model_dump_json()) + state = directory / "base_state.json" + state.write_text( + ConversationState( + id=conversation_id, + agent=Agent(llm=LLM(model="test-model"), tools=[]), + workspace=stored.workspace, + confirmation_policy=stored.confirmation_policy, + ).model_dump_json() + ) + old = time.time() - 2 * 86400 + os.utime(state, (old, old)) + + def reject_container_start(_): + pytest.fail("expired runtime must never start a container") + + monkeypatch.setattr(registry, "_build_container", reject_container_start) + app = FastAPI() + app.state.conversation_registry = registry + app.include_router(conversation_catalog_router, prefix="/api") + app.add_api_route("/api/conversations/{conversation_id}", get_conversation) + app.include_router(event_read_router, prefix="/api") + app.include_router(docker_conversation_router, prefix="/api") + service = ConversationService(conversations_dir=config.conversations_path) + registry.configure_service(service) + async with service: + app.state.conversation_service = service + await registry.reclaimer.run_pass() + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + info = (await client.get(f"/api/conversations/{conversation_id}")).json() + assert info["archived_at"] is None + assert info["runtime_info"]["can_resume"] is False + assert info["runtime_info"]["runtime_error"]["code"] == "runtime_expired" + assert not workspace.exists() + active = (await client.get("/api/conversations/search")).json()["items"] + assert [item["id"] for item in active] == [str(conversation_id)] + + archived = await client.post( + f"/api/conversations/{conversation_id}/archive" + ) + assert archived.status_code == 200 + assert archived.json()["archived_at"] is not None + assert (await client.get("/api/conversations/search")).json()["items"] == [] + archived_items = ( + await client.get("/api/conversations/search", params={"archived": True}) + ).json()["items"] + assert [item["id"] for item in archived_items] == [str(conversation_id)] + + restored = await client.post( + f"/api/conversations/{conversation_id}/unarchive" + ) + assert restored.status_code == 200 + assert restored.json()["archived_at"] is None + assert restored.json()["runtime_info"] == info["runtime_info"] + for path in ("runtime/resume", "runtime/reprovision"): + response = await client.post( + f"/api/conversations/{conversation_id}/{path}" + ) + assert response.status_code == 410 + assert "expired" in response.json()["detail"].lower() + history = await client.get( + f"/api/conversations/{conversation_id}/events/search" + ) + assert history.status_code == 200 + assert not workspace.exists() + assert ( + await client.delete(f"/api/conversations/{conversation_id}") + ).status_code == 200 + assert ( + await client.get(f"/api/conversations/{conversation_id}") + ).status_code == 404 diff --git a/tests/agent_server/docker_runtime/test_storage_scenario.py b/tests/agent_server/docker_runtime/test_storage_scenario.py index 0d6e9bbef9e..5ed1af7b7ea 100644 --- a/tests/agent_server/docker_runtime/test_storage_scenario.py +++ b/tests/agent_server/docker_runtime/test_storage_scenario.py @@ -298,7 +298,7 @@ def test_a_reclaimed_conversation_resumes_and_works(server): assert out.split() == ["print(1)", "init", "ok"] -def test_an_archived_conversation_is_read_only_until_deleted(server): +def test_an_expired_runtime_is_read_only_until_deleted(server): with server(OH_CONVERSATION_STORAGE_RETENTION_DAYS="1") as client: registry = _registry(client) old, recent = _start(client), _start(client) @@ -311,9 +311,9 @@ def test_an_archived_conversation_is_read_only_until_deleted(server): _pass(client) - assert registry.is_archived(old) + assert registry.is_expired(old) assert not registry.provisioning.runtime_dir(old).exists() - assert not registry.is_archived(recent) + assert not registry.is_expired(recent) assert (_workspace(client, recent) / "main.py").is_file() # History stays readable; the runtime cannot come back. assert client.get(f"/api/conversations/{old}").status_code == 200 @@ -324,5 +324,5 @@ def test_an_archived_conversation_is_read_only_until_deleted(server): assert registry.get(old) is None assert client.delete(f"/api/conversations/{old}").status_code in (200, 204) - assert not registry.is_archived(old) + assert not registry.is_expired(old) assert not registry.conversation_dir(old).exists() diff --git a/tests/agent_server/storage/test_reclaimer.py b/tests/agent_server/storage/test_reclaimer.py index 4a7a92ba94b..f4f8aa8c736 100644 --- a/tests/agent_server/storage/test_reclaimer.py +++ b/tests/agent_server/storage/test_reclaimer.py @@ -38,10 +38,10 @@ def runtimes(self) -> list[StoredRuntime]: async def idle(self, conversation_id: UUID) -> AsyncIterator[bool]: yield conversation_id not in self.busy - def archive(self, conversation_id: UUID) -> list[Path]: + def expire(self, conversation_id: UUID) -> list[Path]: return [] - async def on_archived(self, conversation_id: UUID) -> None: + async def on_expired(self, conversation_id: UUID) -> None: pass From dc9eeab0901edeedaac6af307e6203159e40e9ad Mon Sep 17 00:00:00 2001 From: openhands Date: Fri, 23 Oct 2026 20:05:07 +0000 Subject: [PATCH 10/10] test(agent-server): make websocket backpressure assertion deterministic Check synchronous publisher admission without a CPU-time threshold, retain a deadlock watchdog, and always close the writer. Co-authored-by: openhands --- tests/agent_server/test_session_socket.py | 31 +++++++++++++---------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/tests/agent_server/test_session_socket.py b/tests/agent_server/test_session_socket.py index e707defe1c9..2282c1386bc 100644 --- a/tests/agent_server/test_session_socket.py +++ b/tests/agent_server/test_session_socket.py @@ -243,20 +243,23 @@ async def test_wedged_connection_drops_instead_of_blocking_the_publisher(): sub = _SessionSubscriber(writer=writer, events=log) # type: ignore[arg-type] sub.go_live(None) - # Publish far past the budget, and time the publisher. - started = asyncio.get_running_loop().time() - for i in range(500): - event = _msg(f"m{i}" * 50) - log.add(event) - await sub(event) - elapsed = asyncio.get_running_loop().time() - started - - assert elapsed < 0.5, "publisher blocked on a wedged consumer" - assert writer.closed, "over-budget connection should be dropped" - assert writer.drop_reason == "slow_consumer" - - ws.gate.set() - await writer.aclose() + yielded = asyncio.Event() + + async def publish(): + asyncio.get_running_loop().call_soon(yielded.set) + for i in range(500): + event = _msg(f"m{i}" * 50) + log.add(event) + await sub(event) + assert not yielded.is_set(), "publisher yielded to a wedged consumer" + + try: + await asyncio.wait_for(publish(), timeout=5) + assert writer.closed, "over-budget connection should be dropped" + assert writer.drop_reason == "slow_consumer" + finally: + ws.gate.set() + await writer.aclose() @pytest.mark.asyncio