From 425dcdef5cea3cb21a3f94a0676771018521b037 Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 19:38:22 +0000 Subject: [PATCH 1/5] refactor: share conversation execution across local and Docker workspaces --- README.md | 44 +++--- openhands/automation/backends/__init__.py | 11 +- openhands/automation/backends/conversation.py | 133 ++++++++++++++++++ openhands/automation/backends/docker.py | 98 +------------ openhands/automation/config.py | 26 ++++ openhands/automation/dispatcher.py | 10 +- openhands/automation/router.py | 2 +- openhands/automation/watchdog.py | 4 +- tests/test_conversation_backend.py | 107 ++++++++++++++ 9 files changed, 311 insertions(+), 124 deletions(-) create mode 100644 openhands/automation/backends/conversation.py create mode 100644 tests/test_conversation_backend.py diff --git a/README.md b/README.md index dc25a454..413aae15 100644 --- a/README.md +++ b/README.md @@ -19,22 +19,30 @@ The Automation Service owns automation definitions, cron scheduling, webhooks, r ## Development -### Local Docker execution - -Set `AUTOMATION_AGENT_SERVER_URL` and `AUTOMATION_AGENT_SERVER_API_KEY` to an -agent-server running in Docker conversation mode. Set -`AUTOMATION_DOCKER_AGENT_PROFILE` to the UUID of a saved agent profile on that -server. Each bundle then gets its own Docker conversation and `/workspace`. -The server must support scoped runtime routes, runtime credential provisioning, -and runtime release. `AUTOMATION_DOCKER_MAX_CONCURRENT_RUNS` defaults to 2; -use the agent-server container CPU, memory, and PID settings to bound each run. - -Bundles receive `AUTOMATION_CONVERSATION_ID`, an inner `AGENT_SERVER_URL`, and -only that runtime's `SESSION_API_KEY`. The outer server key and shared automation -callback key are not forwarded. The watchdog polls the bundle's scoped bash -result and releases finished containers while preserving conversation history. -This opt-in mode currently uses one configured agent profile for all bundles; -the existing local and Cloud execution modes retain their defaults. +### Conversation execution in local or Docker workspaces + +Set `AUTOMATION_AGENT_SERVER_URL`, `AUTOMATION_AGENT_SERVER_API_KEY`, and +`AUTOMATION_AGENT_PROFILE` (a saved agent profile UUID). The backend reads the +server's authoritative `conversation_runtime` and provisions a run conversation +using the same API and profile in either mode. No bundle configuration or workflow +branch changes when switching workspace kind. + +Both modes supply `AUTOMATION_CONVERSATION_ID`, `AGENT_SERVER_URL`, +`SESSION_API_KEY`, and `WORKSPACE_BASE`, and use conversation-scoped upload, +bash execution, and completion verification. Local workspaces live in per-run +subdirectories of the configured workspace root. Docker workspaces use +`/workspace` and receive only the selected inner session key, never the outer +server key or shared callback key. Local mode retains its existing single-tenant +server credential boundary; a local workspace is not a security sandbox. + +`AUTOMATION_CONVERSATION_MAX_CONCURRENT_RUNS` defaults to 2. Docker servers must +support runtime credential provisioning and release; bound container CPU, memory, +and PIDs in the server configuration. Completed Docker runtimes are released while +history remains; the persistent local server and its history are retained. + +The earlier `AUTOMATION_DOCKER_AGENT_PROFILE` and related Docker-only settings +remain compatibility aliases for existing deployments. Without either profile +setting, existing local and Cloud dispatch behavior is unchanged. ### Prerequisites @@ -119,8 +127,8 @@ containers/ # Docker configuration This service is deployed via the [deploy repository](https://github.com/All-Hands-AI/deploy). Docker images are automatically built and pushed to `ghcr.io/openhands/automation` on every push to main and on tags. -For different role permissions, set `AUTOMATION_DOCKER_AGENT_PROFILE_OVERRIDES` +For different role permissions, set `AUTOMATION_AGENT_PROFILE_OVERRIDES` to a JSON object mapping automation UUIDs to saved agent profile UUIDs. Unmapped -automations use `AUTOMATION_DOCKER_AGENT_PROFILE`. This host-controlled mapping +automations use `AUTOMATION_AGENT_PROFILE`. This host-controlled mapping lets deterministic jobs select a profile with no model credential or agent tools, while implementation and review jobs select only their required tools/model. diff --git a/openhands/automation/backends/__init__.py b/openhands/automation/backends/__init__.py index ccb210d4..6efdf9e5 100644 --- a/openhands/automation/backends/__init__.py +++ b/openhands/automation/backends/__init__.py @@ -22,6 +22,7 @@ from openhands.automation.backends.base import ExecutionBackend, ExecutionContext from openhands.automation.backends.cloud import CloudSandboxBackend +from openhands.automation.backends.conversation import ConversationAgentServerBackend from openhands.automation.backends.docker import DockerAgentServerBackend from openhands.automation.backends.local import LocalAgentServerBackend @@ -46,7 +47,9 @@ def get_backend(run: AutomationRun) -> ExecutionBackend: if settings.is_local_mode: backend_type = ( - DockerAgentServerBackend + ConversationAgentServerBackend + if settings.agent_profile + else DockerAgentServerBackend if settings.docker_agent_profile else LocalAgentServerBackend ) @@ -58,9 +61,9 @@ def get_backend(run: AutomationRun) -> ExecutionBackend: callback_api_key=settings.local_api_key, sandbox_agent_server_url=settings.sandbox_agent_server_url or None, ) - if isinstance(backend, DockerAgentServerBackend): - backend.agent_profile_id = settings.docker_agent_profile_overrides.get( - str(run.automation_id), settings.docker_agent_profile + if isinstance(backend, ConversationAgentServerBackend): + backend.agent_profile_id = settings.run_agent_profile_overrides.get( + str(run.automation_id), settings.run_agent_profile ) return backend else: diff --git a/openhands/automation/backends/conversation.py b/openhands/automation/backends/conversation.py new file mode 100644 index 00000000..834e51a0 --- /dev/null +++ b/openhands/automation/backends/conversation.py @@ -0,0 +1,133 @@ +"""One bundle-facing conversation contract for local and Docker workspaces.""" + +from __future__ import annotations + +import httpx + +from openhands.automation.backends.base import ExecutionContext +from openhands.automation.backends.local import LocalAgentServerBackend +from openhands.automation.utils.agent_server import ( + VerificationResult, + verify_run_on_agent_server, +) + + +class ConversationAgentServerBackend(LocalAgentServerBackend): + agent_profile_id: str + runtime_api_key: str = "" + _runtime_kind: str | None = None + + async def _resolve_runtime(self, client: httpx.AsyncClient) -> str: + if self._runtime_kind is None: + response = await client.get( + f"{self.agent_server_url}/server_info", + headers={"X-Session-API-Key": self.api_key}, + ) + response.raise_for_status() + self._runtime_kind = response.json()["conversation_runtime"] + if self._runtime_kind not in ("local", "docker"): + raise ValueError("Unsupported agent-server conversation runtime") + return self._runtime_kind + + @property + def api_prefix(self) -> str: + return f"/api/conversations/{self._run.id}" + + async def get_execution_context( + self, client: httpx.AsyncClient + ) -> ExecutionContext: + runtime_kind = await self._resolve_runtime(client) + response = await client.post( + f"{self.agent_server_url}/api/conversations", + headers={"X-Session-API-Key": self.api_key}, + json={ + "conversation_id": str(self._run.id), + "agent_profile_id": self.agent_profile_id, + "workspace": { + "kind": "LocalWorkspace", + "working_dir": self.get_work_dir(str(self._run.id)), + }, + "title": self._run.automation.name, + "max_iterations": 160, + "tags": {"automationrun": str(self._run.id)}, + }, + timeout=180, + ) + response.raise_for_status() + self.runtime_api_key = self.api_key if runtime_kind == "local" else "" + if runtime_kind == "docker": + try: + credentials = await client.post( + f"{self.agent_server_url}{self.api_prefix}/runtime/credentials", + headers={"X-Session-API-Key": self.api_key}, + ) + credentials.raise_for_status() + self.runtime_api_key = credentials.json()["session_api_key"] + if not self.runtime_api_key: + raise ValueError("Runtime returned an empty session credential") + except Exception: + await self.release_context( + client, ExecutionContext(self.agent_server_url, self.api_key) + ) + raise + return ExecutionContext( + agent_url=self.agent_server_url, + session_key=self.api_key, + api_prefix=self.api_prefix, + ) + + def build_env_vars(self) -> dict[str, str]: + if not self.runtime_api_key: + raise RuntimeError("Runtime credentials have not been provisioned") + return { + "AGENT_SERVER_URL": ( + self.sandbox_agent_server_url + or ( + "http://127.0.0.1:8000" + if self._runtime_kind == "docker" + else self.agent_server_url + ) + ), + "AUTOMATION_CONVERSATION_ID": str(self._run.id), + "WORKSPACE_BASE": self.get_work_dir(str(self._run.id)), + "SESSION_API_KEY": self.runtime_api_key, + } + + def get_work_dir(self, run_id: str) -> str: + if self._runtime_kind is None: + raise RuntimeError( + "Resolve the server runtime before selecting a workspace" + ) + return ( + "/workspace" + if self._runtime_kind == "docker" + else super().get_work_dir(run_id) + ) + + async def release_context( + self, client: httpx.AsyncClient, ctx: ExecutionContext + ) -> None: + if await self._resolve_runtime(client) == "local": + return # Persistent server and conversation history belong to the host. + response = await client.delete( + f"{ctx.agent_url}{self.api_prefix}/runtime", + headers={"X-Session-API-Key": self.api_key}, + timeout=60, + ) + if response.status_code != 404: + response.raise_for_status() + + async def verify_run(self, run_id: str) -> VerificationResult: + return await verify_run_on_agent_server( + agent_url=self.agent_server_url, + session_key=self.api_key, + run_id=run_id, + bash_command_id=self._run.bash_command_id, + api_prefix=self.api_prefix, + ) + + async def cleanup_after_verification(self, run_id: str) -> None: # noqa: ARG002 + async with httpx.AsyncClient() as client: + await self.release_context( + client, ExecutionContext(self.agent_server_url, self.api_key) + ) diff --git a/openhands/automation/backends/docker.py b/openhands/automation/backends/docker.py index f9e6a6ba..0aefaa86 100644 --- a/openhands/automation/backends/docker.py +++ b/openhands/automation/backends/docker.py @@ -1,97 +1,7 @@ -"""Run local automation bundles inside agent-server Docker conversations.""" +"""Compatibility backend for the original Docker-only profile setting.""" -from __future__ import annotations +from openhands.automation.backends.conversation import ConversationAgentServerBackend -import httpx -from openhands.automation.backends.base import ExecutionContext -from openhands.automation.backends.local import LocalAgentServerBackend -from openhands.automation.utils.agent_server import ( - VerificationResult, - verify_run_on_agent_server, -) - - -class DockerAgentServerBackend(LocalAgentServerBackend): - agent_profile_id: str - runtime_api_key: str = "" - - @property - def api_prefix(self) -> str: - return f"/api/conversations/{self._run.id}" - - async def get_execution_context( - self, client: httpx.AsyncClient - ) -> ExecutionContext: - response = await client.post( - f"{self.agent_server_url}/api/conversations", - headers={"X-Session-API-Key": self.api_key}, - json={ - "conversation_id": str(self._run.id), - "agent_profile_id": self.agent_profile_id, - "workspace": {"kind": "LocalWorkspace", "working_dir": "/workspace"}, - "title": self._run.automation.name, - "max_iterations": 160, - "tags": {"automationrun": str(self._run.id)}, - }, - timeout=180, - ) - response.raise_for_status() - try: - credentials = await client.post( - f"{self.agent_server_url}{self.api_prefix}/runtime/credentials", - headers={"X-Session-API-Key": self.api_key}, - ) - credentials.raise_for_status() - self.runtime_api_key = credentials.json()["session_api_key"] - if not self.runtime_api_key: - raise ValueError("Runtime returned an empty session credential") - except Exception: - await self.release_context( - client, ExecutionContext(self.agent_server_url, self.api_key) - ) - raise - return ExecutionContext( - agent_url=self.agent_server_url, - session_key=self.api_key, - api_prefix=self.api_prefix, - ) - - def build_env_vars(self) -> dict[str, str]: - if not self.runtime_api_key: - raise RuntimeError("Runtime credentials have not been provisioned") - return { - "AGENT_SERVER_URL": "http://127.0.0.1:8000", - "AUTOMATION_CONVERSATION_ID": str(self._run.id), - "WORKSPACE_BASE": "/workspace", - "SESSION_API_KEY": self.runtime_api_key, - } - - def get_work_dir(self, run_id: str) -> str: # noqa: ARG002 - return "/workspace" - - async def release_context( - self, client: httpx.AsyncClient, ctx: ExecutionContext - ) -> None: - response = await client.delete( - f"{ctx.agent_url}{self.api_prefix}/runtime", - headers={"X-Session-API-Key": self.api_key}, - timeout=60, - ) - if response.status_code != 404: - response.raise_for_status() - - async def verify_run(self, run_id: str) -> VerificationResult: - return await verify_run_on_agent_server( - agent_url=self.agent_server_url, - session_key=self.api_key, - run_id=run_id, - bash_command_id=self._run.bash_command_id, - api_prefix=self.api_prefix, - ) - - async def cleanup_after_verification(self, run_id: str) -> None: # noqa: ARG002 - async with httpx.AsyncClient() as client: - await self.release_context( - client, ExecutionContext(self.agent_server_url, self.api_key) - ) +class DockerAgentServerBackend(ConversationAgentServerBackend): + _runtime_kind = "docker" diff --git a/openhands/automation/config.py b/openhands/automation/config.py index e50d5834..e99381bb 100644 --- a/openhands/automation/config.py +++ b/openhands/automation/config.py @@ -564,9 +564,35 @@ class ServiceSettings(BaseSettings): # - Authenticates using local_api_key instead of OpenHands SaaS API agent_server_url: str = "" agent_server_api_key: str = "" + # Shared conversation execution; the server advertises its workspace runtime. + agent_profile: str = "" + agent_profile_overrides: dict[str, str] = Field(default_factory=dict) + conversation_max_concurrent_runs: int = Field(default=2, ge=1) + # Compatibility for early Docker-only deployments. docker_agent_profile: str = "" docker_agent_profile_overrides: dict[str, str] = Field(default_factory=dict) docker_max_concurrent_runs: int = Field(default=2, ge=1) + + @property + def run_agent_profile(self) -> str: + return self.agent_profile or self.docker_agent_profile + + @property + def run_agent_profile_overrides(self) -> dict[str, str]: + return ( + self.agent_profile_overrides + if self.agent_profile + else self.docker_agent_profile_overrides + ) + + @property + def run_concurrency_limit(self) -> int: + return ( + self.conversation_max_concurrent_runs + if self.agent_profile + else self.docker_max_concurrent_runs + ) + # Optional override for the AGENT_SERVER_URL env var exported into the # in-sandbox bash chain by LocalAgentServerBackend.build_env_vars. # When empty, defaults to agent_server_url (the URL the backend itself diff --git a/openhands/automation/dispatcher.py b/openhands/automation/dispatcher.py index 3e323de2..568765c7 100644 --- a/openhands/automation/dispatcher.py +++ b/openhands/automation/dispatcher.py @@ -131,9 +131,9 @@ async def _poll_pending_runs( Eagerly loads the ``automation`` relationship so that ``user_id``, ``org_id``, and tarball config are available for dispatch. """ - docker_profile = get_config().service.docker_agent_profile + run_profile = get_config().service.run_agent_profile active = [] - if docker_profile: + if run_profile: active = ( ( await session.execute( @@ -146,7 +146,7 @@ async def _poll_pending_runs( .all() ) batch_size = min( - batch_size, 1, get_config().service.docker_max_concurrent_runs - len(active) + batch_size, 1, get_config().service.run_concurrency_limit - len(active) ) if batch_size <= 0: return [] @@ -163,7 +163,7 @@ async def _poll_pending_runs( .order_by(AutomationRun.created_at.asc()) .limit(batch_size) ) - if docker_profile and active: + if run_profile and active: select_query = select_query.where(AutomationRun.automation_id.not_in(active)) # Apply row locking for PostgreSQL only (SQLite doesn't support it) @@ -510,7 +510,7 @@ async def _fail( # 6. Handle result if result.success: - if get_config().service.docker_agent_profile: + if get_config().service.run_agent_profile: async with session_factory() as link_session: await link_session.execute( update(AutomationRun) diff --git a/openhands/automation/router.py b/openhands/automation/router.py index 2dd4e4a6..6cac6ab2 100644 --- a/openhands/automation/router.py +++ b/openhands/automation/router.py @@ -882,7 +882,7 @@ async def cancel_run( from openhands.automation.config import get_config - if get_config().service.docker_agent_profile: + if get_config().service.run_agent_profile: from openhands.automation.backends import get_backend # Release the transaction before waiting for Docker to stop. diff --git a/openhands/automation/watchdog.py b/openhands/automation/watchdog.py index 2974c3c3..9ef6360c 100644 --- a/openhands/automation/watchdog.py +++ b/openhands/automation/watchdog.py @@ -172,7 +172,7 @@ def _should_cleanup_sandbox_after_terminal( the sandbox carrying a live conversation is already excluded here. """ return ( - bool(run.sandbox_id) or bool(get_config().service.docker_agent_profile) + bool(run.sandbox_id) or bool(get_config().service.run_agent_profile) ) and keep_alive is not True @@ -526,7 +526,7 @@ async def mark_stale_runs( AutomationRun.bash_command_id.isnot(None) | (AutomationRun.timeout_at < now) ) - if settings.docker_agent_profile + if settings.run_agent_profile else AutomationRun.timeout_at < now ), ) diff --git a/tests/test_conversation_backend.py b/tests/test_conversation_backend.py new file mode 100644 index 00000000..154bda73 --- /dev/null +++ b/tests/test_conversation_backend.py @@ -0,0 +1,107 @@ +"""Identical bundle-facing contract across advertised workspace runtimes.""" + +import json +from uuid import uuid4 + +import httpx +import pytest + +from openhands.automation.backends.conversation import ConversationAgentServerBackend +from openhands.automation.execution import execute_in_context +from openhands.automation.models import Automation, AutomationRun + + +@pytest.mark.asyncio +@pytest.mark.parametrize("runtime", ["local", "docker"]) +async def test_same_bundle_contract_and_scoped_execution(runtime, tmp_path): + run = AutomationRun(id=uuid4(), automation=Automation(name="portable workflow")) + backend = ConversationAgentServerBackend( + "http://server", + "host-key", + run, + workspace_base=str(tmp_path), + callback_api_key="callback-key", + ) + backend.agent_profile_id = str(uuid4()) + requests = [] + + def respond(request): + requests.append(request) + if request.url.path == "/server_info": + return httpx.Response(200, json={"conversation_runtime": runtime}) + return httpx.Response( + 200, json={"id": "command", "session_api_key": "inner-key"} + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + context = await backend.get_execution_context(client) + env = backend.build_env_vars() + assert set(env) == { + "AGENT_SERVER_URL", + "SESSION_API_KEY", + "AUTOMATION_CONVERSATION_ID", + "WORKSPACE_BASE", + } + assert env["AUTOMATION_CONVERSATION_ID"] == str(run.id) + assert env["WORKSPACE_BASE"] == backend.get_work_dir(str(run.id)) + assert env["SESSION_API_KEY"] == ( + "inner-key" if runtime == "docker" else "host-key" + ) + if runtime == "docker": + assert "host-key" not in str(env) + creation = next(r for r in requests if r.url.path == "/api/conversations") + payload = json.loads(creation.content) + assert payload["workspace"]["working_dir"] == env["WORKSPACE_BASE"] + assert payload["conversation_id"] == str(run.id) + assert payload["agent_profile_id"] == backend.agent_profile_id + assert payload["max_iterations"] == 160 + assert payload["tags"] == {"automationrun": str(run.id)} + result = await execute_in_context( + client, + context.agent_url, + context.session_key, + "python3 main.py", + b"same-bundle", + env["WORKSPACE_BASE"], + env, + run_id=str(run.id), + api_prefix=context.api_prefix, + ) + assert result.success + assert result.bash_command_id == "command" + execution = [ + r for r in requests if "/file/" in r.url.path or "/bash/" in r.url.path + ] + assert execution + assert all( + r.url.path.startswith(f"/api/conversations/{run.id}/") for r in execution + ) + before = len(requests) + await backend.release_context(client, context) + if runtime == "local": + assert len(requests) == before + else: + assert requests[-1].method == "DELETE" + assert requests[-1].url.path == f"/api/conversations/{run.id}/runtime" + + +def test_generic_profile_selects_shared_backend_and_per_automation_override( + monkeypatch, +): + from openhands.automation.backends import get_backend + from openhands.automation.config import clear_config_cache + + automation_id = uuid4() + monkeypatch.setenv("AUTOMATION_AGENT_SERVER_URL", "http://server") + monkeypatch.setenv("AUTOMATION_AGENT_PROFILE", "default-profile") + monkeypatch.setenv( + "AUTOMATION_AGENT_PROFILE_OVERRIDES", + json.dumps({str(automation_id): "role-profile"}), + ) + clear_config_cache() + try: + backend = get_backend(AutomationRun(id=uuid4(), automation_id=automation_id)) + assert type(backend) is ConversationAgentServerBackend + assert backend.agent_profile_id == "role-profile" + finally: + clear_config_cache() From 6126a7cb9b8ac32a0c33c29019dd700e861c181f Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 20:10:54 +0000 Subject: [PATCH 2/5] refactor(dispatch): delegate Agent Server communication to the SDK --- README.md | 9 +++ openhands/automation/backends/conversation.py | 64 +++++++--------- openhands/automation/execution.py | 39 ++++------ openhands/automation/utils/agent_server.py | 23 ++---- tests/test_execution.py | 73 ++++++------------- 5 files changed, 79 insertions(+), 129 deletions(-) diff --git a/README.md b/README.md index 413aae15..f391dbbc 100644 --- a/README.md +++ b/README.md @@ -132,3 +132,12 @@ to a JSON object mapping automation UUIDs to saved agent profile UUIDs. Unmapped automations use `AUTOMATION_AGENT_PROFILE`. This host-controlled mapping lets deterministic jobs select a profile with no model credential or agent tools, while implementation and review jobs select only their required tools/model. + +### SDK Client Integration Dependency + +The conversation backend and Agent Server execution helpers use public clients +from [software-agent-sdk #5010](https://github.com/OpenHands/software-agent-sdk/pull/5010). +This draft requires that SDK release before merging; the released dependency pin +must be updated then. Integration validation currently supplies the reviewed SDK +source through `PYTHONPATH`, including the server runtime stack. Workflow bundles +receive the same environment contract in local and Docker workspaces. diff --git a/openhands/automation/backends/conversation.py b/openhands/automation/backends/conversation.py index 834e51a0..d835006e 100644 --- a/openhands/automation/backends/conversation.py +++ b/openhands/automation/backends/conversation.py @@ -10,6 +10,7 @@ VerificationResult, verify_run_on_agent_server, ) +from openhands.sdk.client import AsyncAgentServerClient class ConversationAgentServerBackend(LocalAgentServerBackend): @@ -19,52 +20,43 @@ class ConversationAgentServerBackend(LocalAgentServerBackend): async def _resolve_runtime(self, client: httpx.AsyncClient) -> str: if self._runtime_kind is None: - response = await client.get( - f"{self.agent_server_url}/server_info", - headers={"X-Session-API-Key": self.api_key}, - ) - response.raise_for_status() - self._runtime_kind = response.json()["conversation_runtime"] + info = await AsyncAgentServerClient( + self.agent_server_url, self.api_key, http_client=client + ).get_server_info() + self._runtime_kind = info["conversation_runtime"] if self._runtime_kind not in ("local", "docker"): raise ValueError("Unsupported agent-server conversation runtime") return self._runtime_kind @property def api_prefix(self) -> str: - return f"/api/conversations/{self._run.id}" + return ( + AsyncAgentServerClient(self.agent_server_url, self.api_key) + .runtime(str(self._run.id)) + .api_prefix + ) async def get_execution_context( self, client: httpx.AsyncClient ) -> ExecutionContext: runtime_kind = await self._resolve_runtime(client) - response = await client.post( - f"{self.agent_server_url}/api/conversations", - headers={"X-Session-API-Key": self.api_key}, - json={ - "conversation_id": str(self._run.id), - "agent_profile_id": self.agent_profile_id, - "workspace": { - "kind": "LocalWorkspace", - "working_dir": self.get_work_dir(str(self._run.id)), - }, - "title": self._run.automation.name, - "max_iterations": 160, - "tags": {"automationrun": str(self._run.id)}, - }, - timeout=180, + server = AsyncAgentServerClient( + self.agent_server_url, self.api_key, http_client=client + ) + await server.create_conversation( + conversation_id=str(self._run.id), + agent_profile_id=self.agent_profile_id, + working_dir=self.get_work_dir(str(self._run.id)), + title=self._run.automation.name, + max_iterations=160, + tags={"automationrun": str(self._run.id)}, ) - response.raise_for_status() self.runtime_api_key = self.api_key if runtime_kind == "local" else "" if runtime_kind == "docker": try: - credentials = await client.post( - f"{self.agent_server_url}{self.api_prefix}/runtime/credentials", - headers={"X-Session-API-Key": self.api_key}, - ) - credentials.raise_for_status() - self.runtime_api_key = credentials.json()["session_api_key"] - if not self.runtime_api_key: - raise ValueError("Runtime returned an empty session credential") + self.runtime_api_key = await server.runtime( + str(self._run.id) + ).get_session_key() except Exception: await self.release_context( client, ExecutionContext(self.agent_server_url, self.api_key) @@ -109,13 +101,11 @@ async def release_context( ) -> None: if await self._resolve_runtime(client) == "local": return # Persistent server and conversation history belong to the host. - response = await client.delete( - f"{ctx.agent_url}{self.api_prefix}/runtime", - headers={"X-Session-API-Key": self.api_key}, - timeout=60, + await ( + AsyncAgentServerClient(ctx.agent_url, self.api_key, http_client=client) + .runtime(str(self._run.id)) + .release() ) - if response.status_code != 404: - response.raise_for_status() async def verify_run(self, run_id: str) -> VerificationResult: return await verify_run_on_agent_server( diff --git a/openhands/automation/execution.py b/openhands/automation/execution.py index 52231eaf..1109942a 100644 --- a/openhands/automation/execution.py +++ b/openhands/automation/execution.py @@ -27,6 +27,7 @@ from openhands.automation.utils import log_extra from openhands.automation.utils.sandbox import delete_sandbox from openhands.automation.utils.timeout import resolve_automation_timeout_seconds +from openhands.sdk.client import AsyncAgentServerClient # Default working directory for cloud/container mode @@ -179,16 +180,11 @@ async def _upload( with proxies that collapse double-slashes (e.g. //tmp -> /tmp). See: https://github.com/All-Hands-AI/OpenHands/commit/a14158e """ - # Use query param instead of path param to avoid double-slash normalization - from urllib.parse import urlencode - - params = urlencode({"path": dest}) - resp = await client.post( - f"{agent_url}{api_prefix}/file/upload?{params}", - files={"file": ("upload", data)}, - headers={"X-Session-API-Key": session_key}, + await ( + AsyncAgentServerClient(agent_url, session_key, http_client=client) + .runtime_for_api_prefix(api_prefix) + .upload(dest, data) ) - resp.raise_for_status() async def _bash( @@ -202,14 +198,12 @@ async def _bash( """Run a bash command synchronously. Returns ``(exit_code, stdout, stderr)``.""" if timeout is None: timeout = resolve_automation_timeout_seconds(None) - resp = await client.post( - f"{agent_url}{api_prefix}/bash/execute_bash_command", - json={"command": command, "timeout": timeout}, - headers={"X-Session-API-Key": session_key}, - timeout=httpx.Timeout(timeout + 30), + body = ( + await AsyncAgentServerClient(agent_url, session_key, http_client=client) + .runtime_for_api_prefix(api_prefix) + .execute(command, timeout=timeout) ) - resp.raise_for_status() - body = resp.json() + return body.get("exit_code"), body.get("stdout") or "", body.get("stderr") or "" @@ -224,15 +218,12 @@ async def _start_bash( """Start a bash command in the background. Returns the command ID.""" if timeout is None: timeout = resolve_automation_timeout_seconds(None) - http_timeout = get_config().http.http_timeout - resp = await client.post( - f"{agent_url}{api_prefix}/bash/start_bash_command", - json={"command": command, "timeout": timeout}, - headers={"X-Session-API-Key": session_key}, - timeout=http_timeout, + body = ( + await AsyncAgentServerClient(agent_url, session_key, http_client=client) + .runtime_for_api_prefix(api_prefix) + .start(command, timeout=timeout) ) - resp.raise_for_status() - body = resp.json() + return body.get("id") diff --git a/openhands/automation/utils/agent_server.py b/openhands/automation/utils/agent_server.py index 52fc4084..3065a9c8 100644 --- a/openhands/automation/utils/agent_server.py +++ b/openhands/automation/utils/agent_server.py @@ -16,6 +16,7 @@ TransientErrorInfo, classify_httpx_transient_error, ) +from openhands.sdk.client import AsyncAgentServerClient logger = logging.getLogger(__name__) @@ -63,25 +64,11 @@ async def get_last_bash_command_result( BashCommandResult with found=True if command result was retrieved """ try: - # Search for the most recent BashOutput event, scoped to this run's - # bash command whenever we know which one it is. The agent-server's - # search endpoint accepts ``command_id__eq`` and only matches - # BashOutput files whose embedded command_id matches. - params: dict[str, str | int] = { - "kind__eq": "BashOutput", - "sort_order": "TIMESTAMP_DESC", - "limit": 1, - } - if command_id: - params["command_id__eq"] = command_id - resp = await client.get( - f"{agent_url}{api_prefix}/bash/bash_events/search", - params=params, - headers={"X-Session-API-Key": session_key}, - timeout=30.0, + page = ( + await AsyncAgentServerClient(agent_url, session_key, http_client=client) + .runtime_for_api_prefix(api_prefix) + .get_output(command_id) ) - resp.raise_for_status() - page = resp.json() items = page.get("items", []) if not items: diff --git a/tests/test_execution.py b/tests/test_execution.py index d984ee27..b2724226 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -9,6 +9,7 @@ import tarfile from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from openhands.automation.config import get_config @@ -160,59 +161,31 @@ class TestUploadUsesQueryParams: """ @pytest.mark.asyncio - async def test_upload_uses_query_param_for_path(self): - """_upload should use ?path= query param, not path in URL.""" - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.post = AsyncMock(return_value=mock_response) - - await _upload( - client=mock_client, - agent_url="https://agent.example.com", - session_key="test-session-key", - data=b"test data", - dest="/tmp/automation.tar.gz", - ) - - # Verify post was called with query param, not path param - mock_client.post.assert_called_once() - call_args = mock_client.post.call_args - - url = call_args[0][0] - # URL should use query param format - assert "?path=" in url, f"Expected query param in URL, got: {url}" - assert "/tmp/automation.tar.gz" not in url.split("?")[0], ( - f"Path should not be in URL path segment: {url}" - ) - # Verify the path is properly encoded in query string - assert ( - "path=%2Ftmp%2Fautomation.tar.gz" in url - or "path=/tmp/automation.tar.gz" in url - ) - - @pytest.mark.asyncio - async def test_upload_preserves_absolute_path(self): - """_upload should preserve leading slash in path via query param.""" - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() + @pytest.mark.parametrize( + "destination", ["/tmp/automation.tar.gz", "/workspace/file.txt"] + ) + async def test_upload_preserves_path_in_query(self, destination): + """Exercise the SDK transport through the dispatcher's upload entrypoint.""" + requests = [] - mock_client = AsyncMock() - mock_client.post = AsyncMock(return_value=mock_response) + def respond(request): + requests.append(request) + return httpx.Response(200, json={}) - await _upload( - client=mock_client, - agent_url="https://agent.example.com", - session_key="test-session-key", - data=b"test data", - dest="/workspace/file.txt", - ) + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + await _upload( + client=client, + agent_url="https://agent.example.com", + session_key="test-session-key", + data=b"test data", + dest=destination, + ) - url = mock_client.post.call_args[0][0] - # The path in query param should preserve the leading slash - # (either URL-encoded as %2F or literal /) - assert "%2Fworkspace" in url or "/workspace" in url.split("?")[1] + assert len(requests) == 1 + assert requests[0].method == "POST" + assert requests[0].url.path == "/api/file/upload" + assert requests[0].url.params["path"] == destination + assert requests[0].headers["X-Session-API-Key"] == "test-session-key" class TestExecuteInContextErrors: From 8c8cba0b4b7ac3d30d562284f20fc3ca51822345 Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 20:13:08 +0000 Subject: [PATCH 3/5] fix(dispatch): validate the SDK background command result --- openhands/automation/execution.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/openhands/automation/execution.py b/openhands/automation/execution.py index 1109942a..824e3f22 100644 --- a/openhands/automation/execution.py +++ b/openhands/automation/execution.py @@ -224,7 +224,10 @@ async def _start_bash( .start(command, timeout=timeout) ) - return body.get("id") + command_id = body.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 _is_permanent_http_error(stderr: str) -> bool: From f2c1f0f5f4e69d35894f8d4a47fd948fc7f6e664 Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 20:21:56 +0000 Subject: [PATCH 4/5] build(dispatch): pin the SDK orchestration client integration --- README.md | 7 ++++--- pyproject.toml | 3 ++- uv.lock | 10 +++------- 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index f391dbbc..f40f5284 100644 --- a/README.md +++ b/README.md @@ -137,7 +137,8 @@ while implementation and review jobs select only their required tools/model. The conversation backend and Agent Server execution helpers use public clients from [software-agent-sdk #5010](https://github.com/OpenHands/software-agent-sdk/pull/5010). -This draft requires that SDK release before merging; the released dependency pin -must be updated then. Integration validation currently supplies the reviewed SDK -source through `PYTHONPATH`, including the server runtime stack. Workflow bundles +This draft pins the SDK implementation by immutable Git commit so its tests and +source installation are reproducible. Replace that integration pin with the SDK +release before merging. The live factory additionally integrates the server +runtime stack; those server changes are separate from this client dependency. Workflow bundles receive the same environment contract in local and Docker workspaces. diff --git a/pyproject.toml b/pyproject.toml index eea21177..13e582c6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,8 @@ dependencies = [ "google-cloud-storage>=2.18", "httpx>=0.27", "jmespath>=1.0", - "openhands-sdk==1.46.0", + # Temporary integration pin for software-agent-sdk#5010; replace with its release. + "openhands-sdk @ git+https://github.com/OpenHands/software-agent-sdk.git@79021c687c63bd1e925cf157f5897ceeaa12f029#subdirectory=openhands-sdk", "openhands-workspace==1.46.0", "pg8000>=1.31", "prometheus-client>=0.19", diff --git a/uv.lock b/uv.lock index e8a0c2d2..da7fd2ff 100644 --- a/uv.lock +++ b/uv.lock @@ -2269,7 +2269,7 @@ requires-dist = [ { name = "google-cloud-storage", specifier = ">=2.18" }, { name = "httpx", specifier = ">=0.27" }, { name = "jmespath", specifier = ">=1.0" }, - { name = "openhands-sdk", specifier = "==1.46.0" }, + { name = "openhands-sdk", git = "https://github.com/OpenHands/software-agent-sdk.git?subdirectory=openhands-sdk&rev=79021c687c63bd1e925cf157f5897ceeaa12f029" }, { name = "openhands-workspace", specifier = "==1.46.0" }, { name = "pg8000", specifier = ">=1.31" }, { name = "prometheus-client", specifier = ">=0.19" }, @@ -2301,8 +2301,8 @@ dev = [ [[package]] name = "openhands-sdk" -version = "1.46.0" -source = { registry = "https://pypi.org/simple" } +version = "1.47.0" +source = { git = "https://github.com/OpenHands/software-agent-sdk.git?subdirectory=openhands-sdk&rev=79021c687c63bd1e925cf157f5897ceeaa12f029#79021c687c63bd1e925cf157f5897ceeaa12f029" } dependencies = [ { name = "agent-client-protocol" }, { name = "deprecation" }, @@ -2323,10 +2323,6 @@ dependencies = [ { name = "tree-sitter-bash" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/7c/09/ca00f55b3bca9198b257cc733067d13c01f07cfc400f9e03785545ffa2a5/openhands_sdk-1.46.0.tar.gz", hash = "sha256:4240c0d17d681d694b6b0d2c92462b023ff20c16d5233c6229ec57a5d4f34cbb", size = 714276, upload-time = "2026-09-09T00:56:41.775Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/63/63/199953bcb187562763052ef71db500ae22c84d355eb950ea260fd698830e/openhands_sdk-1.46.0-py3-none-any.whl", hash = "sha256:ef6fd74ac80208d67aa138a1da321479cf3b6a80aed2d68c53463c0551fee84d", size = 849925, upload-time = "2026-09-09T00:56:37.127Z" }, -] [[package]] name = "openhands-workspace" From e78002d22f667819f13deecf05a72a23854ad5f7 Mon Sep 17 00:00:00 2001 From: openhands Date: Sat, 12 Sep 2026 20:31:53 +0000 Subject: [PATCH 5/5] test(dispatch): exercise legacy verification through SDK transport --- tests/test_local_mode.py | 211 +++++++++++---------------------------- 1 file changed, 60 insertions(+), 151 deletions(-) diff --git a/tests/test_local_mode.py b/tests/test_local_mode.py index 4fd7bc10..81be68d2 100644 --- a/tests/test_local_mode.py +++ b/tests/test_local_mode.py @@ -91,172 +91,81 @@ def test_verified_failure(self): class TestGetLastBashCommandResult: - """Tests for get_last_bash_command_result function.""" + """Exercise legacy verification through the public SDK's HTTP transport.""" @pytest.mark.asyncio - async def test_handles_http_error(self): - """Returns error result when HTTP request fails.""" - from unittest.mock import AsyncMock, MagicMock - + @pytest.mark.parametrize("status", [404, 429]) + async def test_handles_http_errors(self, status): import httpx - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( - "Not found", request=MagicMock(), response=MagicMock(status_code=404) - ) - mock_client.get = AsyncMock(return_value=mock_response) - - result = await get_last_bash_command_result( - mock_client, "http://localhost:3000", "test-key" - ) - - assert result.found is False - assert result.error is not None and "Not found" in result.error - - @pytest.mark.asyncio - async def test_handles_transient_rate_limit(self): - """Returns structured transient info for retryable agent-server errors.""" - from unittest.mock import AsyncMock, MagicMock - - import httpx - - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( - "Rate limited", request=MagicMock(), response=MagicMock(status_code=429) - ) - mock_client.get = AsyncMock(return_value=mock_response) - - result = await get_last_bash_command_result( - mock_client, "http://localhost:3000", "test-key" - ) - - assert result.found is False - assert result.error is not None and "HTTP 429" in result.error - assert result.error_info is not None - assert ( - result.error_info.fingerprint - == "agent_server:bash_events_search:rate_limited:429" - ) - - @pytest.mark.asyncio - async def test_handles_empty_response(self): - """Returns error result when no bash output found.""" - from unittest.mock import AsyncMock, MagicMock - - import httpx - - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_response.json.return_value = {"items": []} - mock_client.get = AsyncMock(return_value=mock_response) - - result = await get_last_bash_command_result( - mock_client, "http://localhost:3000", "test-key" - ) - + async with httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: httpx.Response(status)) + ) as client: + result = await get_last_bash_command_result( + client, "http://localhost:3000", "test-key" + ) assert result.found is False - assert result.error == "No bash output found" - - @pytest.mark.asyncio - async def test_handles_running_command(self): - """Returns running result when exit_code is None.""" - from unittest.mock import AsyncMock, MagicMock - - import httpx - - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_response.json.return_value = { - "items": [{"exit_code": None, "stdout": "", "stderr": ""}] - } - mock_client.get = AsyncMock(return_value=mock_response) - - result = await get_last_bash_command_result( - mock_client, "http://localhost:3000", "test-key" - ) - - assert result.found is True - assert result.exit_code is None - assert result.error == "Command still running" - - @pytest.mark.asyncio - async def test_handles_completed_command(self): - """Returns completed result with exit code and output.""" - from unittest.mock import AsyncMock, MagicMock - - import httpx - - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_response.json.return_value = { - "items": [{"exit_code": 0, "stdout": "Hello", "stderr": ""}] - } - mock_client.get = AsyncMock(return_value=mock_response) - - result = await get_last_bash_command_result( - mock_client, "http://localhost:3000", "test-key" - ) - - assert result.found is True - assert result.exit_code == 0 - assert result.stdout == "Hello" + assert result.error and str(status) in result.error + if status == 429: + assert result.error_info is not None + assert ( + result.error_info.fingerprint + == "agent_server:bash_events_search:rate_limited:429" + ) + else: + assert result.error_info is None @pytest.mark.asyncio - async def test_adds_command_id_filter_when_provided(self): - """When command_id is provided, params include command_id__eq.""" - from unittest.mock import AsyncMock, MagicMock - + @pytest.mark.parametrize( + "items,expected", + [ + ([], BashCommandResult(found=False, error="No bash output found")), + ( + [{"exit_code": None}], + BashCommandResult(found=True, error="Command still running"), + ), + ( + [{"exit_code": 0, "stdout": "Hello", "stderr": ""}], + BashCommandResult(found=True, exit_code=0, stdout="Hello"), + ), + ], + ) + async def test_command_result_states(self, items, expected): import httpx - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_response.json.return_value = {"items": []} - mock_client.get = AsyncMock(return_value=mock_response) - - await get_last_bash_command_result( - mock_client, - "http://localhost:3000", - "test-key", - command_id="abc123", - ) - - # Verify the request was made with command_id__eq in params - mock_client.get.assert_called_once() - _, kwargs = mock_client.get.call_args - assert kwargs["params"]["command_id__eq"] == "abc123" - assert kwargs["params"]["kind__eq"] == "BashOutput" - assert kwargs["params"]["sort_order"] == "TIMESTAMP_DESC" - assert kwargs["params"]["limit"] == 1 + async with httpx.AsyncClient( + transport=httpx.MockTransport( + lambda _: httpx.Response(200, json={"items": items}) + ) + ) as client: + result = await get_last_bash_command_result( + client, "http://localhost:3000", "test-key" + ) + assert result == expected @pytest.mark.asyncio - async def test_omits_command_id_filter_when_none(self): - """When command_id is None, params do NOT include command_id__eq.""" - from unittest.mock import AsyncMock, MagicMock - + @pytest.mark.parametrize("command_id", [None, "abc123"]) + async def test_correlates_the_selected_command(self, command_id): import httpx - mock_client = MagicMock(spec=httpx.AsyncClient) - mock_response = MagicMock() - mock_response.raise_for_status = MagicMock() - mock_response.json.return_value = {"items": []} - mock_client.get = AsyncMock(return_value=mock_response) + requests = [] - await get_last_bash_command_result( - mock_client, - "http://localhost:3000", - "test-key", - ) + def respond(request): + requests.append(request) + return httpx.Response(200, json={"items": []}) - mock_client.get.assert_called_once() - _, kwargs = mock_client.get.call_args - assert "command_id__eq" not in kwargs["params"] - assert kwargs["params"]["kind__eq"] == "BashOutput" + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + await get_last_bash_command_result( + client, "http://localhost:3000", "test-key", command_id=command_id + ) + assert len(requests) == 1 + request = requests[0] + assert request.url.path == "/api/bash/bash_events/search" + assert request.headers["X-Session-API-Key"] == "test-key" + assert request.url.params.get("command_id__eq") == command_id + assert request.url.params["kind__eq"] == "BashOutput" + assert request.url.params["sort_order"] == "TIMESTAMP_DESC" + assert request.url.params["limit"] == "1" class TestVerifyRunOnAgentServer: