fix(gateway): admit cross-process runs when a producer lease blips (#7136)

A cross-process stream bridge (RedisStreamBridge) returns False from
start_producer_lease only after a transient backend error whose partial
lease registration was safely rolled back, leaving nothing tracked. The
admission guard treated that as fatal and raised, surfacing a 500 for an
otherwise healthy run whenever Redis had a momentary blip.

Degrade instead of rejecting: admit the run and log a warning. An
untracked producer falls back to the historical conservative recovery
path (wait for the real END or the retained-stream TTL), so the run is
no less safe than it was before producer leases existed. The lease is
never transferred or finished because none was taken; the worker
re-attempts its own lease independently.

Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/gateway/services.py

@@ -2609,8 +2609,20 @@ async def start_run(
                 )
                 if start_producer_lease is not None:
                     producer_lease_started = await start_producer_lease(admission_run_id)
-                if getattr(bridge, "supports_cross_process", False) is True and not producer_lease_started:
-                    raise RuntimeError("Cross-process stream producer tracking is unavailable")
+                    if not producer_lease_started and getattr(bridge, "supports_cross_process", False) is True:
+                        # Best-effort: a cross-process bridge returns False only
+                        # after a transient backend error whose partial lease
+                        # registration was safely rolled back (nothing is left
+                        # tracked). An untracked producer falls back to the
+                        # historical conservative recovery path -- wait for the
+                        # real END or the retained-stream TTL rather than
+                        # synthesizing one -- so this run is no less safe than it
+                        # was before producer leases existed. Rejecting admission
+                        # here would instead turn a transient blip into a 500.
+                        logger.warning(
+                            "Stream producer tracking unavailable for run %s; admitting it with untracked (conservative) stream recovery",
+                            sanitize_log_param(admission_run_id),
+                        )
                 try:
                     record = await run_mgr.create_or_reject(
                         thread_id,

backend/tests/test_gateway_services.py

@@ -2271,6 +2271,81 @@ async def test_start_run_releases_preallocated_producer_after_admission_conflict
     assert events[1] == ("abort", events[0][1])
 
 
+@pytest.mark.asyncio
+async def test_start_run_admits_cross_process_run_when_producer_tracking_unavailable(
+    _stub_app_config,
+):
+    """A transient lease failure must degrade, not reject the run with a 500.
+
+    ``RedisStreamBridge.start_producer_lease`` returns ``False`` only after a
+    transient Redis error whose partial registration was safely rolled back
+    (see its ``except (RedisError, TimeoutError)`` branch): nothing is left
+    tracked. An untracked producer falls back to the historical conservative
+    recovery path (wait for the real END or the retained-stream TTL), which is
+    no less safe than pre-lease behaviour. Admission must therefore proceed;
+    the lease is simply never transferred or finished because none was taken.
+    """
+    from unittest.mock import patch
+
+    from app.gateway.services import start_run
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    run_store = MemoryRunStore()
+    run_manager = RunManager(store=run_store)
+    events = []
+    agent_started = asyncio.Event()
+    release = asyncio.Event()
+
+    class UntrackedCrossProcessBridge:
+        supports_cross_process = True
+
+        async def start_producer_lease(self, run_id):
+            events.append(("start", run_id))
+            return False
+
+        def transfer_producer_lease(self, run_id, owner_task):
+            events.append(("transfer", run_id, owner_task))
+
+        async def finish_producer_lease(self, run_id):
+            events.append(("finish", run_id))
+
+        async def abort_producer_lease(self, run_id):
+            events.append(("abort", run_id))
+
+    bridge = UntrackedCrossProcessBridge()
+
+    async def fake_run_agent(_bridge, _manager, record, **_kwargs):
+        agent_started.set()
+        await release.wait()
+
+    request = _make_start_run_request(
+        run_manager,
+        stream_bridge=bridge,
+    )
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+    ):
+        record = await start_run(
+            _run_create_request(),
+            "thread-stream-producer-untracked",
+            request,
+        )
+        try:
+            await asyncio.wait_for(agent_started.wait(), timeout=1)
+            # Admitted (no 500); no lease was transferred or finished because
+            # none could be established.
+            assert events == [("start", record.run_id)]
+            assert (await run_store.get(record.run_id))["status"] == "pending"
+        finally:
+            release.set()
+            await asyncio.wait_for(record.task, timeout=1)
+
+    # No lease was taken, so none is transferred, finished, or aborted.
+    assert events == [("start", record.run_id)]
+
+
 @pytest.mark.asyncio
 async def test_pending_cancel_bypasses_thread_metadata_and_logs_failure(_stub_app_config, caplog):
     from unittest.mock import AsyncMock, patch