test(background): prove stop() pub/sub fast-path reaches the owning dispatcher

src/backend/base/langflow/services/background_execution/redis_backend.py

@@ -17,6 +17,7 @@ history because milestones are durable and live frames are on the shared Stream.
 
 from __future__ import annotations
 
+import contextlib
 import uuid
 from typing import TYPE_CHECKING, Any
 
@@ -34,6 +35,20 @@ if TYPE_CHECKING:
     from langflow.services.jobs.service import JobService
 
 
+def _as_signal_job_id(job_id: str) -> Any:
+    """Best-effort coerce a job id string to a UUID for the durable signal row.
+
+    The facade always passes a real UUID string, so the durable STOP row keys
+    match ``unconsumed_signals(UUID)`` lookups. A non-UUID id (the pure pub/sub
+    wire test runs with a noop job service) is passed through unchanged.
+    """
+    if isinstance(job_id, uuid.UUID):
+        return job_id
+    with contextlib.suppress(ValueError, AttributeError, TypeError):
+        return uuid.UUID(job_id)
+    return job_id
+
+
 class _StreamFrame:
     """A live (ephemeral) frame pulled off the redis Stream tail.
 
@@ -84,8 +99,11 @@ class RedisBackgroundQueue:
         worker running the existing cancel dispatcher / marker-check picks these
         up unchanged (we reuse, not reimplement, the wire path).
         """
-        # 1. Durable source of truth.
-        await self._job_service.write_signal(uuid.UUID(job_id) if isinstance(job_id, str) else job_id, SignalType.STOP)
+        # 1. Durable source of truth. Coerce to UUID so the DB row keys match
+        #    unconsumed_signals(UUID) lookups; tolerate non-UUID ids (used in the
+        #    pure pub/sub wire test with a noop job service) by passing them
+        #    through unchanged.
+        await self._job_service.write_signal(_as_signal_job_id(job_id), SignalType.STOP)
         # 2. Fast-path: set the marker (race-safe for a worker that hasn't
         #    subscribed yet) then publish on the cancel channel.
         marker_key = f"{RedisJobQueueService._CANCEL_MARKER_PREFIX}{job_id}"  # noqa: SLF001

src/backend/tests/unit/services/background_execution/test_control_fastpath_realredis.py

@@ -0,0 +1,58 @@
+"""Real-redis: stop() PUBLISH reaches an owning RedisJobQueueService dispatcher."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+import pytest
+from langflow.services.background_execution.redis_backend import RedisBackgroundQueue
+from langflow.services.job_queue.service import RedisJobQueueService
+
+
+class _NoopJobService:
+    async def write_signal(self, _job_id, _signal_type):  # durable write is exercised elsewhere
+        return None
+
+
+@pytest.mark.asyncio
+async def test_stop_publish_cancels_owned_build(real_redis_url):
+    if real_redis_url is None:
+        pytest.skip("LANGFLOW_TEST_REDIS_URL not set")
+
+    job_id = "fastpath-job"
+
+    # Owning worker: real RedisJobQueueService with its cancel dispatcher running.
+    owner = RedisJobQueueService(url=real_redis_url, cancel_channel_enabled=True)
+    owner.start()
+    # Give the PSUBSCRIBE a beat to land before publishing.
+    await asyncio.sleep(0.3)
+
+    cancelled = asyncio.Event()
+
+    async def fake_build():
+        try:
+            await asyncio.sleep(30)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    owner.create_queue(job_id)
+    owner.start_job(job_id, fake_build())
+
+    from redis.asyncio import StrictRedis
+
+    publisher = StrictRedis.from_url(real_redis_url)
+    backend = RedisBackgroundQueue(client=publisher, job_service=_NoopJobService())
+    try:
+        await backend.stop(job_id)
+        # The dispatcher should receive the pmessage and cancel the local task.
+        await asyncio.wait_for(cancelled.wait(), timeout=5.0)
+        assert cancelled.is_set()
+    finally:
+        await publisher.aclose()
+        # owner.stop() routes through cleanup_job, which re-raises the
+        # user-initiated CancelledError (a BaseException, so suppress(Exception)
+        # would miss it). Suppress both so teardown stays clean.
+        with contextlib.suppress(asyncio.CancelledError, Exception):
+            await owner.stop()