fix(api/v2): report unconfirmed workflow stops LE-1389

src/backend/base/langflow/api/v2/workflow.py

@@ -105,8 +105,12 @@ async def generate_flow_events(*args, **kwargs) -> None:
 async def _cancel_workflow_queue_job(job_id: str) -> bool:
     """Lazily call the shared build cancellation path to avoid import cycles."""
     from langflow.api.build import cancel_flow_build
+    from langflow.services.job_queue.service import JobQueueNotFoundError
 
-    return await cancel_flow_build(job_id=job_id, queue_service=get_queue_service())
+    try:
+        return await cancel_flow_build(job_id=job_id, queue_service=get_queue_service())
+    except JobQueueNotFoundError:
+        return False
 
 
 def _unknown_protocol_http_exception(exc: UnknownStreamProtocolError) -> HTTPException:

src/backend/tests/unit/api/v2/test_workflow.py

@@ -300,6 +300,45 @@ class TestWorkflowStop:
             mock_cancel_workflow_queue_job.assert_awaited_once_with(job_id)
             mock_job_service.update_job_status.assert_called_once_with(UUID(job_id), JobStatus.CANCELLED)
 
+    async def test_stop_workflow_returns_503_when_queue_cancel_cannot_be_confirmed(
+        self,
+        client: AsyncClient,
+        created_api_key,
+    ):
+        """Do not mark persisted jobs cancelled when no queue owner can be reached."""
+        from langflow.services.job_queue.service import JobQueueNotFoundError
+
+        job_id = str(uuid4())
+
+        mock_job = MagicMock()
+        mock_job.job_id = job_id
+        mock_job.status = JobStatus.IN_PROGRESS
+        mock_job.type = JobType.WORKFLOW
+        mock_job.user_id = None
+
+        class MissingQueueService:
+            def get_queue_data(self, seen_job_id):
+                raise JobQueueNotFoundError(seen_job_id)
+
+        with (
+            patch("langflow.api.v2.workflow.get_job_service") as mock_get_job_service,
+            patch("langflow.api.v2.workflow.get_queue_service", return_value=MissingQueueService()),
+        ):
+            mock_job_service = MagicMock()
+            mock_job_service.get_job_by_job_id = AsyncMock(return_value=mock_job)
+            mock_job_service.update_job_status = AsyncMock()
+            mock_get_job_service.return_value = mock_job_service
+
+            response = await client.post(
+                "api/v2/workflows/stop",
+                json={"job_id": job_id},
+                headers={"x-api-key": created_api_key.api_key},
+            )
+
+            assert response.status_code == 503
+            assert response.json()["detail"]["code"] == "WORKFLOW_CANCEL_UNAVAILABLE"
+            mock_job_service.update_job_status.assert_not_awaited()
+
     async def test_stop_workflow_not_found(
         self,
         client: AsyncClient,

src/backend/tests/unit/api/v2/test_workflow_agui.py

@@ -2025,6 +2025,7 @@ class TestStopWorkflowEndToEnd:
 
         job_id = uuid4()
         current_user_id = uuid4()
+        events: list[tuple[str, object, object | None]] = []
 
         class FakeJobService:
             def __init__(self) -> None:
@@ -2039,6 +2040,7 @@ class TestStopWorkflowEndToEnd:
                 )
 
             async def update_job_status(self, seen_job_id, status):
+                events.append(("update", seen_job_id, status))
                 self.updates.append((seen_job_id, status))
 
         class CrossWorkerQueueService:
@@ -2056,6 +2058,7 @@ class TestStopWorkflowEndToEnd:
                 self.local_cancels.append(seen_job_id)
 
             async def signal_cancel(self, seen_job_id):
+                events.append(("signal", seen_job_id, None))
                 self.signals.append(seen_job_id)
                 return 0
 
@@ -2073,3 +2076,7 @@ class TestStopWorkflowEndToEnd:
         assert queue_service.local_cancels == []
         assert queue_service.signals == [str(job_id)]
         assert job_service.updates == [(job_id, workflow_module.JobStatus.CANCELLED)]
+        assert events == [
+            ("signal", str(job_id), None),
+            ("update", job_id, workflow_module.JobStatus.CANCELLED),
+        ]