feat: dispatch automation bundles through SDK conversation runtimes

README.md

@@ -19,6 +19,33 @@ The Automation Service owns automation definitions, cron scheduling, webhooks, r
 
 ## Development
 
+### 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.
+
+Conversation completion is detected by the watchdog through the scoped SDK
+runtime on each scan (`AUTOMATION_WATCHDOG_INTERVAL_SECONDS`, default 60 seconds).
+This is the primary completion path for profile-backed runs: workers deliberately
+do not receive the shared Automation callback credential. Legacy backends retain
+their callback path.
+
+`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.
+
 ### Prerequisites
 
 - Python 3.12+
@@ -101,3 +128,14 @@ containers/          # Docker configuration
 ## Deployment
 
 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.
+
+### SDK Integration Dependency
+
+The conversation backend and Agent Server execution helpers reuse `RemoteConversation` and `RemoteWorkspace`
+from [software-agent-sdk #5010](https://github.com/OpenHands/software-agent-sdk/pull/5010).
+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 SDK dependency.
+Workflow bundles
+receive the same environment contract in local and Docker workspaces.

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.local import LocalAgentServerBackend
 
 
@@ -44,14 +45,22 @@ def get_backend(run: AutomationRun) -> ExecutionBackend:
     settings = config.service
 
     if settings.is_local_mode:
-        return LocalAgentServerBackend(
+        backend_type = (
+            ConversationAgentServerBackend
+            if settings.agent_profile
+            else LocalAgentServerBackend
+        )
+        backend = backend_type(
             agent_server_url=settings.agent_server_url,
             api_key=settings.agent_server_api_key,
             run=run,
             workspace_base=settings.workspace_base,
             callback_api_key=settings.local_api_key,
             sandbox_agent_server_url=settings.sandbox_agent_server_url or None,
         )
+        if isinstance(backend, ConversationAgentServerBackend):
+            backend.agent_profile_id = settings.agent_profile
+        return backend
     else:
         return CloudSandboxBackend(
             api_url=settings.openhands_api_base_url,

openhands/automation/backends/base.py

@@ -3,6 +3,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
+from uuid import UUID
 
 import httpx
 
@@ -28,6 +29,7 @@ class ExecutionContext:
     sandbox_id: str | None = None
     api_url: str | None = None
     api_key: str | None = None
+    runtime_conversation_id: UUID | None = None
 
 
 class ExecutionBackend(ABC):

openhands/automation/backends/conversation.py

@@ -0,0 +1,164 @@
+"""One bundle-facing conversation contract for local and Docker workspaces."""
+
+from __future__ import annotations
+
+import asyncio
+from uuid import UUID
+
+import httpx
+
+from openhands.automation.backends.base import ExecutionContext
+from openhands.automation.backends.local import LocalAgentServerBackend
+from openhands.automation.subjects import conversation_id_for
+from openhands.automation.utils.agent_server import (
+    VerificationResult,
+    verify_run_on_agent_server,
+)
+from openhands.sdk import RemoteConversation
+from openhands.sdk.conversation.request import StartConversationRequest
+from openhands.sdk.workspace import (
+    AsyncRemoteWorkspace,
+    LocalWorkspace,
+    RemoteWorkspace,
+)
+
+
+class ConversationAgentServerBackend(LocalAgentServerBackend):
+    agent_profile_id: str
+    runtime_api_key: str = ""
+    _runtime_kind: str | None = None
+
+    @property
+    def conversation_id(self) -> UUID:
+        automation = self._run.automation
+        source = (automation.trigger or {}).get("source")
+        if self._run.subject_key and source:
+            return UUID(
+                conversation_id_for(
+                    automation.org_id, automation.id, source, self._run.subject_key
+                )
+            )
+        return self._run.id
+
+    async def _resolve_runtime(self) -> str:
+        if self._runtime_kind is None:
+            async with AsyncRemoteWorkspace(
+                host=self.agent_server_url,
+                api_key=self.api_key,
+                working_dir="/",
+            ) as workspace:
+                info = await workspace.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
+
+    def _create_conversation(self) -> None:
+        workspace = RemoteWorkspace(
+            host=self.agent_server_url,
+            api_key=self.api_key,
+            working_dir=self.get_work_dir(str(self._run.id)),
+        )
+        conversation = None
+        try:
+            conversation = RemoteConversation.create(
+                workspace=workspace,
+                request=StartConversationRequest(
+                    workspace=LocalWorkspace(working_dir=workspace.working_dir),
+                    conversation_id=self.conversation_id,
+                    agent_profile_id=UUID(self.agent_profile_id),
+                    max_iterations=160,
+                    tags={"automationrun": str(self._run.id)},
+                ),
+                visualizer=None,
+            )
+            conversation.set_title(self._run.automation.name)
+        finally:
+            if conversation is not None:
+                conversation.close()
+            workspace.reset_client()
+
+    async def get_execution_context(
+        self, client: httpx.AsyncClient
+    ) -> ExecutionContext:
+        runtime_kind = await self._resolve_runtime()
+        await asyncio.to_thread(self._create_conversation)
+        self.runtime_api_key = self.api_key if runtime_kind == "local" else ""
+        context = ExecutionContext(
+            agent_url=self.agent_server_url,
+            session_key=self.api_key,
+            runtime_conversation_id=self.conversation_id,
+        )
+        if runtime_kind == "docker":
+            try:
+                async with AsyncRemoteWorkspace(
+                    host=context.agent_url,
+                    api_key=self.api_key,
+                    working_dir="/",
+                    runtime_conversation_id=context.runtime_conversation_id,
+                ) as workspace:
+                    self.runtime_api_key = await workspace.get_runtime_session_key()
+            except Exception:
+                await self.release_context(client, context)
+                raise
+        return context
+
+    def build_env_vars(self) -> dict[str, str]:
+        # Completion is polled through the scoped SDK runtime. Never expose the
+        # shared Automation callback key to a profile-scoped worker.
+        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.conversation_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,  # noqa: ARG002
+        ctx: ExecutionContext,
+    ) -> None:
+        if await self._resolve_runtime() == "local":
+            return  # Persistent server and conversation history belong to the host.
+        async with AsyncRemoteWorkspace(
+            host=ctx.agent_url,
+            api_key=self.api_key,
+            working_dir="/",
+            runtime_conversation_id=self.conversation_id,
+        ) as workspace:
+            await workspace.release_runtime()
+
+    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,
+            runtime_conversation_id=self.conversation_id,
+        )
+
+    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)
+            )

openhands/automation/config.py

@@ -581,6 +581,10 @@ 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 = ""
+    conversation_max_concurrent_runs: int = Field(default=2, ge=1)
+
     # 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

openhands/automation/dispatcher.py

@@ -22,7 +22,7 @@
 from typing import Any
 
 import httpx
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from sqlalchemy.orm import selectinload
 
@@ -131,6 +131,28 @@ async def _poll_pending_runs(
     Eagerly loads the ``automation`` relationship so that ``user_id``,
     ``org_id``, and tarball config are available for dispatch.
     """
+    run_profile = get_config().service.agent_profile
+    active = []
+    if run_profile:
+        active = (
+            (
+                await session.execute(
+                    select(AutomationRun.automation_id).where(
+                        AutomationRun.status == AutomationRunStatus.RUNNING
+                    )
+                )
+            )
+            .scalars()
+            .all()
+        )
+        batch_size = min(
+            batch_size,
+            1,
+            get_config().service.conversation_max_concurrent_runs - len(active),
+        )
+        if batch_size <= 0:
+            return []
+
     select_query = (
         select(AutomationRun)
         .join(AutomationRun.automation)
@@ -143,6 +165,8 @@ async def _poll_pending_runs(
         .order_by(AutomationRun.created_at.asc())
         .limit(batch_size)
     )
+    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)
     if not using_sqlite():
@@ -207,8 +231,9 @@ async def _execute_run(
     4. Execute in context (upload tarball, start entrypoint)
     5. Store sandbox_id for watchdog verification (if applicable)
 
-    The SDK inside the execution environment fires the completion callback on exit.
-    The watchdog will verify status if the callback is missed.
+    Legacy backends receive SDK completion callbacks, with watchdog fallback.
+    Profile-backed conversations use watchdog verification through the SDK
+    runtime so workers do not need the shared Automation callback credential.
     """
     run_id = str(run.id)
     automation = run.automation
@@ -326,10 +351,17 @@ async def _fail(
     # 3. Build env vars (must be after get_execution_context for cloud mode API key)
     callback_url = f"{settings.resolved_base_url.rstrip('/')}/v1/runs/{run_id}/complete"
     env_vars = backend.build_env_vars()
-    env_vars["AUTOMATION_CALLBACK_URL"] = callback_url
-    env_vars["AUTOMATION_PHASE_URL"] = (
-        f"{settings.resolved_base_url.rstrip('/')}/v1/runs/{run_id}/phase"
-    )
+    # Callbacks are optional. A restricted backend without callback credentials
+    # uses the existing runtime watchdog instead of receiving the service key.
+    if (
+        env_vars.get("AUTOMATION_CALLBACK_API_KEY")
+        or env_vars.get("OPENHANDS_API_KEY")
+        or not settings.local_api_key
+    ):
+        env_vars["AUTOMATION_CALLBACK_URL"] = callback_url
+        env_vars["AUTOMATION_PHASE_URL"] = (
+            f"{settings.resolved_base_url.rstrip('/')}/v1/runs/{run_id}/phase"
+        )
     env_vars["AUTOMATION_RUN_ID"] = run_id
     env_vars["AUTOMATION_USER_ID"] = str(automation.user_id)
     env_vars["AUTOMATION_ORG_ID"] = str(automation.org_id)
@@ -435,7 +467,6 @@ async def _fail(
     work_dir = backend.get_work_dir(run_id)
     try:
         result = await execute_in_context(
-            client=client,
             agent_url=ctx.agent_url,
             session_key=ctx.session_key,
             entrypoint=automation.entrypoint,
@@ -445,6 +476,7 @@ async def _fail(
             timeout=effective_timeout,
             run_id=run_id,
             sandbox_id=ctx.sandbox_id,
+            runtime_conversation_id=ctx.runtime_conversation_id,
         )
     except PermanentDispatchError as exc:
         logger.error(
@@ -487,6 +519,14 @@ async def _fail(
 
     # 6. Handle result
     if result.success:
+        if ctx.runtime_conversation_id is not None:
+            async with session_factory() as link_session:
+                await link_session.execute(
+                    update(AutomationRun)
+                    .where(AutomationRun.id == run.id)
+                    .values(conversation_id=env_vars["AUTOMATION_CONVERSATION_ID"])
+                )
+                await link_session.commit()
         await update_run_current_phase(session_factory, run.id, "Starting automation")
         if ctx.sandbox_id:
             await update_sandbox_id(session_factory, run.id, ctx.sandbox_id)
@@ -511,7 +551,7 @@ async def _fail(
             ),
         )
         logger.info(
-            "Automation dispatched successfully, waiting for callback",
+            "Automation dispatched successfully, waiting for completion",
             extra=_log_ctx(sandbox_id=ctx.sandbox_id),
         )
         return