fix(standalone): repair in-process execution flow (#780)

* fix(standalone): repair in-process execution flow

* docs(standalone): add in-process executor docstrings

* test(standalone): refine in-process executor coverage

* test(standalone): add docstrings to in-process executor tests

* test(standalone): add docstrings to executor tests

Author: code-wangdi

backend/app/services/execution/inprocess_executor.py

@@ -27,6 +27,7 @@
 )
 from shared.models.responses_api import ResponsesAPIStreamEvents
 from shared.models.responses_api_emitter import EventTransport
+from shared.status import TaskStatus
 
 from .dispatcher import _require_non_empty_tool_use_id, extract_completed_result
 from .emitters import ResultEmitter
@@ -263,11 +264,17 @@ async def execute(
         request: ExecutionRequest,
         emitter: ResultEmitter,
     ) -> None:
-        """Execute task in-process.
+        """Execute one standalone task inside the backend process.
 
         Args:
             request: Execution request
             emitter: Result emitter for event emission
+
+        Raises:
+            RuntimeError: When agent creation fails or the in-process lifecycle
+                returns a terminal failure status. The exception is re-raised
+                after emitting an error event so outer dispatcher layers can
+                keep their normal failure handling.
         """
         logger.info(
             f"[InprocessExecutor] Starting in-process execution: "
@@ -297,15 +304,12 @@ async def execute(
                 .build()
             )
 
-            # Convert request to task_data dict for AgentService
-            task_data = request.to_dict()
-
             # Get or create agent service (singleton)
             agent_service = AgentService()
 
             # Create agent with custom emitter
             # AgentService caches agents by task_id, so same task reuses the same agent
-            agent = agent_service.create_agent(task_data)
+            agent = agent_service.create_agent(request)
             if not agent:
                 raise RuntimeError(f"Failed to create agent for task {request.task_id}")
 
@@ -337,6 +341,12 @@ async def execute(
                 f"task_id={request.task_id}, status={status}, error={error_message}"
             )
 
+            if status == TaskStatus.FAILED:
+                raise RuntimeError(
+                    error_message
+                    or f"In-process execution failed for task {request.task_id}"
+                )
+
             # Note: The agent's emitter will have already sent DONE/ERROR events
             # through the bridge transport, so we don't need to emit them again here
 
@@ -363,28 +373,33 @@ async def _execute_agent_async(
         self,
         agent,
     ) -> tuple:
-        """Execute agent asynchronously in the current event loop.
+        """Run the agent lifecycle in the current event loop.
 
-        This method handles the agent lifecycle (pre_execute, execute, post_execute)
-        similar to Agent.handle(), but uses async methods to stay in the current
-        event loop.
+        This mirrors ``Agent.handle()`` for standalone mode by awaiting
+        ``pre_execute()`` and then using ``execute_async()`` when available.
+        It returns a ``FAILED`` status tuple instead of raising so ``execute()``
+        can centralize frontend error emission and dispatcher propagation.
 
         Args:
             agent: The agent instance to execute
 
         Returns:
-            Tuple of (TaskStatus, error_message)
+            Tuple of ``(TaskStatus, error_message)`` where ``error_message`` is
+            populated only for failure states.
         """
         from shared.status import TaskStatus
 
         try:
-            # Pre-execute phase (sync, but fast)
+            # Mirror Agent.handle() semantics in the current event loop.
             logger.info(
                 f"Agent[{agent.get_name()}][{agent.task_id}] handle: Starting pre_execute."
             )
