fix(analyzer): address review comments on second_pass parallelization (#688)

- Unify the resolve path: drop the `n_workers == 1` special case and
  always dispatch through ThreadPoolExecutor (max_workers=1 just runs one
  file at a time). This makes the per-file failure policy identical for
  every worker count (galshubeli, Copilot).
- Track files whose resolution raises and exclude them from phase B edge
  writes, so a worker failure no longer leaves a partial graph when
  workers > 1 (coderabbit). One bad file degrades to a logged skip instead
  of aborting or partially writing.
- Preserve the traceback on resolution failures via exc_info=True (Copilot).
- De-duplicate the resolvable file list while preserving order so a path
  appearing twice can't be resolved concurrently by two workers racing on
  the same entity.resolved_symbols sets (Copilot).
- Document that the default (CODE_GRAPH_INDEX_WORKERS=4) runs resolution in
  parallel by default and that workers=1 is a single resolver thread, not a
  main-thread serial path.
- Add a determinism regression test asserting the full relationship
  multiset is identical for workers=1 vs workers=4 (Copilot).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: DvirDukhan

api/analyzers/source_analyzer.py

@@ -148,10 +148,19 @@ def second_pass(self, graph: Graph, files: list[Path], path: Path) -> None:
              FalkorDB MERGE locks and produces a deterministic edge order
              matching the pre-parallel implementation.
 
-        Pool size is controlled by `CODE_GRAPH_INDEX_WORKERS` (default 4).
-        Set it to 1 to fall back to the historical fully-serial behavior
-        (useful for debugging or for hosts where multilspy/jedi misbehaves
-        under concurrency).
+        Pool size is controlled by `CODE_GRAPH_INDEX_WORKERS` (default 4),
+        so resolution runs in parallel by default. This is an intentional
+        behaviour change, but the edge output is identical to the
+        single-worker path (verified on a 204-file repo): phase B always
+        writes in the original file order regardless of how phase A
+        interleaves. Set the var to 1 to run a single resolver thread (useful
+        when multilspy/jedi misbehaves under concurrency); note this still
+        dispatches through the pool rather than the main thread.
+
+        Files whose resolution raises are logged with their traceback and
+        excluded from phase B, so one bad file degrades to a logged skip
+        instead of a partial or aborted graph -- and that behaviour no longer
+        depends on the worker count.
         """
         import os
         from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -208,9 +217,16 @@ def second_pass(self, graph: Graph, files: list[Path], path: Path) -> None:
 
             # Drop files we don't actually have an entry for and skip files
             # whose language has no real LSP (NullLanguageServer provides
-            # no symbol info, so resolution would be a no-op).
+            # no symbol info, so resolution would be a no-op). De-duplicate
+            # while preserving order so a path that appears twice in `files`
+            # isn't resolved concurrently by two workers racing on the same
+            # entity.resolved_symbols sets.
             resolvable: list[Path] = []
+            seen: set[Path] = set()
             for file_path in files:
+                if file_path in seen:
+                    continue
+                seen.add(file_path)
                 if file_path not in self.files:
                     # first_pass skipped this file (e.g. parse error, empty,
                     # untracked, or ignored after entering the candidate list).
@@ -245,33 +261,38 @@ def _resolve_file(file_path: Path) -> Path:
                     )
                 return file_path
 
+            # Phase A: resolve symbols. A single code path for every worker
+            # count keeps the failure policy identical regardless of
+            # CODE_GRAPH_INDEX_WORKERS -- a ThreadPoolExecutor with
+            # max_workers=1 simply processes one file at a time.
+            failed: set[Path] = set()
             done = 0
             log_every = max(1, total // 50) if total else 1
-            if n_workers == 1:
-                for fp in resolvable:
-                    _resolve_file(fp)
+            with ThreadPoolExecutor(max_workers=n_workers, thread_name_prefix="sa-resolve") as ex:
+                futures = {ex.submit(_resolve_file, fp): fp for fp in resolvable}
+                for fut in as_completed(futures):
+                    fp = futures[fut]
+                    try:
+                        fut.result()
+                    except Exception:
+                        # Exclude this file from phase B so we never persist a
+                        # partially resolved file; keep going so one bad file
+                        # doesn't abort the whole index.
+                        failed.add(fp)
+                        logging.warning(
+                            "second_pass: resolution failed for %s; excluding from edge writes",
+                            fp, exc_info=True,
+                        )
                     done += 1
                     if done % log_every == 0 or done == total:
                         logging.info("second_pass: resolved %d/%d files", done, total)
-            else:
-                with ThreadPoolExecutor(max_workers=n_workers, thread_name_prefix="sa-resolve") as ex:
-                    futures = {ex.submit(_resolve_file, fp): fp for fp in resolvable}
-                    for fut in as_completed(futures):
-                        fp = futures[fut]
-                        try:
-                            fut.result()
-                        except Exception as exc:
-                            logging.warning(
-                                "second_pass: resolution failed for %s: %s",
-                                fp, exc,
-                            )
-                        done += 1
-                        if done % log_every == 0 or done == total:
-                            logging.info("second_pass: resolved %d/%d files", done, total)
 
             # Phase B: serial edge writes, in the original file order so
-            # the graph is bit-identical to the single-threaded path.
+            # the graph is bit-identical to the single-threaded path. Files
+            # whose resolution failed are skipped (see phase A).
             for file_path in resolvable:
+                if file_path in failed:
+                    continue
                 file = self.files[file_path]
                 for _, entity in file.entities.items():
                     for key, resolved_set in entity.resolved_symbols.items():

tests/test_py_analyzer.py

@@ -1,10 +1,22 @@
 import os
 import unittest
+from collections import Counter
 from pathlib import Path
 
 from api import SourceAnalyzer, Graph
 
 
+def _edge_snapshot(g: Graph) -> Counter:
+    """Return a multiset of all relationships in the graph keyed by
+    (rel_type, src_path, src_name, dst_path, dst_name) so two indexing runs
+    can be compared independent of node ids or write order."""
+    rows = g._query(
+        "MATCH (a)-[r]->(b) "
+        "RETURN type(r), a.path, a.name, b.path, b.name"
+    ).result_set
+    return Counter(tuple(row) for row in rows)
+
+
 class Test_PY_Analyzer(unittest.TestCase):
     def test_analyzer(self):
         path = Path(__file__).parent
@@ -73,3 +85,44 @@ def test_analyzer(self):
         self.assertIn('__init__', callers)
         self.assertIn('log', callers)
 
+    def test_index_workers_edges_are_deterministic(self):
+        """second_pass must produce identical edges regardless of
+        CODE_GRAPH_INDEX_WORKERS, since phase A only parallelises symbol
+        resolution while phase B writes edges in a fixed order.
+
+        Regression guard for the parallel-resolution path (#688): runs the
+        same fixture serially (workers=1) and in parallel (workers=4) and
+        asserts the full relationship multiset matches.
+        """
+        path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'source_files', 'py')
+
+        prev = os.environ.get("CODE_GRAPH_INDEX_WORKERS")
+        g1 = Graph("py_workers1")
+        g4 = Graph("py_workers4")
+        try:
+            os.environ["CODE_GRAPH_INDEX_WORKERS"] = "1"
+            SourceAnalyzer().analyze_local_folder(path, g1)
+            serial = _edge_snapshot(g1)
+
+            os.environ["CODE_GRAPH_INDEX_WORKERS"] = "4"
+            SourceAnalyzer().analyze_local_folder(path, g4)
+            parallel = _edge_snapshot(g4)
+
+            self.assertEqual(
+                serial, parallel,
+                "edge multiset differs between workers=1 and workers=4",
+            )
+            # Sanity: the fixture has resolvable CALLS, so guard against a
+            # vacuous all-empty comparison when an LSP is available.
+            self.assertGreater(
+                sum(serial.values()), 0,
+                "expected at least one resolved edge in the fixture",
+            )
+        finally:
+            if prev is None:
+                os.environ.pop("CODE_GRAPH_INDEX_WORKERS", None)
+            else:
+                os.environ["CODE_GRAPH_INDEX_WORKERS"] = prev
+            g1.delete()
+            g4.delete()
+