UN-4318 [FEAT] Show stuck executions and let owners abort running ones

backend/api_v2/api_deployment_views.py

@@ -29,6 +29,7 @@
 from utils.hubspot_notify import notify_hubspot_event
 from utils.pagination import CustomPagination
 from workflow_manager.workflow_v2.dto import ExecutionResponse
+from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 
 from api_v2.api_deployment_dto_registry import ApiDeploymentDTORegistry
@@ -171,7 +172,10 @@ def post(
 
         # Determine response status based on execution result
         execution_status = response.get("execution_status", "")
-        has_error = response.get("error") or execution_status == "ERROR"
+        has_error = response.get("error") or execution_status in (
+            ExecutionStatus.ERROR.value,
+            ExecutionStatus.STOPPED.value,
+        )
 
         if has_error:
             # Check for tool not found in registry error - return 500 Internal Server Error
@@ -253,6 +257,10 @@ def get(
         response_status = status.HTTP_422_UNPROCESSABLE_ENTITY
         if execution_status_value == CeleryTaskState.COMPLETED.value:
             response_status = status.HTTP_200_OK
+        if execution_status_value in (
+            CeleryTaskState.COMPLETED.value,
+            ExecutionStatus.STOPPED.value,
+        ):
             DeploymentHelper.process_completed_execution(
                 response=response,
                 deployment_execution_dto=deployment_execution_dto,

backend/api_v2/deployment_helper.py

@@ -296,10 +296,13 @@ def execute_workflow(
             )
         except Exception as error:
             # Isolate the DB error-marking so cleanup still runs if it raises.
+            stored_status = ExecutionStatus.ERROR.value
             try:
-                WorkflowExecutionServiceHelper.update_execution_err(
+                marked = WorkflowExecutionServiceHelper.update_execution_err(
                     str(execution_id), str(error)
                 )
+                # An aborted run stays STOPPED; answer with what is stored.
+                stored_status = marked.status if marked else stored_status
             except Exception:
                 logger.exception(f"Failed to mark execution {execution_id} as ERROR")
 
@@ -317,7 +320,7 @@ def execute_workflow(
                 ExecutionResponse(
                     workflow_id=workflow_id,
                     execution_id=execution_id,
-                    execution_status=ExecutionStatus.ERROR.value,
+                    execution_status=stored_status,
                     error=str(error),
                 )
             ).data
@@ -360,6 +363,11 @@ def execute_workflow(
                         f"Failed to acknowledge results for execution {execution_id}"
                     )
 
+            # An aborted run was already reported by the abort.
+            if (
+                execution is not None
+                and execution.status == ExecutionStatus.COMPLETED.value
+            ):
                 try:
                     # Terminalising here bypasses the usual notifier, so subscribers
                     # would otherwise hear nothing.

backend/api_v2/openapi_schema.py

@@ -246,7 +246,9 @@ def _get_error_response_examples(self) -> list[OpenApiExample]:
     "later call for that execution answers 406. Keep the payload of the call "
     "that returns it — it cannot be fetched again.\n\n"
     "A still-running execution answers 422 carrying its current `status`, so a "
-    "polling loop should treat 422 as the normal reply and stop on 200."
+    "polling loop should treat 422 as the normal reply and stop on 200, or when "
+    "`status` is `ERROR` or `STOPPED`. Those also answer 422 but the run has ended, "
+    "and that read is the only one that carries its results."
 )
 
 
@@ -263,7 +265,9 @@ def _get_error_response_examples(self) -> list[OpenApiExample]:
         responses={
             200: ExecuteResponse,
             422: OpenApiResponse(
-                ExecuteResponse, description="The execution finished with an error."
+                ExecuteResponse,
+                description="The execution finished with an error, or was aborted "
+                "(`STOPPED`).",
             ),
             500: OpenApiResponse(
                 ExecuteResponse,
@@ -289,8 +293,8 @@ def _get_error_response_examples(self) -> list[OpenApiExample]:
             ),
             422: OpenApiResponse(
                 StatusResponse,
-                description="The execution is still running, or it finished with an "
-                "error; read `status` to tell them apart.",
+                description="The execution is still running, finished with an error, "
+                "or was aborted (`STOPPED`); read `status` to tell them apart.",
             ),
             500: OpenApiResponse(
                 StatusResponse,

backend/api_v2/tests/test_deployment_aborted.py

@@ -0,0 +1,111 @@
+"""API deployment calls report an aborted run as an error (UN-4318)."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from account_v2.models import Organization
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import TestCase
+from rest_framework.response import Response
+from rest_framework.test import APIRequestFactory
+from utils.user_context import UserContext
+from workflow_manager.workflow_v2.dto import ExecutionResponse
+from workflow_manager.workflow_v2.models.execution import WorkflowExecution
+from workflow_manager.workflow_v2.models.workflow import Workflow
+
+from api_v2.api_deployment_views import DeploymentExecution
+from api_v2.models import APIDeployment, APIKey
+
+ORG_ID = "org-aborted"
+
+
+class SyncCallForAbortedRunTest(TestCase):
+    def setUp(self) -> None:
+        Organization.objects.create(
+            name=ORG_ID, display_name="Org", organization_id=ORG_ID
+        )
+        UserContext.set_organization_identifier(ORG_ID)
+        self.workflow = Workflow.objects.create(workflow_name="wf", is_active=True)
+        self.api = APIDeployment.objects.create(api_name="api", workflow=self.workflow)
+        self.key = APIKey.objects.create(api=self.api)
+
+    def _post_with_status(self, execution_status: str) -> Response:
+        request = APIRequestFactory().post(
+            f"/deployment/api/{ORG_ID}/api/",
+            {"files": SimpleUploadedFile("doc.txt", b"hello")},
+            format="multipart",
+            HTTP_AUTHORIZATION=f"Bearer {self.key.api_key}",
+        )
+        with (
+            patch(
+                "api_v2.api_deployment_views.APIDeploymentRateLimiter.check_and_acquire",
+                return_value=(True, {}),
+            ),
+            patch(
+                "api_v2.api_deployment_views.DeploymentHelper.execute_workflow",
+                return_value={"execution_status": execution_status},
+            ),
+        ):
+            return DeploymentExecution.as_view()(
+                request, org_name=ORG_ID, api_name="api"
+            )
+
+    def test_status_codes(self) -> None:
+        for execution_status, code in (
+            ("STOPPED", 422),
+            ("ERROR", 422),
+            ("COMPLETED", 200),
+        ):
+            with self.subTest(execution_status=execution_status):
+                self.assertEqual(
+                    self._post_with_status(execution_status).status_code, code
+                )
+
+    def _get_stopped_result(self) -> Response:
+        item = {
+            "file": "doc.txt",
+            "metadata": {"workflow_id": "w", "highlight_data": [1]},
+            "result": {
+                "output": {"a": 1},
+                "metadata": {"inner": 1},
+                "metrics": {"llm": 1},
+            },
+        }
+        execution = WorkflowExecution.objects.create(
+            workflow=self.workflow,
+            pipeline_id=self.api.id,
+            execution_mode=WorkflowExecution.Mode.QUEUE,
+            execution_method=WorkflowExecution.Method.DIRECT,
+            execution_type=WorkflowExecution.Type.COMPLETE,
+            status="STOPPED",
+        )
+        stopped = ExecutionResponse(
+            workflow_id="w",
+            execution_id="e",
+            execution_status="STOPPED",
+            result=[item],
+        )
+        request = APIRequestFactory().get(
+            f"/deployment/api/{ORG_ID}/api/",
+            {"execution_id": str(execution.id)},
+            HTTP_AUTHORIZATION=f"Bearer {self.key.api_key}",
+        )
+        helper = "api_v2.api_deployment_views.DeploymentHelper"
+        with (
+            patch(f"{helper}.get_execution_status", return_value=stopped),
+            patch(f"{helper}._enrich_result_with_workflow_metadata"),
+        ):
+            return DeploymentExecution.as_view()(
+                request, org_name=ORG_ID, api_name="api"
+            )
+
+    def test_get_filters_a_stopped_runs_results_like_post(self) -> None:
+        response = self._get_stopped_result()
+
+        self.assertEqual(response.status_code, 422)
+        item = response.data["message"][0]
+        self.assertNotIn("highlight_data", item["metadata"])
+        self.assertNotIn("metrics", item["result"])
+        self.assertNotIn("metadata", item["result"])
+        self.assertEqual(item["result"]["output"], {"a": 1})

backend/api_v2/tests/test_deployment_helper.py

@@ -92,6 +92,24 @@ def test_staging_failure_cleanup_survives_db_marking_error(collaborators) -> Non
     collaborators["DestinationConnector"].delete_api_storage_dir.assert_called_once()
 
 
+def test_staging_failure_after_an_abort_answers_stopped(
+    collaborators: dict[str, MagicMock],
+) -> None:
+    """The answer matches the stored status, which the abort keeps STOPPED."""
+    collaborators["WorkflowExecutionServiceHelper"].update_execution_err.return_value = (
+        MagicMock(status="STOPPED")
+    )
+
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[],
+        timeout=-1,
+    )
+
+    assert response["execution_status"] == "STOPPED"
+
+
 @pytest.fixture
 def staging_rejects_everything():
     """Patch execute_workflow's collaborators; staging returns no dispatchable files."""
@@ -195,6 +213,27 @@ def test_all_files_rejected_acknowledges_and_notifies(
     )
 
 
+def test_an_aborted_run_is_not_reported_again(
+    staging_rejects_everything: dict[str, MagicMock],
+) -> None:
+    """The abort already notified subscribers of a run it stopped."""
+    mocks = staging_rejects_everything
+    stopped_row = mocks[
+        "WorkflowExecutionServiceHelper"
+    ].update_execution_completed.return_value
+    stopped_row.status = "STOPPED"
+
+    response = dh.DeploymentHelper.execute_workflow(
+        organization_name="org",
+        api=_api(),
+        file_objs=[MagicMock()],
+        timeout=-1,
+    )
+
+    mocks["PipelineUtils"].update_pipeline_status.assert_not_called()
+    assert response["execution_status"] == "STOPPED"
+
+
 def test_notification_survives_a_failed_acknowledgement(
     staging_rejects_everything,
 ) -> None:

backend/backend/settings/base.py

@@ -407,6 +407,10 @@ def validate_agent_kv_path_prefix(raw, path_prefix, deployment_prefix):
 INSTANT_WF_POLLING_TIMEOUT = int(
     os.environ.get("INSTANT_WF_POLLING_TIMEOUT", "300")
 )  # 5 minutes
+# A PENDING/EXECUTING run with no activity for this long is listed as stuck.
+STUCK_EXECUTION_THRESHOLD_SECONDS = int(
+    os.environ.get("STUCK_EXECUTION_THRESHOLD_SECONDS", "9000")
+)
 
 # ETL Pipeline minimum schedule interval (in seconds)
 # Default: 1800 seconds (30 minutes)

backend/mcp_server/registry.py

@@ -55,9 +55,9 @@ class MCPTool:
         consumes_result: True when *reading* is itself destructive — the result
             store is one-shot, so a successful read acknowledges the execution
             and the payload cannot be fetched again. ``getExecutionStatus`` on
-            a COMPLETED execution is the case: it neither writes configuration
-            nor spends money, so ``writes``/``billable`` are both false, but
-            calling it twice loses data. Without this flag the derivation below
+            a finished execution is the case: it neither writes
+            configuration nor spends money, so ``writes``/``billable`` are both
+            false, but calling it twice loses data. Without this flag the derivation below
             would advertise ``readOnlyHint: true`` and tell an agent it is safe
             to poll freely — the opposite of the truth.
         preflight: Optional ``preflight(context, **arguments)`` run *before* the
@@ -266,14 +266,14 @@ def build_deployment_registry() -> MCPToolRegistry:
                 "Fetch the status and, once available, the result of an "
                 "extraction previously started by `extractDocument`. Pass the "
                 "`execution_id` that call returned.\n\n"
-                "Returns an `execution_status` of PENDING, EXECUTING, COMPLETED "
-                "or ERROR. Poll while it is PENDING or EXECUTING, leaving a few "
+                "Returns an `execution_status` of PENDING, EXECUTING, COMPLETED, "
+                "ERROR or STOPPED. Poll while it is PENDING or EXECUTING, leaving a few "
                 "seconds between calls."
             ),
             input_schema=get_execution_status_schema(),
             handler=get_execution_status,
-            # Reading a COMPLETED execution acknowledges it and the result
-            # store is one-shot, so the payload cannot be fetched again.
+            # Reading a finished execution acknowledges it and the
+            # result store is one-shot, so the payload cannot be fetched again.
             # Not `writes` (no configuration changes) and not `billable`
             # (no spend), but emphatically not read-only either.
             consumes_result=True,
@@ -714,14 +714,14 @@ def build_platform_registry() -> MCPToolRegistry:
                 "Poll for the result of an extraction started by "
                 "extractDocument. Free to call.\n\n"
                 "If extractDocument already returned execution_status "
-                "COMPLETED, the result is in that response and no polling is "
-                "needed. Otherwise poll this until COMPLETED or ERROR, "
+                "COMPLETED, ERROR or STOPPED, that response is final and no "
+                "polling is needed. Otherwise poll this until one of those, "
                 "pausing a few seconds between calls."
             ),
             input_schema=platform_execution_status_schema(),
             handler=get_platform_execution_status,
-            # Reading a COMPLETED execution acknowledges it and the result
-            # store is one-shot, so the payload cannot be fetched again.
+            # Reading a finished execution acknowledges it and the
+            # result store is one-shot, so the payload cannot be fetched again.
             # Not `writes` (no configuration changes) and not `billable`
             # (no spend), but emphatically not read-only either.
             consumes_result=True,

backend/mcp_server/tests/test_tool_execution.py

@@ -314,6 +314,20 @@ def test_completed_execution_is_enriched_and_returned(self) -> None:
         assert result["result"][0]["result"]["output"]["total"] == 42
         assert result["execution_id"] == EXECUTION_ID
 
+    def test_stopped_execution_is_enriched_too(self) -> None:
+        response = ExecutionResponse(
+            workflow_id="wf",
+            execution_id=EXECUTION_ID,
+            execution_status="STOPPED",
+            result=[{"file": "invoice.pdf", "result": {"output": {"total": 42}}}],
+        )
+
+        result, process = self._run(response, include_metrics=True)
+
+        process.assert_called_once()
+        assert process.call_args.kwargs["include_metrics"] is True
+        assert result["execution_status"] == "STOPPED"
+
     def test_pending_execution_is_not_enriched(self) -> None:
         response = ExecutionResponse(
             workflow_id="wf",

backend/mcp_server/tools/execution.py

@@ -14,6 +14,7 @@
 from tags.serializers import TagParamsSerializer
 from utils.enums import CeleryTaskState
 from workflow_manager.workflow_v2.dto import ExecutionResponse
+from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.exceptions import ExecutionDoesNotExistError
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 
@@ -375,7 +376,11 @@ def get_execution_status(
             ),
         }
 
-    if response.execution_status == CeleryTaskState.COMPLETED.value:
+    # A stopped run returns its results so far, filtered the same way.
+    if response.execution_status in (
+        CeleryTaskState.COMPLETED.value,
+        ExecutionStatus.STOPPED.value,
+    ):
         deployment_execution_dto = DeploymentExecutionDTO(
             api=context.api, api_key=context.api_key
         )

backend/mcp_server/tools/info.py

@@ -47,10 +47,11 @@ def read_me_first(context: MCPContext) -> dict[str, Any]:
             "1. Call getApiInfo to confirm the deployment is active and see "
             "what it is meant to process.",
             "2. Call extractDocument with the URL(s) of the document(s).",
-            "3. If the response has execution_status COMPLETED, the result is "
-            "already in the response — you are done.",
+            "3. If the response has execution_status COMPLETED, ERROR or "
+            "STOPPED, it is final and any result is already in it — you are "
+            "done.",
             "4. Otherwise poll getExecutionStatus with the returned "
-            "execution_id until the status is COMPLETED or ERROR, pausing a "
+            "execution_id until the status is COMPLETED, ERROR or STOPPED, pausing a "
             "few seconds between polls.",
         ],
         "notes": [

backend/mcp_server/tools/platform.py

@@ -178,8 +178,8 @@ def platform_read_me_first(context: PlatformMCPContext) -> dict[str, Any]:
         "to_run_an_extraction": (
             "Call listApiDeployments for an api_name, then extractDocument "
             "with that name and your document URLs. If the response is not "
-            "already COMPLETED, poll getExecutionStatus with the returned "
-            "execution_id.\n\n"
+            "already COMPLETED, ERROR or STOPPED, poll getExecutionStatus with "
+            "the returned execution_id.\n\n"
             "The same extraction is also available on a deployment-scoped "
             "server at /deployment/api/<org>/<api_name>/mcp, which takes that "
             "one deployment's own API key and can reach nothing else. Both "