diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index da6e8a8ba4..2b4e67cb1a 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/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: diff --git a/src/backend/tests/unit/api/v2/test_workflow.py b/src/backend/tests/unit/api/v2/test_workflow.py index 519b7d6e85..3f365865c5 100644 --- a/src/backend/tests/unit/api/v2/test_workflow.py +++ b/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, diff --git a/src/backend/tests/unit/api/v2/test_workflow_agui.py b/src/backend/tests/unit/api/v2/test_workflow_agui.py index d58aab0b5a..a8fe06e5b1 100644 --- a/src/backend/tests/unit/api/v2/test_workflow_agui.py +++ b/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), + ]