Improve query parsing, chunking, uploads & reindex

Multiple robustness and correctness fixes across ingestion, search, upload and ranking:

- scripts/hybrid/filters.py: tighten DSL tokenizer to only match compact key:value (no spaces) to avoid capturing prose like "file: upload" or "repo: billing".
- scripts/hybrid/ranking.py: fix span sorting to correctly order by score (negate score uniformly) to avoid wasting token budget on low-ranked penalized spans.
- scripts/ingest/chunking.py: emit gap chunks for top-level code between symbols so imports/docstrings/top-level code are not dropped.
- scripts/ingest/pipeline.py: move dedupe delete to after embeddings/upsert and add per-file lock wrapper for smart reindexing; add retries/fallback in pseudo_backfill_tick.
- scripts/mcp_impl/context_search.py: scope fallback search to the same repo to avoid returning other repos' code.
- scripts/mcp_impl/search.py: run in-process reranker with a timed ThreadPoolExecutor to honor rerank timeout, and ensure rerank_return_m is treated as a quality cap while honoring caller limits.
- scripts/upload_delta_bundle.py: add atomic file writes, better error isolation for malformed/unsafe ops (skip and count failed ops instead of aborting bundle), collision warnings and safer apply logic.
- scripts/upload_service.py: persist per-workspace upload sequence to disk, warn on workspace key collisions, handle idempotent replay, apply bundles inline (ack after apply) and return processing metrics; improved error responses.
- scripts/workspace_state.py: make set_cached_file_hash atomic across threads/processes to avoid cache races.
- tests: add query DSL tests, chunking regression test and update upload_service tests to assert failed ops are isolated rather than raising.

These changes reduce silent failures, prevent data loss from partial operations, and make indexing/search behaviour more predictable.

Author: m1rl0k

scripts/hybrid/filters.py