-            pre_status = agent.pre_execute()
-            if pre_status not in (TaskStatus.SUCCESS, TaskStatus.RUNNING):
-                return pre_status, f"Pre-execute failed with status: {pre_status}"
+            pre_status, pre_error = await agent.pre_execute()
+            if pre_status != TaskStatus.SUCCESS:
+                error_msg = f"Pre-execute failed with status: {pre_status}"
+                if pre_error:
+                    error_msg = f"{error_msg}\n{pre_error}"
+                return TaskStatus.FAILED, error_msg
 
             # Execute phase - use async method if available
             logger.info(
@@ -404,9 +419,6 @@ async def _execute_agent_async(
                 f"Agent[{agent.get_name()}][{agent.task_id}] handle: execute finished with result: {status}"
             )
 
-            # Post-execute phase (sync, but fast)
-            agent.post_execute()
-
             return status, None
 
         except Exception as e:

backend/tests/services/execution/test_inprocess_executor.py

@@ -0,0 +1,121 @@
+import sys
+import types
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.services.execution.inprocess_executor import InprocessExecutor
+from shared.models.execution import ExecutionRequest
+from shared.status import TaskStatus
+
+
+@pytest.mark.asyncio
+async def test_execute_passes_execution_request_to_agent_service() -> None:
+    """Pass an ExecutionRequest object directly to AgentService."""
+    request = ExecutionRequest(task_id=1, subtask_id=2, message_id=3)
+    emitter = AsyncMock()
+    executor = InprocessExecutor()
+
+    agent = MagicMock()
+    agent.get_name.return_value = "fake-agent"
+
+    agent_service = MagicMock()
+    agent_service.create_agent.return_value = agent
+    fake_module = types.ModuleType("executor.services.agent_service")
+    fake_module.AgentService = MagicMock(return_value=agent_service)
+
+    with (
+        patch.dict(sys.modules, {"executor.services.agent_service": fake_module}),
+        patch.object(
+            executor,
+            "_execute_agent_async",
+            new=AsyncMock(return_value=(TaskStatus.SUCCESS, None)),
+        ) as mock_execute_agent,
+        patch.object(
+            executor,
+            "_cleanup_task_clients",
+            new=AsyncMock(),
+        ) as mock_cleanup,
+    ):
+        await executor.execute(request, emitter)
+
+    agent_service.create_agent.assert_called_once()
+    assert agent_service.create_agent.call_args.args[0] is request
+    mock_execute_agent.assert_awaited_once_with(agent)
+    mock_cleanup.assert_awaited_once_with(request.task_id)
+
+
+@pytest.mark.asyncio
+async def test_execute_agent_async_awaits_pre_execute_and_skips_missing_post_execute() -> None:
+    """Await async lifecycle hooks without requiring post_execute()."""
+    executor = InprocessExecutor()
+    agent = SimpleNamespace(
+        task_id=3,
+        get_name=lambda: "fake-agent",
+        pre_execute=AsyncMock(return_value=(TaskStatus.SUCCESS, None)),
+        execute_async=AsyncMock(return_value=TaskStatus.COMPLETED),
+    )
+
+    status, error = await executor._execute_agent_async(agent)
+
+    assert status == TaskStatus.COMPLETED
+    assert error is None
+    agent.pre_execute.assert_awaited_once()
+    agent.execute_async.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_execute_agent_async_returns_failed_when_pre_execute_fails() -> None:
+    """Return FAILED when pre_execute reports a failure status."""
+    executor = InprocessExecutor()
+    agent = SimpleNamespace(
+        task_id=4,
+        get_name=lambda: "fake-agent",
+        pre_execute=AsyncMock(return_value=(TaskStatus.FAILED, "boom")),
+        execute_async=AsyncMock(),
+    )
+
+    status, error = await executor._execute_agent_async(agent)
+
+    assert status == TaskStatus.FAILED
+    assert error is not None
+    assert "Pre-execute failed" in error
+    assert "boom" in error
+    agent.pre_execute.assert_awaited_once()
+    agent.execute_async.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_execute_raises_and_emits_error_when_agent_lifecycle_returns_failed() -> None:
+    """Raise and emit an error when the agent lifecycle returns FAILED."""
+    request = ExecutionRequest(task_id=10, subtask_id=11, message_id=12)
+    emitter = AsyncMock()
+    executor = InprocessExecutor()
+
+    agent = MagicMock()
+    agent.get_name.return_value = "fake-agent"
+
+    agent_service = MagicMock()
+    agent_service.create_agent.return_value = agent
+    fake_module = types.ModuleType("executor.services.agent_service")
+    fake_module.AgentService = MagicMock(return_value=agent_service)
+
+    with (
+        patch.dict(sys.modules, {"executor.services.agent_service": fake_module}),
+        patch.object(
+            executor,
+            "_execute_agent_async",
+            new=AsyncMock(return_value=(TaskStatus.FAILED, "pre-execute failed")),
+        ),
+        patch.object(
+            executor,
+            "_cleanup_task_clients",
+            new=AsyncMock(),
+        ) as mock_cleanup,
+        pytest.raises(RuntimeError, match="pre-execute failed"),
+    ):
+        await executor.execute(request, emitter)
+
+    emitter.emit_error.assert_awaited_once()
+    mock_cleanup.assert_not_awaited()