diff --git a/.pr/runtime-status-validation.md b/.pr/runtime-status-validation.md new file mode 100644 index 0000000000..ff7bc16861 --- /dev/null +++ b/.pr/runtime-status-validation.md @@ -0,0 +1,11 @@ +# Runtime status validation + +After a failed start, an explicit stop or server shutdown clears the recorded +runtime error. A concurrent start failure cannot recreate that error after the +stop has claimed its startup task. Retained conversation state reports a missing, +resumable runtime instead of a stale error. + +Validation: 27 registry and scoped-route tests passed, including stop/shutdown +races. Ruff, Pyright, and the repository pre-commit checks passed locally. +The parent Docker-runtime branch now contains the current main baseline; CI must +use that refreshed parent when checking baseline monotonicity. diff --git a/clients/typescript/src/__tests__/api-clients.test.ts b/clients/typescript/src/__tests__/api-clients.test.ts index 3efa1d0004..cb19b0d8af 100644 --- a/clients/typescript/src/__tests__/api-clients.test.ts +++ b/clients/typescript/src/__tests__/api-clients.test.ts @@ -2285,6 +2285,8 @@ describe('Auxiliary API clients', () => { await client.respondToConfirmation('c1', { accept: true }); await client.deleteConversation('c1'); await client.updateConversation('c1', { title: 'New title' }); + await client.getRuntime('c1'); + await client.reprovisionRuntime('c1'); expect(global.fetch).toHaveBeenNthCalledWith( 1, @@ -2314,6 +2316,16 @@ describe('Auxiliary API clients', () => { 'http://example.com/api/conversations/c1', expect.objectContaining({ method: 'PATCH', body: JSON.stringify({ title: 'New title' }) }) ); + expect(global.fetch).toHaveBeenNthCalledWith( + 9, + 'http://example.com/api/conversations/c1/runtime', + expect.objectContaining({ method: 'GET' }) + ); + expect(global.fetch).toHaveBeenNthCalledWith( + 10, + 'http://example.com/api/conversations/c1/runtime/reprovision', + 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 cb032b691f..d24d3849c0 100644 --- a/clients/typescript/src/client/conversation-client.ts +++ b/clients/typescript/src/client/conversation-client.ts @@ -9,6 +9,7 @@ import type { ConversationEventPage, ConversationEventSearchOptions, ConversationInfo, + ConversationRuntimeInfo, ConversationSearchRequest, ConversationSearchResponse, ForkConversationRequest, @@ -135,6 +136,21 @@ export class ConversationClient { return response.data; } + async getRuntime(conversationId: string): Promise { + const response = await this.client.get( + `/api/conversations/${conversationId}/runtime` + ); + return response.data; + } + + async reprovisionRuntime(conversationId: string): Promise { + const response = await this.client.post( + `/api/conversations/${conversationId}/runtime/reprovision`, + {} + ); + return response.data; + } + async searchEvents( conversationId: string, options: ConversationEventSearchOptions = {} diff --git a/clients/typescript/src/index.ts b/clients/typescript/src/index.ts index b042546fef..89d01a9717 100644 --- a/clients/typescript/src/index.ts +++ b/clients/typescript/src/index.ts @@ -255,6 +255,9 @@ export type { SwitchPlan } from './profiles/derive-switch-plan'; // Conversation models export type { ConversationInfo, + ConversationRuntimeStatus, + ConversationRuntimeError, + ConversationRuntimeInfo, ACPAgentConfig, ACPConversationInfo, SendMessageRequest, diff --git a/clients/typescript/src/models/conversation.ts b/clients/typescript/src/models/conversation.ts index 0eb5f5d13e..d5c57d9b78 100644 --- a/clients/typescript/src/models/conversation.ts +++ b/clients/typescript/src/models/conversation.ts @@ -23,6 +23,24 @@ export enum ConversationSortOrder { UPDATED_AT_DESC = 'UPDATED_AT_DESC', } +export type ConversationRuntimeStatus = + | 'available' + | 'starting' + | 'missing' + | 'ownership_lost' + | 'error'; + +export interface ConversationRuntimeError { + code: string; + message: string; +} + +export interface ConversationRuntimeInfo { + runtime_status: ConversationRuntimeStatus; + can_resume: boolean; + runtime_error: ConversationRuntimeError | null; +} + export interface ConversationInfo { id: ConversationID; /** @@ -30,6 +48,12 @@ export interface ConversationInfo { * Note: This field was renamed from agent_status to execution_status in the API. */ execution_status: ConversationExecutionStatus; + /** Runtime availability. Absent on agent-server versions before this contract. */ + runtime_status?: ConversationRuntimeStatus; + /** Whether retained state permits explicit execution resumption. */ + can_resume?: boolean; + /** Latest structured runtime failure, when known. */ + runtime_error?: ConversationRuntimeError | null; /** * @deprecated Use execution_status instead. This field is kept for backward compatibility. */ diff --git a/openhands-agent-server/openhands/agent_server/conversation_router.py b/openhands-agent-server/openhands/agent_server/conversation_router.py index 6e252f667c..109ff1b16b 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_router.py +++ b/openhands-agent-server/openhands/agent_server/conversation_router.py @@ -31,6 +31,8 @@ AskAgentResponse, ConversationInfo, ConversationPage, + ConversationRuntimeInfo, + ConversationRuntimeStatus, ConversationSortOrder, ForkConversationRequest, NavigateConversationRequest, @@ -61,6 +63,16 @@ from openhands.tools.preset.default import get_default_tools +def _with_runtime_lifecycle( + request: Request, conversation: ConversationInfo +) -> ConversationInfo: + registry = getattr(request.app.state, "docker_registry", None) + if registry is None: + return conversation + lifecycle = registry.runtime_info(UUID(str(conversation.id))) + return conversation.model_copy(update=lifecycle.model_dump()) + + conversation_router = APIRouter(prefix="/conversations", tags=["Conversations"]) # Examples @@ -88,6 +100,7 @@ @conversation_router.get("/search") async def search_conversations( + request: Request, page_id: Annotated[ str | None, Query(title="Optional next_page_id from the previously returned page"), @@ -113,6 +126,11 @@ async def search_conversations( page = await conversation_service.search_conversations( page_id, limit, status, sort_order ) + page = page.model_copy( + update={ + "items": [_with_runtime_lifecycle(request, item) for item in page.items] + } + ) if not include_skills: # ``model_copy`` rather than in-place mutation so we never # write back into whatever the upstream service handed us @@ -146,6 +164,7 @@ async def count_conversations( ) async def get_conversation( conversation_id: UUID, + request: Request, include_skills: Annotated[bool, Query(title=INCLUDE_SKILLS_PARAM_TITLE)] = False, conversation_service: ConversationService = Depends(get_conversation_service), ) -> ConversationInfo: @@ -153,11 +172,35 @@ async def get_conversation( conversation = await conversation_service.get_conversation(conversation_id) if conversation is None: raise HTTPException(status.HTTP_404_NOT_FOUND) + conversation = _with_runtime_lifecycle(request, conversation) if not include_skills: conversation = trim_conversation_response_skills(conversation) return conversation +@conversation_router.get("/{conversation_id}/runtime") +async def get_local_conversation_runtime( + conversation_id: UUID, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> ConversationRuntimeInfo: + """Inspect the always-available in-process runtime.""" + if await conversation_service.get_conversation(conversation_id) is None: + raise HTTPException(status.HTTP_404_NOT_FOUND) + return ConversationRuntimeInfo( + runtime_status=ConversationRuntimeStatus.AVAILABLE, + can_resume=True, + ) + + +@conversation_router.post("/{conversation_id}/runtime/reprovision") +async def reprovision_local_conversation_runtime( + conversation_id: UUID, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> ConversationRuntimeInfo: + """Return local runtime state; local mode has no infrastructure to provision.""" + return await get_local_conversation_runtime(conversation_id, conversation_service) + + @conversation_router.get( "/{conversation_id}/agent_final_response", responses={404: {"description": "Conversation not found"}}, @@ -181,6 +224,7 @@ async def get_conversation_agent_final_response( @conversation_router.get("") async def batch_get_conversations( + request: Request, ids: Annotated[list[UUID], Query()], include_skills: Annotated[bool, Query(title=INCLUDE_SKILLS_PARAM_TITLE)] = False, conversation_service: ConversationService = Depends(get_conversation_service), @@ -189,6 +233,10 @@ async def batch_get_conversations( any missing item""" assert len(ids) < 100 conversations = await conversation_service.batch_get_conversations(ids) + conversations = [ + _with_runtime_lifecycle(request, item) if item is not None else None + for item in conversations + ] if not include_skills: return [ trim_conversation_response_skills(c) if c is not None else None 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 e3bef71122..d0feab2c9d 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/registry.py @@ -23,6 +23,11 @@ from openhands.agent_server.config import V1_SESSION_API_KEY_ENV, Config from openhands.agent_server.docker_runtime.broker import RuntimeCredentialBroker from openhands.agent_server.docker_runtime.provisioning import RuntimeProvisioningStore +from openhands.agent_server.models import ( + ConversationRuntimeError, + ConversationRuntimeInfo, + ConversationRuntimeStatus, +) from openhands.agent_server.persistence import FileSecretsStore from openhands.agent_server.persistence.store import _get_persistence_dir from openhands.sdk.llm.auth.credentials import CredentialStore @@ -92,6 +97,7 @@ def __init__(self, config: Config) -> None: self._provisioning: RuntimeProvisioningStore | None = None self._brokers: dict[UUID, RuntimeCredentialBroker] = {} self._mutations: dict[UUID, asyncio.Lock] = {} + self._runtime_errors: dict[UUID, ConversationRuntimeError] = {} def mutation_lock(self, conversation_id: UUID) -> asyncio.Lock: return self._mutations.setdefault(conversation_id, asyncio.Lock()) @@ -153,6 +159,31 @@ def cleanup_stale_containers(self) -> None: def get(self, conversation_id: UUID) -> RunningConversationContainer | None: return self._containers.get(conversation_id) + def runtime_info(self, conversation_id: UUID) -> ConversationRuntimeInfo: + """Inspect runtime state without provisioning or contacting a container.""" + directory = self.conversation_dir(conversation_id) + has_state = (directory / "base_state.json").is_file() + has_metadata = (directory / "meta.json").is_file() + manifest = self.provisioning.manifest_path(conversation_id) + can_resume = has_state and has_metadata and manifest.is_file() + + if conversation_id in self._containers: + status = ConversationRuntimeStatus.AVAILABLE + elif conversation_id in self._starts: + status = ConversationRuntimeStatus.STARTING + elif has_state and has_metadata and not manifest.is_file(): + status = ConversationRuntimeStatus.OWNERSHIP_LOST + elif conversation_id in self._runtime_errors: + status = ConversationRuntimeStatus.ERROR + else: + status = ConversationRuntimeStatus.MISSING + + return ConversationRuntimeInfo( + runtime_status=status, + can_resume=can_resume, + runtime_error=self._runtime_errors.get(conversation_id), + ) + def items(self) -> list[tuple[UUID, RunningConversationContainer]]: return list(self._containers.items()) @@ -191,13 +222,18 @@ async def get_or_create( try: container = await asyncio.shield(task) - except Exception: + except Exception as exc: async with self._lock: if self._starts.get(conversation_id) is task: self._starts.pop(conversation_id, None) + self._runtime_errors[conversation_id] = ConversationRuntimeError( + code="runtime_start_failed", + message=str(exc), + ) raise async with self._lock: + self._runtime_errors.pop(conversation_id, None) existing = self._containers.get(conversation_id) if existing is not None: return existing, False @@ -219,6 +255,7 @@ async def stop(self, conversation_id: UUID) -> bool: async with self._lock: container = self._containers.pop(conversation_id, None) start_task = self._starts.pop(conversation_id, None) + self._runtime_errors.pop(conversation_id, None) stopped = False if container is not None: @@ -248,6 +285,7 @@ async def shutdown(self) -> None: start_tasks = list(self._starts.values()) self._containers.clear() self._starts.clear() + self._runtime_errors.clear() for task in start_tasks: try: 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 0770847bf0..5821923a04 100644 --- a/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py +++ b/openhands-agent-server/openhands/agent_server/docker_runtime/routers.py @@ -39,6 +39,7 @@ DockerConversationRegistry, RunningConversationContainer, ) +from openhands.agent_server.models import ConversationRuntimeInfo from openhands.agent_server.runtime_router import add_legacy_runtime_routes from openhands.agent_server.utils import safe_rmtree from openhands.sdk.logger import get_logger @@ -356,6 +357,56 @@ async def docker_proxy_conversation_root_mutation( ) +@docker_conversation_proxy_router.get( + "/{conversation_id}/runtime", + response_model=ConversationRuntimeInfo, +) +async def get_conversation_runtime( + conversation_id: UUID, request: Request +) -> ConversationRuntimeInfo: + """Inspect runtime availability without provisioning a container.""" + registry = get_registry(request) + info = registry.runtime_info(conversation_id) + if not registry.conversation_dir(conversation_id).joinpath("meta.json").is_file(): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Conversation not found: {conversation_id}", + ) + return info + + +@docker_conversation_proxy_router.post( + "/{conversation_id}/runtime/reprovision", + response_model=ConversationRuntimeInfo, +) +async def reprovision_conversation_runtime( + conversation_id: UUID, request: Request +) -> ConversationRuntimeInfo: + """Start missing infrastructure without resuming agent execution.""" + registry = get_registry(request) + info = registry.runtime_info(conversation_id) + if not registry.conversation_dir(conversation_id).joinpath("meta.json").is_file(): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Conversation not found: {conversation_id}", + ) + if not info.can_resume: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Conversation runtime cannot be safely reprovisioned", + ) + try: + await registry.prepare(conversation_id) + await registry.get_or_create(conversation_id) + except Exception as exc: + logger.exception("Could not reprovision conversation %s", conversation_id) + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail="Could not reprovision conversation runtime", + ) from exc + return registry.runtime_info(conversation_id) + + @docker_conversation_proxy_router.api_route( "/{conversation_id}/{tail:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"], diff --git a/openhands-agent-server/openhands/agent_server/models.py b/openhands-agent-server/openhands/agent_server/models.py index 0182ccf1aa..05311ea22b 100644 --- a/openhands-agent-server/openhands/agent_server/models.py +++ b/openhands-agent-server/openhands/agent_server/models.py @@ -45,6 +45,31 @@ from openhands.sdk.workspace.base import BaseWorkspace +class ConversationRuntimeStatus(StrEnum): + """Availability of the runtime that executes a conversation.""" + + AVAILABLE = "available" + STARTING = "starting" + MISSING = "missing" + OWNERSHIP_LOST = "ownership_lost" + ERROR = "error" + + +class ConversationRuntimeError(BaseModel): + """Structured details for the latest runtime lifecycle failure.""" + + code: str + message: str + + +class ConversationRuntimeInfo(BaseModel): + """Runtime availability and recovery information for a conversation.""" + + runtime_status: ConversationRuntimeStatus + can_resume: bool + runtime_error: ConversationRuntimeError | None = None + + class ServerErrorEvent(Event): """Event emitted by the agent server when a server-level error occurs. @@ -150,6 +175,18 @@ class _ConversationInfoBase(BaseModel): execution_status: ConversationExecutionStatus = Field( default=ConversationExecutionStatus.IDLE ) + runtime_status: ConversationRuntimeStatus = Field( + default=ConversationRuntimeStatus.AVAILABLE, + description="Availability of the runtime that executes this conversation.", + ) + can_resume: bool = Field( + default=True, + description="Whether retained state allows explicit execution resumption.", + ) + runtime_error: ConversationRuntimeError | None = Field( + default=None, + description="Latest runtime lifecycle failure, if one is known.", + ) confirmation_policy: ConfirmationPolicyBase = Field(default=NeverConfirm()) security_analyzer: SecurityAnalyzerBase | None = Field( default=None, diff --git a/tests/agent_server/docker_runtime/test_docker_routers.py b/tests/agent_server/docker_runtime/test_docker_routers.py index 3428b16175..393e2f0fe1 100644 --- a/tests/agent_server/docker_runtime/test_docker_routers.py +++ b/tests/agent_server/docker_runtime/test_docker_routers.py @@ -30,6 +30,10 @@ from openhands.agent_server.api import create_app from openhands.agent_server.config import Config from openhands.agent_server.docker_runtime.provisioning import RuntimeProvisioningStore +from openhands.agent_server.models import ( + ConversationRuntimeInfo, + ConversationRuntimeStatus, +) # --------------------------------------------------------------------------- @@ -213,6 +217,22 @@ def preregister(self, cid: UUID) -> _FakeWorkspace: def get(self, cid: UUID) -> _FakeWorkspace | None: return self._workspaces.get(cid) + def runtime_info(self, cid: UUID) -> ConversationRuntimeInfo: + directory = self.conversation_dir(cid) + can_resume = ( + (directory / "meta.json").is_file() + and (directory / "base_state.json").is_file() + and self.provisioning.manifest_path(cid).is_file() + ) + return ConversationRuntimeInfo( + runtime_status=( + ConversationRuntimeStatus.AVAILABLE + if cid in self._workspaces + else ConversationRuntimeStatus.MISSING + ), + can_resume=can_resume, + ) + def conversation_dir(self, cid: UUID) -> Path: return self.conversations_dir / cid.hex @@ -382,6 +402,59 @@ def test_global_router_404_for_unknown_cid(docker_app): # --------------------------------------------------------------------------- +def test_runtime_inspection_does_not_start_container(docker_app): + client, app = docker_app + registry = app.state.docker_registry + cid = uuid4() + registry.provisioning.create(cid) + directory = registry.conversation_dir(cid) + directory.mkdir(parents=True) + (directory / "meta.json").write_text("{}") + (directory / "base_state.json").write_text("{}") + + response = client.get(f"/api/conversations/{cid}/runtime") + + assert response.status_code == 200 + assert response.json() == { + "runtime_status": "missing", + "can_resume": True, + "runtime_error": None, + } + assert registry.get(cid) is None + + +def test_runtime_reprovision_starts_infrastructure_without_run(docker_app): + client, app = docker_app + registry = app.state.docker_registry + cid = uuid4() + registry.provisioning.create(cid) + directory = registry.conversation_dir(cid) + directory.mkdir(parents=True) + (directory / "meta.json").write_text("{}") + (directory / "base_state.json").write_text("{}") + + response = client.post(f"/api/conversations/{cid}/runtime/reprovision") + + assert response.status_code == 200 + assert response.json()["runtime_status"] == "available" + assert registry.get(cid) is not None + + +def test_runtime_reprovision_rejects_ownership_loss(docker_app): + client, app = docker_app + registry = app.state.docker_registry + cid = uuid4() + directory = registry.conversation_dir(cid) + directory.mkdir(parents=True) + (directory / "meta.json").write_text("{}") + (directory / "base_state.json").write_text("{}") + + response = client.post(f"/api/conversations/{cid}/runtime/reprovision") + + assert response.status_code == 409 + assert registry.get(cid) is None + + def test_metadata_routes_are_mounted_locally_in_docker_mode(tmp_path): """``GET /api/conversations``, ``/api/conversations/count``, and ``/api/conversations/search`` must come from the LOCAL conversation diff --git a/tests/agent_server/docker_runtime/test_registry.py b/tests/agent_server/docker_runtime/test_registry.py index 36325617b0..baba2cc339 100644 --- a/tests/agent_server/docker_runtime/test_registry.py +++ b/tests/agent_server/docker_runtime/test_registry.py @@ -14,6 +14,7 @@ DockerConversationRegistry, RunningConversationContainer, ) +from openhands.agent_server.models import ConversationRuntimeStatus def _container(conversation_id: UUID) -> RunningConversationContainer: @@ -25,9 +26,135 @@ def _container(conversation_id: UUID) -> RunningConversationContainer: ) +def _persisted_conversation(registry: DockerConversationRegistry, cid: UUID) -> None: + registry.provisioning.create(cid) + directory = registry.conversation_dir(cid) + directory.mkdir(parents=True) + (directory / "meta.json").write_text("{}") + (directory / "base_state.json").write_text("{}") + + +def test_runtime_info_does_not_provision_missing_runtime(tmp_path): + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) + cid = uuid4() + _persisted_conversation(registry, cid) + + info = registry.runtime_info(cid) + + assert info.runtime_status == ConversationRuntimeStatus.MISSING + assert info.can_resume is True + assert registry.get(cid) is None + + +def test_runtime_info_distinguishes_available_and_ownership_lost(tmp_path): + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) + available = uuid4() + _persisted_conversation(registry, available) + registry._containers[available] = _container(available) + assert ( + registry.runtime_info(available).runtime_status + == ConversationRuntimeStatus.AVAILABLE + ) + + ownership_lost = uuid4() + directory = registry.conversation_dir(ownership_lost) + directory.mkdir(parents=True) + (directory / "meta.json").write_text("{}") + (directory / "base_state.json").write_text("{}") + info = registry.runtime_info(ownership_lost) + assert info.runtime_status == ConversationRuntimeStatus.OWNERSHIP_LOST + assert info.can_resume is False + + +@pytest.mark.asyncio +async def test_runtime_info_retains_start_failure(tmp_path): + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) + cid = uuid4() + _persisted_conversation(registry, cid) + + def fail(conversation_id: UUID) -> RunningConversationContainer: + raise RuntimeError(f"container failed: {conversation_id}") + + registry._build_container = fail + with pytest.raises(RuntimeError, match="container failed"): + await registry.get_or_create(cid) + + info = registry.runtime_info(cid) + assert info.runtime_status == ConversationRuntimeStatus.ERROR + assert info.can_resume is True + assert info.runtime_error is not None + assert info.runtime_error.code == "runtime_start_failed" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("shutdown", [False, True]) +async def test_stop_clears_runtime_start_failure(tmp_path, shutdown): + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) + cid = uuid4() + _persisted_conversation(registry, cid) + + def fail(conversation_id: UUID) -> RunningConversationContainer: + raise RuntimeError("container failed") + + registry._build_container = fail + with pytest.raises(RuntimeError, match="container failed"): + await registry.get_or_create(cid) + assert registry.runtime_info(cid).runtime_status == ConversationRuntimeStatus.ERROR + if shutdown: + await registry.shutdown() + else: + await registry.stop(cid) + info = registry.runtime_info(cid) + assert info.runtime_status == ConversationRuntimeStatus.MISSING + assert info.can_resume is True + assert info.runtime_error is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("shutdown", [False, True]) +async def test_stopped_start_cannot_restore_runtime_error(tmp_path, shutdown): + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) + cid = uuid4() + _persisted_conversation(registry, cid) + started = threading.Event() + release = threading.Event() + + def fail(conversation_id: UUID) -> RunningConversationContainer: + started.set() + assert release.wait(timeout=5) + raise RuntimeError("container failed") + + registry._build_container = fail + start_task = asyncio.create_task(registry.get_or_create(cid)) + assert await asyncio.to_thread(started.wait, 5) + stop_task = asyncio.create_task( + registry.shutdown() if shutdown else registry.stop(cid) + ) + await asyncio.sleep(0) + release.set() + with pytest.raises(RuntimeError, match="container failed"): + await start_task + await stop_task + info = registry.runtime_info(cid) + assert info.runtime_status == ConversationRuntimeStatus.MISSING + assert info.runtime_error is None + + @pytest.mark.asyncio async def test_get_or_create_deduplicates_same_conversation_start(tmp_path): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) conversation_id = uuid4() calls = 0 @@ -51,7 +178,9 @@ def build(conversation_id: UUID) -> RunningConversationContainer: @pytest.mark.asyncio async def test_get_or_create_starts_different_conversations_concurrently(tmp_path): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) entered: set[UUID] = set() entered_lock = threading.Lock() release = threading.Event() @@ -124,7 +253,9 @@ def cleanup(target: RunningConversationContainer) -> None: @pytest.mark.asyncio async def test_failed_start_can_be_retried(tmp_path): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) conversation_id = uuid4() calls = 0 @@ -188,7 +319,9 @@ def run_container(**kwargs) -> RunningConversationContainer: def test_run_container_uses_host_identity_for_bind_mounts(tmp_path, monkeypatch): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) commands: list[list[str]] = [] def execute(command, **kwargs): @@ -279,7 +412,9 @@ def execute(command, **kwargs): def test_cleanup_stale_containers_is_scoped_to_registry_owner(tmp_path, monkeypatch): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) commands: list[list[str]] = [] def execute(command, **kwargs): @@ -339,7 +474,9 @@ def test_docker_launch_preserves_explicit_credentials(tmp_path, monkeypatch): docker.chmod(0o755) monkeypatch.setenv("PATH", f"{tmp_path}{os.pathsep}{os.environ['PATH']}") monkeypatch.setenv("OH_SESSION_API_KEYS_1", "not-forwarded") - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) container = registry._run_container( conversation_id=uuid4(), image="test-image", @@ -354,7 +491,9 @@ def test_docker_launch_preserves_explicit_credentials(tmp_path, monkeypatch): @pytest.mark.asyncio async def test_cancelled_start_waiter_does_not_cancel_other_waiters(tmp_path): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) cid = uuid4() entered = threading.Event() release = threading.Event() @@ -391,7 +530,9 @@ def build(conversation_id): async def test_lone_cancelled_waiter_can_recover_or_shutdown( tmp_path, recover, monkeypatch ): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) cid = uuid4() entered = threading.Event() release = threading.Event() @@ -427,7 +568,9 @@ def build(conversation_id): @pytest.mark.parametrize("binding", ["", "0.0.0.0:32123", "127.0.0.1:0"]) def test_invalid_assigned_port_cleans_up_container(tmp_path, monkeypatch, binding): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) commands = [] def execute(command, **kwargs): @@ -457,7 +600,9 @@ def execute(command, **kwargs): def test_failed_docker_run_surfaces_stderr(tmp_path, monkeypatch): - registry = DockerConversationRegistry(Config(conversations_path=tmp_path)) + registry = DockerConversationRegistry( + Config(conversations_path=tmp_path, secret_key=SecretStr("test-key")) + ) def execute(command, **kwargs): return subprocess.CompletedProcess(