feat: dispatch automation bundles through SDK conversation runtimes

README.md

@@ -19,6 +19,15 @@ The Automation Service owns automation definitions, cron scheduling, webhooks, r
 
 ## Development
 
+### Conversation-scoped execution
+
+Definitions default to `execution_scope: "run"`. A definition with
+`execution_scope: "conversation"` and an `agent_profile_id` runs its bundle in
+that Agent Server conversation. The server decides whether the conversation
+workspace is local or Docker-backed; the bundle receives the same environment
+in either case. Docker-backed bundles receive only the conversation runtime key,
+and the watchdog releases the runtime after the command finishes.
+
 ### Prerequisites
 
 - Python 3.12+

migrations/versions/025_add_agent_profile.py

@@ -0,0 +1,34 @@
+"""Persist automation execution settings on definitions and queued runs.
+
+Revision ID: 025
+Revises: 024
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "025"
+down_revision = "024"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    for table in ("automations", "automation_runs"):
+        op.add_column(table, sa.Column("agent_profile_id", sa.Uuid(), nullable=True))
+        op.add_column(
+            table,
+            sa.Column(
+                "execution_scope",
+                sa.String(20),
+                nullable=False,
+                server_default="run",
+            ),
+        )
+
+
+def downgrade() -> None:
+    for table in ("automation_runs", "automations"):
+        op.drop_column(table, "execution_scope")
+        op.drop_column(table, "agent_profile_id")

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 ConversationBackend
 from openhands.automation.backends.local import LocalAgentServerBackend
 
 
@@ -43,15 +44,21 @@ def get_backend(run: AutomationRun) -> ExecutionBackend:
     config = get_config()
     settings = config.service
 
+    if run.execution_scope == "conversation" and not settings.is_local_mode:
+        raise ValueError("Conversation execution requires an existing Agent Server")
+
     if settings.is_local_mode:
-        return LocalAgentServerBackend(
+        server = LocalAgentServerBackend(
             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 run.execution_scope == "conversation":
+            return ConversationBackend(server, run)
+        return server
     else:
         return CloudSandboxBackend(
             api_url=settings.openhands_api_base_url,
@@ -63,6 +70,7 @@ def get_backend(run: AutomationRun) -> ExecutionBackend:
     "ExecutionBackend",
     "ExecutionContext",
     "CloudSandboxBackend",
+    "ConversationBackend",
     "LocalAgentServerBackend",
     "get_backend",
 ]

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,175 @@
+"""Execute an automation bundle inside an Agent Server conversation."""
+
+from __future__ import annotations
+
+import asyncio
+from uuid import UUID
+
+import httpx
+
+from openhands.automation.backends.base import ExecutionBackend, ExecutionContext
+from openhands.automation.backends.local import LocalAgentServerBackend
+from openhands.automation.models import AutomationRun
+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 RemoteConversationControl
+from openhands.sdk.conversation.request import StartConversationRequest
+from openhands.sdk.workspace import (
+    AsyncRemoteWorkspace,
+    LocalWorkspace,
+    RemoteWorkspace,
+)
+
+
+class ConversationBackend(ExecutionBackend):
+    """Add conversation lifecycle to an existing Agent Server backend."""
+
+    def __init__(self, server: LocalAgentServerBackend, run: AutomationRun) -> None:
+        if run.agent_profile_id is None:
+            raise ValueError("Conversation execution requires an agent profile")
+        self.server = server
+        self.run = run
+        self.runtime_api_key = ""
+        self.runtime_kind: str | None = None
+
+    @property
+    def is_local_mode(self) -> bool:
+        return True
+
+    @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.server.agent_server_url,
+                api_key=self.server.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.server.agent_server_url,
+            api_key=self.server.api_key,
+            working_dir=self.get_work_dir(str(self.run.id)),
+        )
+        try:
+            RemoteConversationControl.create(
+                workspace=workspace,
+                request=StartConversationRequest(
+                    workspace=LocalWorkspace(working_dir=workspace.working_dir),
+                    conversation_id=self.conversation_id,
+                    agent_profile_id=self.run.agent_profile_id,
+                    tags={"automationrun": str(self.run.id)},
+                ),
+            )
+        finally:
+            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.server.api_key if runtime_kind == "local" else ""
+        context = ExecutionContext(
+            agent_url=self.server.agent_server_url,
+            session_key=self.server.api_key,
+            runtime_conversation_id=self.conversation_id,
+        )
+        if runtime_kind == "docker":
+            try:
+                async with AsyncRemoteWorkspace(
+                    host=context.agent_url,
+                    api_key=self.server.api_key,
+                    working_dir="/",
+                    runtime_conversation_id=self.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
+
+    async def release_context(
+        self,
+        client: httpx.AsyncClient,  # noqa: ARG002
+        ctx: ExecutionContext,
+    ) -> None:
+        if await self._resolve_runtime() == "local":
+            return
+        async with AsyncRemoteWorkspace(
+            host=ctx.agent_url,
+            api_key=self.server.api_key,
+            working_dir="/",
+            runtime_conversation_id=self.conversation_id,
+        ) as workspace:
+            await workspace.release_runtime()
+
+    async def get_api_key(self) -> str:
+        return await self.server.get_api_key()
+
+    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.server.sandbox_agent_server_url
+                or (
+                    "http://127.0.0.1:8000"
+                    if self.runtime_kind == "docker"
+                    else self.server.agent_server_url
+                )
+            ),
+            "AUTOMATION_AGENT_PROFILE_ID": str(self.run.agent_profile_id),
+            "AUTOMATION_CONVERSATION_ID": str(self.conversation_id),
+            "SESSION_API_KEY": self.runtime_api_key,
+            "WORKSPACE_BASE": self.get_work_dir(str(self.run.id)),
+        }
+
+    async def verify_run(self, run_id: str) -> VerificationResult:
+        return await verify_run_on_agent_server(
+            agent_url=self.server.agent_server_url,
+            session_key=self.server.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.server.agent_server_url,
+                    self.server.api_key,
+                    runtime_conversation_id=self.conversation_id,
+                ),
+            )
+
+    def get_work_dir(self, run_id: str) -> str:
+        if self.runtime_kind is None:
+            raise RuntimeError("Resolve the runtime before selecting a workspace")
+        if self.runtime_kind == "docker":
+            return "/workspace"
+        return self.server.get_work_dir(run_id)

openhands/automation/capabilities_router.py

@@ -46,7 +46,10 @@
 )
 from openhands.automation.trigger_matcher import matches_trigger
 from openhands.automation.utils.cron import min_interval_seconds
-from openhands.automation.utils.model_profiles import validate_model_profile_for_user
+from openhands.automation.utils.model_profiles import (
+    validate_agent_profile_selection,
+    validate_model_profile_for_user,
+)
 from openhands.automation.utils.webhook import get_webhook_config
 
 
@@ -119,6 +122,8 @@ async def get_capabilities(
     event_sources = sorted({*builtin, *await _custom_sources(user.org_id, session)})
 
     features = [*_STATIC_FEATURES]
+    if config.service.is_local_mode:
+        features.append("agentProfiles")
     if event_sources:
         features.append("webhookDelivery")
     if config.kv.enabled:
@@ -182,6 +187,17 @@ async def validate_draft(
             )
         )
 
+    try:
+        validate_agent_profile_selection(draft.agent_profile_id, draft.model)
+    except HTTPException as e:
+        errors.append(
+            DraftValidationError(
+                field="agent_profile_id",
+                code="invalid_agent_profile",
+                message=str(e.detail),
+            )
+        )
+
     trigger = draft.trigger
     if isinstance(trigger, CronTrigger):
         errors.extend(_cron_errors(trigger))