diff --git a/openviking/storage/viking_fs.py b/openviking/storage/viking_fs.py index 983a19b78a..a9f28fe8b1 100644 --- a/openviking/storage/viking_fs.py +++ b/openviking/storage/viking_fs.py @@ -1086,6 +1086,14 @@ async def _grep_vikingdb_then_fs( # use the maximum limit to avoid truncation. remote_return_limit = min(node_limit * 5, 100000) if node_limit else 100000 + # VikingDB BM25 can silently time out / return empty on large corpora + # (see #2850). Bound the query so a hung remote cannot stall grep; a + # timeout is treated as unreliability and falls back to fs, same as an + # raised exception. + vikingdb_timeout = float( + os.environ.get("OPENVIKING_GREP_VIKINGDB_TIMEOUT_SEC", "10") + ) + # Step 1: vikingdb recall candidate files try: logger.debug( @@ -1096,11 +1104,31 @@ async def _grep_vikingdb_then_fs( filter_expr, ["uri"], ) - result = await vector_store.search_by_keywords( - query=query, - limit=remote_return_limit, - filter=filter_expr, - output_fields=["uri"], + result = await asyncio.wait_for( + vector_store.search_by_keywords( + query=query, + limit=remote_return_limit, + filter=filter_expr, + output_fields=["uri"], + ctx=ctx, + ), + timeout=vikingdb_timeout, + ) + except asyncio.TimeoutError: + logger.warning( + "grep vikingdb query timed out (%.1fs) for pattern %r at %s, " + "falling back to fs", + vikingdb_timeout, + pattern, + uri, + ) + return await self._grep_fs( + uri=uri, + pattern=pattern, + exclude_uri=exclude_uri, + case_insensitive=case_insensitive, + node_limit=node_limit, + level_limit=level_limit, ctx=ctx, ) except Exception as e: @@ -1123,8 +1151,25 @@ async def _grep_vikingdb_then_fs( if u != excluded_prefix and not u.startswith(excluded_prefix + "/") ] if not candidate_uris: - # BM25 returned no candidates — the index confirms no matching content - return {"matches": [], "count": 0, "match_count": 0, "files_scanned": 0} + # BM25 returned no candidates. On large/stale corpora this can be a + # false negative (silent timeout, index lag) rather than a true + # absence of matches (#2850). Retry with fs grep so a VikingDB gap + # does not masquerade as "no matching content". + logger.warning( + "grep vikingdb returned 0 candidates for pattern %r at %s, " + "retrying with fs grep", + pattern, + uri, + ) + return await self._grep_fs( + uri=uri, + pattern=pattern, + exclude_uri=exclude_uri, + case_insensitive=case_insensitive, + node_limit=node_limit, + level_limit=level_limit, + ctx=ctx, + ) # Step 2: local fs precise matching on candidate files return await self._grep_in_files( diff --git a/tests/storage/test_viking_fs_grep.py b/tests/storage/test_viking_fs_grep.py index 47aa71ad56..1a285266b7 100644 --- a/tests/storage/test_viking_fs_grep.py +++ b/tests/storage/test_viking_fs_grep.py @@ -1,6 +1,7 @@ # Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. # SPDX-License-Identifier: AGPL-3.0 +import asyncio import time import pytest @@ -147,6 +148,90 @@ async def fake_grep_fs(**kwargs): ] +class _SlowVectorStore: + """Vector store whose search_by_keywords never returns within the timeout.""" + + async def search_by_keywords(self, **kwargs): + await asyncio.sleep(30) + return [] + + +@pytest.mark.asyncio +async def test_grep_vikingdb_empty_recall_falls_back_to_fs(monkeypatch): + """VikingDB returning zero candidates must retry via fs grep (#2850).""" + fs = VikingFS(agfs=_DummyAgfs()) + # _DummyVectorStore() defaults to empty results -> empty recall. + vector_store = _DummyVectorStore() + monkeypatch.setattr(fs, "_get_vector_store", lambda: vector_store) + monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None) + + calls = [] + + async def fake_grep_fs(**kwargs): + calls.append(kwargs) + return { + "matches": [{"uri": "viking://resources/a.md"}], + "count": 1, + "match_count": 1, + "files_scanned": 1, + } + + monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs) + + result = await fs._grep_vikingdb_then_fs( + uri="viking://resources", + pattern="needle", + exclude_uri=None, + case_insensitive=False, + node_limit=10, + level_limit=3, + ctx=None, + ) + + # fs fallback was invoked and its result surfaced (not the empty VikingDB hit). + assert len(calls) == 1 + assert calls[0]["pattern"] == "needle" + assert result["count"] == 1 + assert result["matches"][0]["uri"] == "viking://resources/a.md" + + +@pytest.mark.asyncio +async def test_grep_vikingdb_timeout_falls_back_to_fs(monkeypatch): + """A VikingDB query that exceeds the timeout must fall back to fs (#2850).""" + fs = VikingFS(agfs=_DummyAgfs()) + monkeypatch.setattr(fs, "_get_vector_store", lambda: _SlowVectorStore()) + monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None) + # Keep the test fast: 0.1s timeout instead of the 10s default. + monkeypatch.setenv("OPENVIKING_GREP_VIKINGDB_TIMEOUT_SEC", "0.1") + + calls = [] + + async def fake_grep_fs(**kwargs): + calls.append(kwargs) + return { + "matches": [{"uri": "viking://resources/b.md"}], + "count": 1, + "match_count": 1, + "files_scanned": 1, + } + + monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs) + + result = await fs._grep_vikingdb_then_fs( + uri="viking://resources", + pattern="needle", + exclude_uri=None, + case_insensitive=False, + node_limit=10, + level_limit=3, + ctx=None, + ) + + assert len(calls) == 1 + assert result["count"] == 1 + assert result["matches"][0]["uri"] == "viking://resources/b.md" + + @pytest.mark.asyncio async def test_grep_vikingdb_pushes_exclude_uri_to_filter(monkeypatch): fs = VikingFS(agfs=_DummyAgfs()) @@ -154,6 +239,13 @@ async def test_grep_vikingdb_pushes_exclude_uri_to_filter(monkeypatch): monkeypatch.setattr(fs, "_get_vector_store", lambda: vector_store) monkeypatch.setattr(fs, "_ensure_access", lambda uri, ctx=None: None) + # Empty VikingDB recall now falls back to fs grep (#2850); stub fs so the + # exclude_uri filter assertion stays focused on the remote query path. + async def fake_grep_fs(**kwargs): + return {"matches": [], "count": 0, "match_count": 0, "files_scanned": 0} + + monkeypatch.setattr(fs, "_grep_fs", fake_grep_fs) + result = await fs._grep_vikingdb_then_fs( uri="viking://resources", pattern="needle",