Improve automation run observability and LLM preflight

openhands/automation/dispatcher.py

@@ -228,6 +228,7 @@ async def _fail(error: str, disable: bool = False) -> None:
     env_vars["AUTOMATION_USER_ID"] = str(automation.user_id)
     env_vars["AUTOMATION_ORG_ID"] = str(automation.org_id)
     env_vars["AUTOMATION_API_URL"] = settings.resolved_base_url
+    env_vars["AUTOMATION_RUN_TIMEOUT"] = str(effective_timeout)
     env_vars["AUTOMATION_EVENT_PAYLOAD"] = json.dumps(
         _build_event_payload(automation, run)
     )

openhands/automation/presets/plugin/sdk_main.py

@@ -70,6 +70,7 @@
 import random
 import sys
 import time
+import urllib.request
 
 # Detect execution mode based on AGENT_SERVER_URL presence
 agent_server_url = os.environ.get("AGENT_SERVER_URL", "").rstrip("/")
@@ -142,6 +143,44 @@ def _conversation_supports_user_id() -> bool:
         return False
 
 
+def _conversation_run_timeout() -> float:
+    raw_timeout = os.environ.get("AUTOMATION_RUN_TIMEOUT")
+    if not raw_timeout:
+        return 3600.0
+    try:
+        outer_timeout = float(raw_timeout)
+    except ValueError:
+        return 3600.0
+    if outer_timeout <= 90:
+        return max(30.0, outer_timeout - 10.0)
+    return outer_timeout - 60.0
+
+
+def _notify_conversation_started(conversation_id: str) -> None:
+    callback_url = os.environ.get("AUTOMATION_CALLBACK_URL")
+    if not callback_url:
+        return
+    if not callback_url.endswith("/complete"):
+        print("  conversation update skipped: callback URL has unexpected shape")
+        return
+    update_url = callback_url[: -len("/complete")] + "/conversation"
+    data = json.dumps({"conversation_id": str(conversation_id)}).encode()
+    request = urllib.request.Request(
+        update_url,
+        data=data,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            print(f"  conversation update sent: {response.status}")
+    except Exception as e:
+        print(f"  conversation update failed: {type(e).__name__}: {e}", file=sys.stderr)
+
+
 # Workspace base directory (for RemoteWorkspace working_dir)
 # Expand ~ to home directory before validation
 workspace_base = os.path.expanduser(os.environ.get("WORKSPACE_BASE", "/workspace"))
@@ -288,16 +327,6 @@ def _conversation_supports_user_id() -> bool:
 
 {USER_PROMPT}"""
 
-    # Deserialize plugin sources using Pydantic validation
-    plugin_sources = [PluginSource.model_validate(p) for p in plugins_config]
-
-    print("\n=== PLUGINS CONFIG ===")
-    print(f"  loading {len(plugin_sources)} plugin(s):")
-    for ps in plugin_sources:
-        ref_str = f"@{ps.ref}" if ps.ref else ""
-        path_str = f" ({ps.repo_path})" if ps.repo_path else ""
-        print(f"    - {ps.source}{ref_str}{path_str}")
-
     # Get LLM config via workspace/profile APIs
     print("\n=== GET_LLM ===")
     try:
@@ -314,6 +343,16 @@ def _conversation_supports_user_id() -> bool:
     print(f"  model: {llm.model}")
     print(f"  api_key present: {bool(llm.api_key)}")
 
+    # Deserialize plugin sources using Pydantic validation.
+    plugin_sources = [PluginSource.model_validate(p) for p in plugins_config]
+
+    print("\n=== PLUGINS CONFIG ===")
+    print(f"  loading {len(plugin_sources)} plugin(s):")
+    for ps in plugin_sources:
+        ref_str = f"@{ps.ref}" if ps.ref else ""
+        path_str = f" ({ps.repo_path})" if ps.repo_path else ""
+        print(f"    - {ps.source}{ref_str}{path_str}")
+
     # Get secrets via workspace
     print("\n=== GET_SECRETS ===")
     secrets = {}
@@ -409,10 +448,12 @@ def event_callback(event) -> None:
         conversation.update_secrets({"AUTOMATION_SESSION_URL": session_url})
         print(f"  session URL: {session_url}")
 
+    _notify_conversation_started(str(conversation.id))
+
     try:
         print(f"  sending prompt: {USER_PROMPT[:80]}...")
         conversation.send_message(USER_PROMPT)
-        conversation.run()
+        conversation.run(timeout=_conversation_run_timeout())
 
         # Wait for the stream to settle
         while time.time() - last_event_time["ts"] < 2.0:

openhands/automation/presets/prompt/sdk_main.py

@@ -73,6 +73,7 @@
 import os
 import sys
 import time
+import urllib.request
 
 # Detect execution mode based on AGENT_SERVER_URL presence
 agent_server_url = os.environ.get("AGENT_SERVER_URL", "").rstrip("/")
@@ -145,6 +146,44 @@ def _conversation_supports_user_id() -> bool:
         return False
 
 
+def _conversation_run_timeout() -> float:
+    raw_timeout = os.environ.get("AUTOMATION_RUN_TIMEOUT")
+    if not raw_timeout:
+        return 3600.0
+    try:
+        outer_timeout = float(raw_timeout)
+    except ValueError:
+        return 3600.0
+    if outer_timeout <= 90:
+        return max(30.0, outer_timeout - 10.0)
+    return outer_timeout - 60.0
+
+
+def _notify_conversation_started(conversation_id: str) -> None:
+    callback_url = os.environ.get("AUTOMATION_CALLBACK_URL")
+    if not callback_url:
+        return
+    if not callback_url.endswith("/complete"):
+        print("  conversation update skipped: callback URL has unexpected shape")
+        return
+    update_url = callback_url[: -len("/complete")] + "/conversation"
+    data = json.dumps({"conversation_id": str(conversation_id)}).encode()
+    request = urllib.request.Request(
+        update_url,
+        data=data,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            print(f"  conversation update sent: {response.status}")
+    except Exception as e:
+        print(f"  conversation update failed: {type(e).__name__}: {e}", file=sys.stderr)
+
+
 # Workspace base directory (for RemoteWorkspace working_dir).
 # Default is /workspace because preset scripts run inside containers where this path
 # is guaranteed to exist. For native local mode (outside containers), the dispatcher
@@ -282,7 +321,6 @@ def _conversation_supports_user_id() -> bool:
     print(f"  profile: {model_profile or 'DEFAULT'}")
     print(f"  model: {llm.model}")
     print(f"  api_key present: {bool(llm.api_key)}")
-
     # Get secrets via workspace
     print("\n=== GET_SECRETS ===")
     secrets = {}
@@ -358,10 +396,12 @@ def event_callback(event) -> None:
         conversation.update_secrets({"AUTOMATION_SESSION_URL": session_url})
         print(f"  session URL: {session_url}")
 
+    _notify_conversation_started(str(conversation.id))
+
     try:
         print(f"  sending prompt: {USER_PROMPT[:80]}...")
         conversation.send_message(USER_PROMPT)
-        conversation.run()
+        conversation.run(timeout=_conversation_run_timeout())
 
         # Wait for the stream to settle
         while time.time() - last_event_time["ts"] < 2.0:

openhands/automation/router.py

@@ -28,6 +28,7 @@
     AutomationRunResponse,
     CreateAutomationRequest,
     RunCompleteRequest,
+    RunConversationRequest,
     UpdateAutomationRequest,
 )
 from openhands.automation.storage import FileStore, get_file_store
@@ -318,6 +319,50 @@ async def list_automation_runs(
     )
 
 
+# --- Run conversation metadata ---
+
+@router.post("/runs/{run_id}/conversation")
+async def update_run_conversation(
+    run_id: uuid.UUID,
+    body: RunConversationRequest,
+    user: AuthenticatedUser = Depends(_require_manage_automations),
+    session: AsyncSession = Depends(get_session),
+) -> AutomationRunResponse:
+    """Associate a live conversation with an automation run.
+
+    Preset scripts call this immediately after creating the RemoteConversation so
+    the activity log can link to running or watchdog-failed runs even if the
+    process is later killed before the terminal completion callback fires.
+    """
+    result = await session.execute(
+        select(AutomationRun)
+        .where(AutomationRun.id == run_id)
+        .options(selectinload(AutomationRun.automation))
+    )
+    run = result.scalars().first()
+    if run is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Run not found")
+
+    automation = run.automation
+    if automation.user_id != user.user_id or automation.org_id != user.org_id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Not your automation")
+
+    if run.status not in {
+        AutomationRunStatus.PENDING,
+        AutomationRunStatus.RUNNING,
+    }:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            detail=f"Run is {run.status.value}, expected PENDING or RUNNING",
+        )
+
+    run.conversation_id = body.conversation_id
+    await session.flush()
+    await session.refresh(run)
+    logger.info("Run %s conversation_id=%s", run_id, body.conversation_id)
+    return AutomationRunResponse.model_validate(run)
+
+
 # --- Run completion callback ---
 
 

openhands/automation/schemas.py

@@ -625,6 +625,14 @@ class RunCompleteRequest(BaseModel):
     error: str | None = None
 
 
+class RunConversationRequest(BaseModel):
+    """Payload for associating a live conversation with a run."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    conversation_id: str = Field(..., min_length=1, max_length=255)
+
+
 class AutomationRunResponse(BaseModel):
     """Response for a single automation run."""
 

tests/test_preset_router.py

@@ -184,6 +184,11 @@ def test_generate_tarball_main_py_content(self):
             assert "RemoteWorkspace" in main_content
             assert "workspace.get_llm(profile_name=model_profile)" in main_content
             assert "falling back to active/default profile" in main_content
+            assert "_notify_conversation_started(str(conversation.id))" in main_content
+            assert (
+                "conversation.run(timeout=_conversation_run_timeout())"
+                in main_content
+            )
             assert "workspace.get_secrets()" in main_content
             assert "workspace.get_mcp_config()" in main_content
             assert "workspace.clone_repos" in main_content
@@ -860,6 +865,11 @@ def test_generate_plugin_tarball_main_py_content(self):
             assert "RemoteWorkspace" in main_content
             assert "workspace.get_llm(profile_name=model_profile)" in main_content
             assert "falling back to active/default profile" in main_content
+            assert "_notify_conversation_started(str(conversation.id))" in main_content
+            assert (
+                "conversation.run(timeout=_conversation_run_timeout())"
+                in main_content
+            )
             assert "workspace.get_secrets()" in main_content
             assert "workspace.clone_repos" in main_content
             assert "workspace.load_skills_from_agent_server" in main_content

tests/test_router.py

@@ -1864,6 +1864,80 @@ async def test_complete_run_not_running_returns_409(
         assert "PENDING" in response.json()["detail"]
 
 
+class TestUpdateRunConversation:
+    """Tests for POST /runs/{run_id}/conversation endpoint."""
+
+    async def test_update_run_conversation_for_running_run(
+        self, async_client, async_session
+    ):
+        """A running run can be linked to its conversation before completion."""
+        from openhands.automation.models import AutomationRun, AutomationRunStatus
+
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Test Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.RUNNING,
+        )
+        async_session.add(run)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/runs/{run.id}/conversation",
+            json={"conversation_id": "conv-live-123"},
+        )
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "RUNNING"
+        assert data["conversation_id"] == "conv-live-123"
+
+        await async_session.refresh(run)
+        assert run.conversation_id == "conv-live-123"
+        assert run.status == AutomationRunStatus.RUNNING
+
+    async def test_update_run_conversation_rejects_terminal_run(
+        self, async_client, async_session
+    ):
+        """Terminal runs are not mutated by the live conversation endpoint."""
+        from openhands.automation.models import AutomationRun, AutomationRunStatus
+
+        automation = Automation(
+            user_id=TEST_USER_ID,
+            org_id=TEST_ORG_ID,
+            name="Test Automation",
+            trigger={"type": "cron", "schedule": "0 9 * * *", "timezone": "UTC"},
+            tarball_path="s3://bucket/code.tar.gz",
+            entrypoint="uv run script.py",
+        )
+        async_session.add(automation)
+        await async_session.commit()
+
+        run = AutomationRun(
+            automation_id=automation.id,
+            status=AutomationRunStatus.FAILED,
+        )
+        async_session.add(run)
+        await async_session.commit()
+
+        response = await async_client.post(
+            f"/api/automation/v1/runs/{run.id}/conversation",
+            json={"conversation_id": "conv-too-late"},
+        )
+
+        assert response.status_code == 409
+        assert "FAILED" in response.json()["detail"]
+
+
 class TestDownloadTarball:
     """Tests for GET /{automation_id}/tarball endpoint."""