@@ -150,8 +150,13 @@ def parse_query_dsl(queries: List[str]) -> Tuple[List[str], Dict[str, str]]:
     """
     clean: List[str] = []
     extracted: Dict[str, str] = {}
+    # Compact form only (`file:foo.py` — no whitespace around the colon).
+    # Allowing `\s*` here turned natural-language queries like
+    # "explain the file: upload flow" into hard filters (under="upload"),
+    # and a prose "repo: billing" silently redirected the search to another
+    # repo.
     token_re = re.compile(
-        r"\b(?:(lang|language|file|path|under|kind|symbol|ext|not|case|repo))\s*:\s*([^\s]+)",
+        r"\b(?:(lang|language|file|path|under|kind|symbol|ext|not|case|repo)):([^\s]+)",
         re.IGNORECASE,
     )
     for q in queries:

scripts/hybrid/ranking.py

@@ -872,10 +872,11 @@ def _flat_key(c):
         path = str(c.get("p") or "")
         start = int(c.get("start") or 0)
         score = float(m.get("raw_score") or m.get("score") or m.get("s") or 0.0)
-        if score < 0:
-            return (score, path, start)
-        else:
-            return (-score, path, start)
+        # Highest score first for ALL scores. Negating only non-negatives
+        # reversed the order among penalized (negative) spans and sorted
+        # every negative ahead of weak positives — the token budget got
+        # spent on the worst spans first.
+        return (-score, path, start)
 
     flattened.sort(key=_flat_key)
 

scripts/ingest/chunking.py

@@ -134,6 +134,38 @@ def chunk_semantic(
                     break
 
         if best_symbol:
+            # Emit any gap between our current position and this symbol's start
+            # (e.g. leading imports/module docstring, or top-level code between
+            # two symbols) as its own chunk first, so it isn't silently dropped
+            # from the index. Bounded by max_lines: best_symbol.start <= chunk_end.
+            if best_symbol.start > chunk_start:
+                gap_start = chunk_start
+                gap_end = best_symbol.start - 1
+                # Mirror the fallback branch below: don't let a chunk cross out
+                # of an enclosing symbol we're still inside of. Split off any
+                # remainder past it (unrelated top-level code) as its own,
+                # unlabeled chunk instead.
+                if enclosing_symbol is not None and gap_start <= enclosing_symbol.end:
+                    enclosed_end = min(gap_end, enclosing_symbol.end)
+                    chunks.append(
+                        {
+                            "text": "\n".join(lines[gap_start - 1 : enclosed_end]),
+                            "start": gap_start,
+                            "end": enclosed_end,
+                            "symbol": enclosing_symbol.name,
+                            "kind": enclosing_symbol.kind,
+                        }
+                    )
+                    gap_start = enclosed_end + 1
+                if gap_start <= gap_end:
+                    chunks.append(
+                        {
+                            "text": "\n".join(lines[gap_start - 1 : gap_end]),
+                            "start": gap_start,
+                            "end": gap_end,
+                        }
+                    )
+
             # Chunk this complete symbol
             chunk_text = "\n".join(lines[best_symbol.start - 1 : best_symbol.end])
             chunks.append(

scripts/ingest/pipeline.py

@@ -413,9 +413,9 @@ def _index_single_file_inner(
                 print(f"Skipping unchanged file: {file_path}")
                 return False
 
-    if dedupe:
-        delete_points_by_path(client, collection, str(file_path))
-
+    # NOTE: with dedupe, old points are deleted just before the upsert below,
+    # AFTER embedding succeeded — deleting up front left the file entirely
+    # unsearchable whenever the embedder failed mid-flight.
     symbols = _extract_symbols(language, text)
     imports, calls = _get_imports_calls(language, text)
     last_mod, churn_count, author_count = _git_metadata(file_path)
@@ -652,6 +652,8 @@ def make_point(pid, dense_vec, lex_vec, payload, lex_text: str = "", code_text:
             make_point(i, v, lx, m, lt, ct)
             for i, v, lx, m, lt, ct in zip(batch_ids, vectors, batch_lex, batch_meta, batch_lex_text, batch_code)
         ]
+        if dedupe:
+            delete_points_by_path(client, collection, str(file_path))
         upsert_points(client, collection, points)
         try:
             ws = os.environ.get("WATCH_ROOT") or os.environ.get("WORKSPACE_PATH") or "/work"
@@ -661,6 +663,10 @@ def make_point(pid, dense_vec, lex_vec, payload, lex_text: str = "", code_text:
         except Exception:
             pass
         return True
+    if dedupe:
+        # The file yields no indexable chunks anymore (emptied or newly
+        # excluded): still drop its stale points.
+        delete_points_by_path(client, collection, str(file_path))
     return False
 
 
@@ -883,6 +889,57 @@ def process_file_with_smart_reindexing(
     *,
     allowed_vectors: set[str] | None = None,
     allowed_sparse: set[str] | None = None,
+) -> str:
+    """Cross-process file lock around smart reindexing.
+
+    Same guard as `index_single_file`: the smart scroll→delete→upsert
+    sequence must not interleave with a concurrent full index of the same
+    file — whichever writer finishes last would silently discard the
+    other's points.
+    """
+    _file_lock_ctx = None
+    if file_indexing_lock is not None:
+        try:
+            _file_lock_ctx = file_indexing_lock(str(file_path))
+            _file_lock_ctx.__enter__()
+        except FileExistsError:
+            print(f"[SMART_REINDEX] {file_path}: locked by another indexer, skipping")
+            return "skipped"
+        except Exception:
+            pass
+    try:
+        return _process_file_with_smart_reindexing_inner(
+            file_path,
+            text,
+            language,
+            client,
+            current_collection,
+            per_file_repo,
+            model,
+            vector_name,
+            allowed_vectors=allowed_vectors,
+            allowed_sparse=allowed_sparse,
+        )
+    finally:
+        if _file_lock_ctx is not None:
+            try:
+                _file_lock_ctx.__exit__(None, None, None)
+            except Exception:
+                pass
+
+
+def _process_file_with_smart_reindexing_inner(
+    file_path,
+    text: str,
+    language: str,
+    client: QdrantClient,
+    current_collection: str,
+    per_file_repo,
+    model,
+    vector_name: str | None,
+    *,
+    allowed_vectors: set[str] | None = None,
+    allowed_sparse: set[str] | None = None,
 ) -> str:
     """Smart, chunk-level reindexing for a single file.
 
@@ -1506,7 +1563,15 @@ def _maybe_ensure_collection() -> bool:
                 offset=next_offset,
             )
         except Exception:
-            if _maybe_ensure_collection():
+            # A single transient error must not abort the whole tick — the
+            # unprocessed tail would wait for a future tick. Retry briefly,
+            # then fall back to the ensure-collection path.
+            recovered = False
+            for _retry in (1, 2):
+                try:
+                    time.sleep(0.4 * _retry)
+                except Exception:
+                    pass
                 try:
                     points, next_offset = client.scroll(
                         collection_name=collection,
@@ -1516,10 +1581,25 @@ def _maybe_ensure_collection() -> bool:
                         with_vectors=True,
                         offset=next_offset,
                     )
+                    recovered = True
+                    break
                 except Exception:
