diff --git a/openhands-sdk/openhands/sdk/conversation/impl/remote_conversation.py b/openhands-sdk/openhands/sdk/conversation/impl/remote_conversation.py index 07df8b4497..970788ab7e 100644 --- a/openhands-sdk/openhands/sdk/conversation/impl/remote_conversation.py +++ b/openhands-sdk/openhands/sdk/conversation/impl/remote_conversation.py @@ -7,7 +7,7 @@ import uuid from collections.abc import Callable, Mapping from queue import Empty, Queue -from typing import TYPE_CHECKING, Final, SupportsIndex, overload +from typing import TYPE_CHECKING, Final, Self, SupportsIndex, overload from urllib.parse import urlparse import httpx @@ -19,6 +19,7 @@ if TYPE_CHECKING: + from openhands.sdk.conversation.request import StartConversationRequest from openhands.sdk.tool.schema import Action, Observation from openhands.sdk.conversation.conversation_stats import ConversationStats from openhands.sdk.conversation.events_list_base import EventsListBase @@ -63,7 +64,7 @@ logger = get_logger(__name__) -LEGACY_CONVERSATIONS_PATH = "/api/conversations" +CONVERSATIONS_PATH = "/api/conversations" FATAL_WS_CLOSE_CODES = frozenset({4001, 4004}) _WEBSOCKET_AUTH_TYPE: Final = "auth" _WEBSOCKET_SESSION_API_KEY_FIELD: Final = "session_api_key" @@ -304,7 +305,7 @@ def __init__( self, client: httpx.Client, conversation_id: str, - events_base_path: str = LEGACY_CONVERSATIONS_PATH, + events_base_path: str = CONVERSATIONS_PATH, ): self._client = client self._conversation_id = conversation_id @@ -512,8 +513,8 @@ def __init__( self, client: httpx.Client, conversation_id: str, - conversation_info_base_path: str = LEGACY_CONVERSATIONS_PATH, - events_base_path: str = LEGACY_CONVERSATIONS_PATH, + conversation_info_base_path: str = CONVERSATIONS_PATH, + events_base_path: str = CONVERSATIONS_PATH, ): self._client = client self._conversation_id = conversation_id @@ -699,8 +700,6 @@ class RemoteConversation(BaseConversation): _cleanup_initiated: bool _terminal_status_queue: Queue[str] _run_armed: threading.Event - _conversation_info_base_path: str - _conversation_action_base_path: str delete_on_close: bool = False def __init__( @@ -765,18 +764,6 @@ def __init__( observability_span_name: Optional child span name for observability backends. The root span remains named "conversation". """ - super().__init__() # Initialize base class with span tracking - self.agent = agent - self._callbacks = callbacks or [] - self.max_iteration_per_run = max_iteration_per_run - self.workspace = workspace - self._client = workspace.client - self._conversation_info_base_path = LEGACY_CONVERSATIONS_PATH - self._conversation_action_base_path = LEGACY_CONVERSATIONS_PATH - self._cleanup_initiated = False - self._terminal_status_queue: Queue[str] = Queue() - self._run_armed = threading.Event() - # Client tool specs the server already has persisted for this # conversation (populated when re-attaching to an existing one). These # must be registered locally before the initial event sync so that @@ -787,9 +774,9 @@ def __init__( if conversation_id is not None: # Try to attach to existing conversation resp = _send_request( - self._client, + workspace.client, "GET", - f"{self._conversation_info_base_path}/{conversation_id}", + f"{CONVERSATIONS_PATH}/{conversation_id}", acceptable_status_codes={404}, ) if resp.status_code == 404: @@ -808,8 +795,6 @@ def __init__( attached_client_tools.append( ClientToolSpec.model_validate(raw_spec) ) - # Conversation exists, use the provided ID - self._id = conversation_id if should_create: # Import here to avoid circular imports @@ -832,7 +817,7 @@ def __init__( "stuck_detection": stuck_detection, # We need to convert RemoteWorkspace to LocalWorkspace for the server "workspace": LocalWorkspace( - working_dir=self.workspace.working_dir + working_dir=workspace.working_dir ).model_dump(), # Include tool module qualnames for dynamic registration on server "tool_module_qualnames": tool_qualnames, @@ -874,9 +859,9 @@ def __init__( if conversation_id is not None: payload["conversation_id"] = str(conversation_id) resp = _send_request( - self._client, + workspace.client, "POST", - self._conversation_info_base_path, + CONVERSATIONS_PATH, json=payload, ) data = resp.json() @@ -886,29 +871,162 @@ def __init__( raise RuntimeError( "Invalid response from server: missing conversation id" ) - self._id = uuid.UUID(cid) + conversation_id = uuid.UUID(cid) + + workspace.register_conversation(str(conversation_id)) + + assert conversation_id is not None + self._initialize_connection( + agent=agent, + workspace=workspace, + conversation_id=conversation_id, + callbacks=callbacks, + max_iteration_per_run=max_iteration_per_run, + client_tools=[*(client_tools or []), *attached_client_tools], + visualizer=visualizer, + ) + + # Initialize secrets if provided + if secrets: + # Convert dict[str, str] to dict[str, SecretValue] + secret_values: dict[str, SecretValue] = {k: v for k, v in secrets.items()} + self.update_secrets(secret_values) + + self._start_observability_span( + str(self._id), + span_name=observability_span_name, + user_id=user_id, + metadata=observability_metadata, + tags=observability_tags, + conversation_tags=tags, + ) + # All hooks (including SessionStart/SessionEnd) are executed server-side. + # hook_config is sent in the creation payload. + self.delete_on_close = delete_on_close + + @classmethod + def create( + cls, + workspace: RemoteWorkspace, + request: "StartConversationRequest", + *, + callbacks: list[ConversationCallbackType] | None = None, + visualizer: ( + type[ConversationVisualizerBase] | ConversationVisualizerBase | None + ) = DefaultConversationVisualizer, + ) -> Self: + """Submit a creation request and connect to the returned conversation. + + The request selects the agent or saved server profile and all server + options. A supplied conversation ID follows the server's idempotency + contract; this method does not probe for an existing conversation. + """ + response = _send_request( + workspace.client, + "POST", + CONVERSATIONS_PATH, + json=request.model_dump( + mode="json", exclude_none=True, context={"expose_secrets": True} + ), + ) + info = response.json() + workspace.register_conversation(info["id"]) + conversation = cls._from_info(workspace, info, callbacks, visualizer) + conversation._start_observability_span( + str(conversation.id), + span_name=request.observability_span_name, + user_id=request.user_id, + metadata=request.observability_metadata, + tags=request.observability_tags, + conversation_tags=request.tags, + ) + return conversation + + @classmethod + def attach( + cls, + workspace: RemoteWorkspace, + conversation_id: ConversationID, + *, + callbacks: list[ConversationCallbackType] | None = None, + visualizer: ( + type[ConversationVisualizerBase] | ConversationVisualizerBase | None + ) = DefaultConversationVisualizer, + ) -> Self: + """Connect using the saved agent; never create or update a conversation. + + Missing or inaccessible conversations raise an HTTP error. + """ + response = _send_request( + workspace.client, "GET", f"{CONVERSATIONS_PATH}/{conversation_id}" + ) + conversation = cls._from_info(workspace, response.json(), callbacks, visualizer) + conversation._start_observability_span(str(conversation.id)) + return conversation - workspace.register_conversation(str(self._id)) + @classmethod + def _from_info( + cls, + workspace: RemoteWorkspace, + info: dict, + callbacks: list[ConversationCallbackType] | None, + visualizer: ( + type[ConversationVisualizerBase] | ConversationVisualizerBase | None + ), + ) -> Self: + conversation = cls.__new__(cls) + conversation._initialize_connection( + agent=_validate_remote_agent(info["agent"]), + workspace=workspace, + conversation_id=uuid.UUID(info["id"]), + callbacks=callbacks, + max_iteration_per_run=info["max_iterations"], + client_tools=[ + ClientToolSpec.model_validate(spec) + for spec in info.get("client_tools") or [] + ], + visualizer=visualizer, + ) + return conversation + + def _initialize_connection( + self, + *, + agent: AgentBase, + workspace: RemoteWorkspace, + conversation_id: ConversationID, + callbacks: list[ConversationCallbackType] | None, + max_iteration_per_run: int, + client_tools: list[ClientToolSpec], + visualizer: ( + type[ConversationVisualizerBase] | ConversationVisualizerBase | None + ), + ) -> None: + super().__init__() # Initialize base class with span tracking + self.agent = agent + self._callbacks = callbacks or [] + self.max_iteration_per_run = max_iteration_per_run + self.workspace = workspace + self._client = workspace.client + self._cleanup_initiated = False + self._terminal_status_queue: Queue[str] = Queue() + self._run_armed = threading.Event() + self._id = conversation_id # Register client tool action types locally so WebSocket/persisted # events with ClientAction_* action_type can be deserialized by the # event loop. This must cover both the specs the caller passed in and # the specs the server already had persisted (when re-attaching), so a # plain reattach by conversation_id can still sync persisted events. seen_client_tool_names: set[str] = set() - for spec in [*(client_tools or []), *attached_client_tools]: + for spec in client_tools: if spec.name in seen_client_tool_names: continue seen_client_tool_names.add(spec.name) ClientTool.from_spec(spec) # Initialize the remote state - self._state = RemoteState( - self._client, - str(self._id), - conversation_info_base_path=self._conversation_info_base_path, - events_base_path=self._conversation_action_base_path, - ) + self._state = RemoteState(self._client, str(self._id)) # Add default callback to maintain local event state default_callback = self._state.events.create_default_callback() @@ -1034,24 +1152,6 @@ def run_complete_callback(event: Event) -> None: # This is the "reconciliation" part of the subscription handshake. self._state.events.reconcile() - # Initialize secrets if provided - if secrets: - # Convert dict[str, str] to dict[str, SecretValue] - secret_values: dict[str, SecretValue] = {k: v for k, v in secrets.items()} - self.update_secrets(secret_values) - - self._start_observability_span( - str(self._id), - span_name=observability_span_name, - user_id=user_id, - metadata=observability_metadata, - tags=observability_tags, - conversation_tags=tags, - ) - # All hooks (including SessionStart/SessionEnd) are executed server-side. - # hook_config is sent in the creation payload. - self.delete_on_close = delete_on_close - def _create_llm_completion_log_callback(self) -> ConversationCallbackType: """Create a callback that writes LLM completion logs to client filesystem.""" @@ -1124,7 +1224,7 @@ def send_message(self, message: str | Message, sender: str | None = None) -> Non _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/events", + f"{CONVERSATIONS_PATH}/{self._id}/events", json=payload, ) @@ -1162,7 +1262,7 @@ def run( resp = _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/run", + f"{CONVERSATIONS_PATH}/{self._id}/run", acceptable_status_codes={200, 201, 204, 409}, timeout=30, # Short timeout for trigger request ) @@ -1333,7 +1433,7 @@ def _poll_status_once(self) -> str | None: resp = _send_request( self._client, "GET", - f"{self._conversation_info_base_path}/{self._id}", + f"{CONVERSATIONS_PATH}/{self._id}", timeout=30, ) info = resp.json() @@ -1406,7 +1506,7 @@ def set_confirmation_policy(self, policy: ConfirmationPolicyBase) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/confirmation_policy", + f"{CONVERSATIONS_PATH}/{self._id}/confirmation_policy", json=payload, ) @@ -1420,7 +1520,7 @@ def set_security_analyzer(self, analyzer: SecurityAnalyzerBase | None) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/security_analyzer", + f"{CONVERSATIONS_PATH}/{self._id}/security_analyzer", json=payload, ) @@ -1429,10 +1529,7 @@ def reject_pending_actions(self, reason: str = "User rejected the action") -> No _send_request( self._client, "POST", - ( - f"{self._conversation_action_base_path}/{self._id}" - "/events/respond_to_confirmation" - ), + (f"{CONVERSATIONS_PATH}/{self._id}/events/respond_to_confirmation"), json={"accept": False, "reason": reason}, ) @@ -1440,21 +1537,21 @@ def pause(self) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/pause", + f"{CONVERSATIONS_PATH}/{self._id}/pause", ) def interrupt(self) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/interrupt", + f"{CONVERSATIONS_PATH}/{self._id}/interrupt", ) def load_plugin(self, plugin_ref: str) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/load_plugin", + f"{CONVERSATIONS_PATH}/{self._id}/load_plugin", json={"plugin_ref": plugin_ref}, ) @@ -1479,7 +1576,7 @@ def update_secrets(self, secrets: Mapping[str, SecretValue]) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/secrets", + f"{CONVERSATIONS_PATH}/{self._id}/secrets", json=payload, ) @@ -1504,12 +1601,21 @@ def ask_agent(self, question: str) -> str: resp = _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/ask_agent", + f"{CONVERSATIONS_PATH}/{self._id}/ask_agent", json=payload, ) data = resp.json() return data["response"] + def set_title(self, title: str) -> None: + """Set the persisted display title without running the agent.""" + _send_request( + self._client, + "PATCH", + f"{CONVERSATIONS_PATH}/{self._id}", + json={"title": title}, + ) + @observe( name="conversation.generate_title", ignore_inputs=["llm"], @@ -1551,7 +1657,7 @@ def condense(self) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/condense", + f"{CONVERSATIONS_PATH}/{self._id}/condense", ) def fork( @@ -1611,7 +1717,7 @@ def fork( resp = _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/fork", + f"{CONVERSATIONS_PATH}/{self._id}/fork", json=body, ) fork_info = resp.json() @@ -1653,7 +1759,7 @@ def navigate_to(self, event_id: EventID | None) -> None: _send_request( self._client, "POST", - f"{self._conversation_action_base_path}/{self._id}/navigate", + f"{CONVERSATIONS_PATH}/{self._id}/navigate", json={"event_id": event_id}, ) self._state.refresh_from_server() @@ -1728,7 +1834,7 @@ def close(self) -> None: _send_request( self._client, "DELETE", - f"{self._conversation_action_base_path}/{self.id}", + f"{CONVERSATIONS_PATH}/{self.id}", ) except Exception: pass diff --git a/openhands-sdk/openhands/sdk/workspace/remote/async_remote_workspace.py b/openhands-sdk/openhands/sdk/workspace/remote/async_remote_workspace.py index 814d6f354e..fa14f08b4d 100644 --- a/openhands-sdk/openhands/sdk/workspace/remote/async_remote_workspace.py +++ b/openhands-sdk/openhands/sdk/workspace/remote/async_remote_workspace.py @@ -1,6 +1,7 @@ from collections.abc import Generator from pathlib import Path -from typing import Any +from types import TracebackType +from typing import Any, Self from urllib.request import urlopen import httpx @@ -16,6 +17,25 @@ class AsyncRemoteWorkspace(RemoteWorkspaceMixin): _client: httpx.AsyncClient | None = PrivateAttr(default=None) + async def __aenter__(self) -> Self: + """Enter a workspace whose HTTP pool is closed on context exit.""" + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Close the client without releasing the server's runtime.""" + await self.reset_client() + + async def get_server_info(self) -> dict[str, Any]: + """Return server metadata, matching RemoteWorkspace.get_server_info.""" + response = await self.client.get("/server_info") + response.raise_for_status() + return response.json() + async def reset_client(self) -> None: """Reset the HTTP client to force re-initialization. @@ -54,6 +74,29 @@ async def _execute(self, generator: Generator[dict[str, Any], httpx.Response, An except StopIteration as e: return e.value + async def start_command( + self, + command: str, + cwd: str | Path | None = None, + timeout: float = 30, + ) -> str: + """Start a command and return its ID without waiting for completion.""" + return await self._execute(self._start_command_generator(command, cwd, timeout)) + + async def get_command_output( + self, command_id: str | None = None + ) -> dict[str, Any] | None: + """Read the latest output; a missing exit code means it is still running.""" + return await self._execute(self._get_command_output_generator(command_id)) + + async def get_runtime_session_key(self) -> str: + """Get the scoped worker credential for this conversation runtime.""" + return await self._execute(self._runtime_lifecycle_generator(release=False)) + + async def release_runtime(self) -> None: + """Release execution resources while retaining conversation history.""" + await self._execute(self._runtime_lifecycle_generator(release=True)) + async def execute_command( self, command: str, @@ -79,7 +122,7 @@ async def execute_command( async def file_upload( self, - source_path: str | Path, + source_path: str | Path | bytes, destination_path: str | Path, ) -> FileOperationResult: """Upload a file to the remote system. @@ -87,7 +130,7 @@ async def file_upload( Reads the local file and sends it to the remote system via HTTP API. Args: - source_path: Path to the local source file + source_path: Local file path or in-memory bytes destination_path: Path where the file should be uploaded on remote system Returns: diff --git a/openhands-sdk/openhands/sdk/workspace/remote/base.py b/openhands-sdk/openhands/sdk/workspace/remote/base.py index def562a86c..31e0a7ba15 100644 --- a/openhands-sdk/openhands/sdk/workspace/remote/base.py +++ b/openhands-sdk/openhands/sdk/workspace/remote/base.py @@ -130,6 +130,29 @@ def get_server_info(self) -> dict[str, Any]: assert isinstance(data, dict) return data + def start_command( + self, + command: str, + cwd: str | Path | None = None, + timeout: float = 30, + ) -> str: + """Start a command and return its ID without waiting for completion.""" + return self._execute(self._start_command_generator(command, cwd, timeout)) + + def get_command_output( + self, command_id: str | None = None + ) -> dict[str, Any] | None: + """Read the latest output; a missing exit code means it is still running.""" + return self._execute(self._get_command_output_generator(command_id)) + + def get_runtime_session_key(self) -> str: + """Get the scoped worker credential for this conversation runtime.""" + return self._execute(self._runtime_lifecycle_generator(release=False)) + + def release_runtime(self) -> None: + """Release execution resources while retaining conversation history.""" + self._execute(self._runtime_lifecycle_generator(release=True)) + def execute_command( self, command: str, @@ -155,7 +178,7 @@ def execute_command( def file_upload( self, - source_path: str | Path, + source_path: str | Path | bytes, destination_path: str | Path, ) -> FileOperationResult: """Upload a file to the remote system. @@ -163,7 +186,7 @@ def file_upload( Reads the local file and sends it to the remote system via HTTP API. Args: - source_path: Path to the local source file + source_path: Local file path or in-memory bytes destination_path: Path where the file should be uploaded on remote system Returns: diff --git a/openhands-sdk/openhands/sdk/workspace/remote/remote_workspace_mixin.py b/openhands-sdk/openhands/sdk/workspace/remote/remote_workspace_mixin.py index cfb674485a..bfea81ca1f 100644 --- a/openhands-sdk/openhands/sdk/workspace/remote/remote_workspace_mixin.py +++ b/openhands-sdk/openhands/sdk/workspace/remote/remote_workspace_mixin.py @@ -3,6 +3,7 @@ from collections.abc import Generator from pathlib import Path, PureWindowsPath from typing import Any +from uuid import UUID import httpx from pydantic import BaseModel, Field, TypeAdapter @@ -52,6 +53,19 @@ class RemoteWorkspaceMixin(BaseModel): "None means no limit, useful for running many conversations in parallel.", ) + runtime_conversation_id: UUID | None = Field( + default=None, + frozen=True, + description="Conversation runtime scope; None uses the host workspace.", + ) + + @property + def api_prefix(self) -> str: + """The immutable runtime scope used by file, command, and Git operations.""" + if self.runtime_conversation_id is None: + return "/api" + return f"/api/conversations/{self.runtime_conversation_id}" + def model_post_init(self, context: Any) -> None: # Set up remote host self.host = self.host.rstrip("/") @@ -64,6 +78,85 @@ def _headers(self): headers["X-Session-API-Key"] = self.api_key return headers + def _start_command_generator( + self, + command: str, + cwd: str | Path | None, + timeout: float, + ) -> Generator[dict[str, Any], httpx.Response, str]: + payload: dict[str, Any] = {"command": command, "timeout": int(timeout)} + if cwd is not None: + payload["cwd"] = _remote_path(cwd) + response = yield { + "method": "POST", + "url": f"{self.host}{self.api_prefix}/bash/start_bash_command", + "json": payload, + "headers": self._headers, + "timeout": timeout + 5, + } + response.raise_for_status() + command_id = response.json().get("id") + if not isinstance(command_id, str) or not command_id: + raise ValueError("Agent Server returned no background command ID") + return command_id + + def _search_command_output_generator( + self, + command_id: str | None, + *, + after_order: int | None = None, + timeout: float = 60, + ) -> Generator[dict[str, Any], httpx.Response, dict[str, Any]]: + params: dict[str, str | int] = { + "kind__eq": "BashOutput", + "sort_order": "TIMESTAMP_DESC" if after_order is None else "TIMESTAMP", + "limit": 1 if after_order is None else 100, + } + if command_id is not None: + params["command_id__eq"] = command_id + if after_order is not None and after_order >= 0: + params["order__gt"] = after_order + response = yield { + "method": "GET", + "url": f"{self.host}{self.api_prefix}/bash/bash_events/search", + "params": params, + "headers": self._headers, + "timeout": timeout, + } + response.raise_for_status() + return response.json() + + def _get_command_output_generator( + self, + command_id: str | None, + ) -> Generator[dict[str, Any], httpx.Response, dict[str, Any] | None]: + page = yield from self._search_command_output_generator(command_id) + return next(iter(page.get("items", [])), None) + + def _runtime_lifecycle_generator( + self, + *, + release: bool, + ) -> Generator[dict[str, Any], httpx.Response, str | None]: + if self.runtime_conversation_id is None: + raise ValueError("Runtime lifecycle requires a conversation scope") + response = yield { + "method": "DELETE" if release else "POST", + "url": f"{self.host}{self.api_prefix}/runtime" + + ("" if release else "/credentials"), + "headers": self._headers, + "timeout": 60, + } + if release and response.status_code == 404: + return None + response.raise_for_status() + if not release: + key = response.json().get("session_api_key") + if not isinstance(key, str) or not key: + raise ValueError("Runtime returned an empty session credential") + return key + return None + def _execute_command_generator( self, command: str, @@ -85,26 +178,8 @@ def _execute_command_generator( """ _logger.debug("Executing remote command") - # Step 1: Start the bash command - payload = { - "command": command, - "timeout": int(timeout), - } - if cwd is not None: - payload["cwd"] = _remote_path(cwd) - try: - # Start the command - response: httpx.Response = yield { - "method": "POST", - "url": f"{self.host}/api/bash/start_bash_command", - "json": payload, - "headers": self._headers, - "timeout": timeout + 5.0, # Add buffer to HTTP timeout - } - response.raise_for_status() - bash_command = response.json() - command_id = bash_command["id"] + command_id = yield from self._start_command_generator(command, cwd, timeout) _logger.debug(f"Started command with ID: {command_id}") @@ -117,25 +192,11 @@ def _execute_command_generator( seen_event_ids: set[str] = set() # Track seen IDs to detect duplicates while time.time() - start_time < timeout: - # Search for new events (order > last_order) - params: dict[str, str | int] = { - "command_id__eq": command_id, - "sort_order": "TIMESTAMP", - "limit": 100, - "kind__eq": "BashOutput", - } - if last_order >= 0: - params["order__gt"] = last_order - - response = yield { - "method": "GET", - "url": f"{self.host}/api/bash/bash_events/search", - "params": params, - "headers": self._headers, - "timeout": timeout, - } - response.raise_for_status() - search_result = response.json() + search_result = yield from self._search_command_output_generator( + command_id, + after_order=last_order, + timeout=timeout, + ) # Process BashOutput events for event in search_result.get("items", []): @@ -208,7 +269,7 @@ def _execute_command_generator( def _file_upload_generator( self, - source_path: str | Path, + source_path: str | Path | bytes, destination_path: str | Path, ) -> Generator[dict[str, Any], httpx.Response, FileOperationResult]: """Upload a file to the remote system. @@ -222,7 +283,7 @@ def _file_upload_generator( Returns: FileOperationResult: Result with success status and metadata """ - source = Path(source_path) + source = Path("upload") if isinstance(source_path, bytes) else Path(source_path) destination = Path(destination_path) destination_remote = _remote_path(destination_path) @@ -230,8 +291,11 @@ def _file_upload_generator( try: # Read the file content - with open(source, "rb") as f: - file_content = f.read() + if isinstance(source_path, bytes): + file_content = source_path + else: + with open(source, "rb") as f: + file_content = f.read() # Prepare the upload files = {"file": (source.name, file_content)} @@ -239,7 +303,7 @@ def _file_upload_generator( # Make HTTP call using query parameter for path response: httpx.Response = yield { "method": "POST", - "url": f"{self.host}/api/file/upload", + "url": f"{self.host}{self.api_prefix}/file/upload", "params": {"path": destination_remote}, "files": files, "headers": self._headers, @@ -292,7 +356,7 @@ def _file_download_generator( # Make HTTP call using query parameter for path response = yield { "method": "GET", - "url": "/api/file/download", + "url": f"{self.api_prefix}/file/download", "params": {"path": source_remote}, "headers": self._headers, "timeout": 60.0, @@ -340,7 +404,7 @@ def _git_changes_generator( remote_path = _join_remote_path(self.working_dir, path) response = yield { "method": "GET", - "url": "/api/git/changes", + "url": f"{self.api_prefix}/git/changes", "params": {"path": remote_path}, "headers": self._headers, "timeout": 60.0, @@ -368,7 +432,7 @@ def _git_diff_generator( remote_path = _join_remote_path(self.working_dir, path) response = yield { "method": "GET", - "url": "/api/git/diff", + "url": f"{self.api_prefix}/git/diff", "params": {"path": remote_path}, "headers": self._headers, "timeout": 60.0, diff --git a/tests/sdk/conversation/remote/test_remote_conversation.py b/tests/sdk/conversation/remote/test_remote_conversation.py index bb05f37e5d..ed9b6cdf6e 100644 --- a/tests/sdk/conversation/remote/test_remote_conversation.py +++ b/tests/sdk/conversation/remote/test_remote_conversation.py @@ -16,6 +16,7 @@ WebSocketConnectionError, ) from openhands.sdk.conversation.impl.remote_conversation import RemoteConversation +from openhands.sdk.conversation.request import StartConversationRequest from openhands.sdk.conversation.secret_registry import SecretValue from openhands.sdk.conversation.visualizer import DefaultConversationVisualizer from openhands.sdk.event import MessageEvent @@ -25,9 +26,10 @@ ConversationStateUpdateEvent, ) from openhands.sdk.event.llm_completion_log import LLMCompletionLogEvent +from openhands.sdk.hooks import HookConfig from openhands.sdk.llm import LLM, Message, Metrics, TextContent from openhands.sdk.security.confirmation_policy import AlwaysConfirm -from openhands.sdk.workspace import RemoteWorkspace +from openhands.sdk.workspace import LocalWorkspace, RemoteWorkspace class TestRemoteConversation: @@ -171,6 +173,137 @@ def custom_side_effect(method, url, **kwargs): mock_client_instance.request.side_effect = custom_side_effect return ws_callback + @pytest.mark.parametrize("kind", ["Agent", "ACPAgent"]) + @patch( + "openhands.sdk.conversation.impl.remote_conversation.WebSocketCallbackClient" + ) + def test_attach_uses_server_agent_without_creating_or_changing_settings( + self, mock_ws_client, kind + ): + cid = uuid.uuid4() + client = self.setup_mock_client(str(cid)) + original = client.request.side_effect + expected = self.agent if kind == "Agent" else ACPAgent(acp_command=["test-acp"]) + + def respond(method, url, **kwargs): + response = original(method, url, **kwargs) + if method == "GET" and url == f"/api/conversations/{cid}": + response.json.return_value["agent"] = expected.model_dump(mode="json") + response.json.return_value["max_iterations"] = 500 + return response + + client.request.side_effect = respond + conversation = RemoteConversation.attach( + workspace=self.workspace, conversation_id=cid, visualizer=None + ) + assert type(conversation.agent) is type(expected) + if kind == "Agent": + assert conversation.agent.llm.model == expected.llm.model + assert conversation.agent.tools == expected.tools + else: + assert isinstance(conversation.agent, ACPAgent) + assert isinstance(expected, ACPAgent) + assert conversation.agent.acp_command == expected.acp_command + assert conversation.id == cid + conversation.close() + assert all(call.args[0] == "GET" for call in client.request.call_args_list) + client.close.assert_not_called() # The caller still owns the workspace. + + @patch( + "openhands.sdk.conversation.impl.remote_conversation.WebSocketCallbackClient" + ) + def test_create_from_profile_uses_resolved_agent(self, mock_ws_client): + cid, profile_id = uuid.uuid4(), uuid.uuid4() + hooks = HookConfig.model_validate({"stop": [{"hooks": [{"command": "true"}]}]}) + client = self.setup_mock_client(str(cid)) + original = client.request.side_effect + created = False + + def respond(method, url, **kwargs): + nonlocal created + if method == "GET" and url == f"/api/conversations/{cid}" and not created: + return httpx.Response( + 404, request=httpx.Request(method, self.host + url) + ) + response = original(method, url, **kwargs) + if method == "POST" and url == "/api/conversations": + payload = kwargs["json"] + assert payload["agent_profile_id"] == str(profile_id) + assert "agent" not in payload and payload["secrets"] == {} + parsed = StartConversationRequest.model_validate(payload) + assert parsed.agent_profile_id == profile_id + assert payload["max_iterations"] == 17 + assert payload["tags"] == {"automationrun": "run-one"} + assert payload["stuck_detection"] is False + assert parsed.hook_config == hooks + assert payload["observability_metadata"] == {"run": "one"} + assert payload["observability_tags"] == ["automation"] + assert payload["observability_span_name"] == "scheduled-task" + assert payload["user_id"] == "operator" + response.json.return_value["agent"] = self.agent.model_dump(mode="json") + response.json.return_value["max_iterations"] = 17 + created = True + return response + + client.request.side_effect = respond + conversation = RemoteConversation.create( + workspace=self.workspace, + request=StartConversationRequest( + workspace=LocalWorkspace(working_dir=self.workspace.working_dir), + conversation_id=cid, + agent_profile_id=profile_id, + max_iterations=17, + tags={"automationrun": "run-one"}, + stuck_detection=False, + hook_config=hooks, + observability_metadata={"run": "one"}, + observability_tags=["automation"], + observability_span_name="scheduled-task", + user_id="operator", + ), + visualizer=None, + ) + assert client.request.call_args_list[0].args == ("POST", "/api/conversations") + assert conversation.max_iteration_per_run == 17 + assert conversation.id == cid + assert conversation.agent.llm.model == self.agent.llm.model + conversation.set_title("Scheduled run") + assert any( + c.args == ("PATCH", f"/api/conversations/{cid}") + and c.kwargs["json"] == {"title": "Scheduled run"} + for c in client.request.call_args_list + ) + conversation.close() + + @pytest.mark.parametrize("status", [403, 404]) + @pytest.mark.parametrize("operation", ["attach", "create"]) + def test_failed_explicit_operation_does_not_try_the_other_operation( + self, status, operation + ): + cid = uuid.uuid4() + client = self.setup_mock_client(str(cid)) + client.request.side_effect = None + client.request.return_value = httpx.Response( + status, request=httpx.Request("GET", f"{self.host}/api/conversations/{cid}") + ) + with pytest.raises(httpx.HTTPStatusError): + if operation == "attach": + RemoteConversation.attach(self.workspace, cid, visualizer=None) + else: + RemoteConversation.create( + self.workspace, + StartConversationRequest( + agent=self.agent, + workspace=LocalWorkspace(working_dir="/tmp"), + conversation_id=cid, + ), + visualizer=None, + ) + expected_method = "GET" if operation == "attach" else "POST" + assert [call.args[0] for call in client.request.call_args_list] == [ + expected_method + ] + @patch( "openhands.sdk.conversation.impl.remote_conversation.WebSocketCallbackClient" ) diff --git a/tests/sdk/workspace/remote/test_runtime_scope.py b/tests/sdk/workspace/remote/test_runtime_scope.py new file mode 100644 index 0000000000..f248e3098a --- /dev/null +++ b/tests/sdk/workspace/remote/test_runtime_scope.py @@ -0,0 +1,155 @@ +"""Scoped workspace operations reuse the normal sync/async request machinery.""" + +import inspect +from unittest.mock import patch +from uuid import uuid4 + +import httpx +import pytest +from pydantic import ValidationError + +from openhands.sdk.workspace import RemoteWorkspace +from openhands.sdk.workspace.remote.async_remote_workspace import AsyncRemoteWorkspace + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("scoped", [False, True]) +@pytest.mark.asyncio +async def test_commands_files_git_and_lifecycle(async_mode, scoped, tmp_path): + cid = uuid4() if scoped else None + prefix = f"/api/conversations/{cid}" if cid else "/api" + requests = [] + output_reads = 0 + + def respond(request): + nonlocal output_reads + requests.append(request) + assert request.headers["X-Session-API-Key"] == "test-scope-key" + path = request.url.path + assert path.startswith(prefix) + if path.endswith("start_bash_command"): + return httpx.Response(200, json={"id": "command-one"}) + if path.endswith("bash_events/search"): + assert request.url.params["command_id__eq"] == "command-one" + output_reads += 1 + return httpx.Response( + 200, + json={ + "items": [ + { + "kind": "BashOutput", + "id": "output-one", + "order": 1, + "exit_code": None if output_reads == 1 else 0, + "stdout": "done", + "stderr": "", + } + ] + }, + ) + if path.endswith("file/upload"): + assert request.url.params["path"] == "/workspace/bundle.tgz" + assert b"bundle-bytes" in request.content + return httpx.Response(200, json={"success": True, "file_size": 12}) + if path.endswith("file/download"): + assert request.url.params["path"] == "/workspace/bundle.tgz" + return httpx.Response(200, content=b"bundle-bytes") + if path.endswith("git/changes"): + assert request.url.params["path"] == "/workspace/repo" + return httpx.Response(200, json=[{"status": "UPDATED", "path": "file.txt"}]) + if path.endswith("git/diff"): + assert request.url.params["path"] == "/workspace/repo/file.txt" + return httpx.Response(200, json={"original": "before", "modified": "after"}) + if path.endswith("runtime/credentials"): + return httpx.Response(200, json={"session_api_key": "worker-key"}) + if path.endswith("/runtime"): + return httpx.Response(404) # Releasing an already-gone runtime is safe. + raise AssertionError(path) + + cls = AsyncRemoteWorkspace if async_mode else RemoteWorkspace + http_cls = httpx.AsyncClient if async_mode else httpx.Client + client = http_cls( + transport=httpx.MockTransport(respond), + base_url="http://test", + headers={"X-Session-API-Key": "test-scope-key"}, + ) + workspace = cls( + host="http://test", + api_key="test-scope-key", + working_dir="/workspace", + runtime_conversation_id=cid, + ) + + async def invoke(method, *args): + result = method(*args) + return await result if inspect.isawaitable(result) else result + + with patch.object( + httpx, "AsyncClient" if async_mode else "Client", return_value=client + ): + try: + command_id = await invoke(workspace.start_command, "echo done") + assert command_id == "command-one" + pending = await invoke(workspace.get_command_output, command_id) + assert pending["exit_code"] is None + done = await invoke(workspace.get_command_output, command_id) + assert done["exit_code"] == 0 + result = await invoke(workspace.execute_command, "echo done") + assert result.exit_code == 0 and result.stdout == "done" + uploaded = await invoke( + workspace.file_upload, b"bundle-bytes", "/workspace/bundle.tgz" + ) + assert uploaded.success + downloaded = await invoke( + workspace.file_download, + "/workspace/bundle.tgz", + tmp_path / "bundle.tgz", + ) + assert downloaded.success + assert (tmp_path / "bundle.tgz").read_bytes() == b"bundle-bytes" + changes = await invoke(workspace.git_changes, "repo") + assert [str(change.path) for change in changes] == ["file.txt"] + diff = await invoke(workspace.git_diff, "repo/file.txt") + assert (diff.original, diff.modified) == ("before", "after") + if scoped: + assert await invoke(workspace.get_runtime_session_key) == "worker-key" + await invoke(workspace.release_runtime) + else: + for method in ( + workspace.get_runtime_session_key, + workspace.release_runtime, + ): + with pytest.raises(ValueError, match="conversation scope"): + await invoke(method) + assert all(r.url.path.startswith(prefix) for r in requests) + with pytest.raises(ValidationError): + workspace.runtime_conversation_id = uuid4() + finally: + await invoke(workspace.reset_client) + + +@pytest.mark.asyncio +async def test_async_workspace_context_closes_client_on_failure(): + async with AsyncRemoteWorkspace( + host="http://test", working_dir="/workspace" + ) as workspace: + client = workspace.client + assert not client.is_closed + assert client.is_closed + + +@pytest.mark.parametrize( + "payload", [{}, {"session_api_key": ""}, {"session_api_key": 1}] +) +def test_rejects_missing_runtime_credential(payload): + with httpx.Client( + transport=httpx.MockTransport(lambda r: httpx.Response(200, json=payload)) + ) as client: + with patch("httpx.Client", return_value=client): + workspace = RemoteWorkspace( + host="http://test", + working_dir="/workspace", + runtime_conversation_id=uuid4(), + ) + with pytest.raises(ValueError, match="empty session credential"): + workspace.get_runtime_session_key()