diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index f34b4e5ffc..8824e272d3 100644 --- a/backend/app/gateway/services.py +++ b/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, diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 7e4837dfd4..fd696b3d04 100644 --- a/backend/tests/test_gateway_services.py +++ b/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