+                    continue
+            if not recovered:
+                if _maybe_ensure_collection():
+                    try:
+                        points, next_offset = client.scroll(
+                            collection_name=collection,
+                            scroll_filter=flt,
+                            limit=batch_limit,
+                            with_payload=True,
+                            with_vectors=True,
+                            offset=next_offset,
+                        )
+                    except Exception:
+                        break
+                else:
                     break
-            else:
-                break
 
         if not points:
             break

scripts/mcp_impl/context_search.py

@@ -748,11 +748,25 @@ def _maybe_dict(val: Any) -> Dict[str, Any]:
         try:
             from scripts.hybrid_search import run_hybrid_search  # type: ignore
 
+            # Scope the fallback exactly like the primary search. The primary
+            # path is repo-scoped via _detect_current_repo(); running this
+            # fallback unscoped silently returned OTHER repos' code whenever
+            # the scoped search had zero hits (run_hybrid_search's own
+            # auto-detect is env-only and misses the /work git checkout).
+            repo_scope = None
+            try:
+                from scripts.mcp_impl.admin_tools import _detect_current_repo
+
+                repo_scope = _detect_current_repo()
+            except Exception:
+                repo_scope = None
+
             model_name = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-base-en-v1.5")
             model = get_embedding_model_fn(model_name) if get_embedding_model_fn else None
             items2 = await asyncio.to_thread(
                 lambda: run_hybrid_search(
                     queries=queries,
+                    repo=repo_scope,
                     limit=int(code_limit),
                     per_path=int(per_path_val),
                     language=language or None,

scripts/mcp_impl/search.py

@@ -39,7 +39,15 @@
     _tokens_from_queries,
     safe_int,
 )
-from scripts.mcp_impl.workspace import _default_collection, _work_script
+from scripts.mcp_impl.workspace import (
+    SESSION_DEFAULTS,
+    SESSION_DEFAULTS_BY_SESSION,
+    _SESSION_CTX_LOCK,
+    _SESSION_LOCK,
+    _default_collection,
+    _work_script,
+)
+from scripts.mcp_impl.code_signals import _detect_code_signals
 from scripts.mcp_impl.admin_tools import _detect_current_repo, _run_async
 from scripts.mcp_toon import _should_use_toon, _format_results_as_toon
 from scripts.mcp_auth import require_collection_access as _require_collection_access
@@ -980,7 +988,20 @@ def _doc_for(obj: dict) -> str:
                     _before_paths = [(o.get("path", "?").split("/")[-1], o.get("score", 0)) for o in cand_objs[:10]]
 
                     pairs = [(rq, d) for d in docs]
-                    scores = _rr_local(pairs)
+                    # Honor rerank_timeout_ms on the in-process path too: the
+                    # bare call blocked the server with no hybrid fallback
+                    # (the subprocess path already enforces the timeout).
+                    # shutdown(wait=False) so a timeout unblocks the request
+                    # instead of joining the stuck worker.
+                    import concurrent.futures as _futures
+
+                    _rr_pool = _futures.ThreadPoolExecutor(max_workers=1)
+                    try:
+                        scores = _rr_pool.submit(_rr_local, pairs).result(
+                            timeout=max(0.05, float(rerank_timeout_ms) / 1000.0)
+                        )
+                    finally:
+                        _rr_pool.shutdown(wait=False, cancel_futures=True)
                     # Blend rerank with fusion score to preserve pre-rerank boosts
                     # (symbol_exact, impl_boost, path boosts are otherwise lost)
                     _rerank_blend = float(os.environ.get("RERANK_BLEND_WEIGHT", "0.6") or 0.6)
@@ -1013,8 +1034,17 @@ def _doc_for(obj: dict) -> str:
                         blended_score += post_boost
                         blended.append((blended_score, rr_score, obj, post_boost))
                     ranked = sorted(blended, key=lambda x: x[0], reverse=True)
+                    # rerank_return_m is a quality cap, not a result cap:
+                    # honor a larger caller limit from the already-retrieved
+                    # pool instead of silently returning fewer results.
+                    _return_cap = int(rerank_return_m)
+                    try:
+                        if limit and int(limit) > _return_cap:
+                            _return_cap = int(limit)
+                    except Exception:
+                        pass
                     tmp = []
-                    for blended_s, rr_s, obj, post_b in ranked[: int(rerank_return_m)]:
+                    for blended_s, rr_s, obj, post_b in ranked[:_return_cap]:
                         why_parts = obj.get("why", []) + [f"rerank_onnx:{float(rr_s):.3f}", f"blend:{float(blended_s):.3f}"]
                         if post_b > 0:
                             why_parts.append(f"post_sym:{float(post_b):.3f}")