fix(conveyor): keep long index stages alive and dedupe aliases

Author: apstenku123

scripts/streaming_conveyor.py

@@ -2499,6 +2499,18 @@ def code_key(repo: str) -> str:
     return f"{repo}::code"
 
 
+def claim_code_project_identity(
+    claims: dict[str, str],
+    repo: str,
+    project_identity: str,
+) -> str | None:
+    """Claim one canonical code project and return an existing alias, if any."""
+    previous = claims.get(project_identity)
+    if previous is None:
+        claims[project_identity] = repo
+    return previous
+
+
 def no_git_key(repo: str) -> str:
     """Checkpoint key proving the .git-preserving stream saw no git metadata."""
     return f"{repo}::no_git"
@@ -4122,13 +4134,14 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
                    help="Parse workers passed to index_project for the CODE half "
                         "(default 2; avoid multiplying --repo-workers by the old "
                         "index_project default of 8 clang workers).")
-    p.add_argument("--code-index-timeout-s", type=int, default=0,
-                   help="Optional fail-loud timeout for each CODE index_project "
-                        "stage. 0 disables the timeout (default).")
-    p.add_argument("--code-index-stall-timeout-s", type=int, default=0,
-                   help="Optional fail-loud CODE index_project stall watchdog: "
-                        "kill when the index log has no size/mtime progress for "
-                        "this many seconds. 0 disables the watchdog (default).")
+    p.add_argument("--code-index-timeout-s", type=int,
+                   default=sr.DEFAULT_CODE_INDEX_TIMEOUT_S,
+                   help="Fail-loud timeout for each CODE index_project stage "
+                        f"(default {sr.DEFAULT_CODE_INDEX_TIMEOUT_S}s; 0 disables).")
+    p.add_argument("--code-index-stall-timeout-s", type=int,
+                   default=sr.DEFAULT_CODE_INDEX_STALL_TIMEOUT_S,
+                   help="Fail-loud log-heartbeat CODE index_project watchdog "
+                        f"(default {sr.DEFAULT_CODE_INDEX_STALL_TIMEOUT_S}s; 0 disables).")
     p.add_argument("--background-code-recompress", action="store_true",
                    help="Defer code parquet zstd-max recompress to a background "
                         "pool so repo processing can continue after valid parquet "
@@ -4220,7 +4233,11 @@ def main(argv: list[str]) -> int:
         raise SystemExit("--max-active-repos must be >= --repo-workers")
     parse_workers = max(1, int(args.parse_workers or 1))
     args.parse_workers = parse_workers
-    code_index_stall_timeout_s = int(args.code_index_stall_timeout_s or 0)
+    code_index_stall_timeout_s = int(
+        args.code_index_stall_timeout_s
+        if args.code_index_stall_timeout_s is not None
+        else sr.DEFAULT_CODE_INDEX_STALL_TIMEOUT_S
+    )
     source_cache_dir = Path(args.source_cache_dir) if args.source_cache_dir else None
     source_dir_roots = [Path(p) for p in args.source_dir_root]
     if args.source_cache_only and source_cache_dir is None:
@@ -4377,6 +4394,14 @@ def _on_signal(signum, _frame):
     # CONVEYOR_MANIFEST; ConcurrentManifest merges + flocks every write so they
     # cannot clobber each other's resume/accounting keys (H4 fix).
     manifest = ConcurrentManifest.load(CONVEYOR_MANIFEST)
+    submitted_code_project_identities: dict[str, str] = {}
+    if args.streams == "code" and repo_list is not None:
+        for key in manifest.done:
+            if not key.endswith("::code"):
+                continue
+            repo_name = key[:-len("::code")]
+            project_id = sr.resolve_project_identity(repo_name, repo_list)
+            submitted_code_project_identities.setdefault(project_id, repo_name)
     try:
         manifest.bind_code_revision(revision_guard.receipt)
         child_guard_path = install_code_revision_child_guard(
@@ -4564,14 +4589,44 @@ def should_process(repo: str) -> bool:
     range_pool = ThreadPoolExecutor(max_workers=workers)
     repo_pool = ThreadPoolExecutor(max_workers=repo_workers) if repo_workers > 1 else None
 
-    def claim_repo_once(repo: str) -> None:
+    def claim_repo_once(repo: str) -> bool:
         if repo in submitted_repo_names:
             raise RepoFailure(
                 repo,
                 "duplicate_repo",
                 f"repo {repo!r} was yielded/submitted twice in one conveyor run",
             )
         submitted_repo_names.add(repo)
+        if args.streams != "code" or repo_list is None:
+            return True
+        try:
+            project_id = sr.resolve_project_identity(repo, repo_list)
+        except Exception as exc:
+            raise RepoFailure(repo, "project_identity", str(exc)) from exc
+        previous = claim_code_project_identity(
+            submitted_code_project_identities,
+            repo,
+            project_id,
+        )
+        if previous is None:
+            return True
+        info = {
+            "source": "code",
+            "skipped": True,
+            "skip_reason": "duplicate_project_identity",
+            "project_identity": project_id,
+            "canonical_repo": previous,
+            "recorded_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
+        }
+        with manifest_lock:
+            manifest.mark_done(code_key(repo), info)
+        progress.emit(
+            "duplicate_project_identity_skipped",
+            repo=repo,
+            stream="code",
+            **info,
+        )
+        return False
 
     def mark_no_git_repo(repo: str) -> None:
         info = {
@@ -4611,7 +4666,8 @@ def mark_no_git_repo(repo: str) -> None:
                     if hasattr(gen, "close"):
                         gen.close()
                     break
-                claim_repo_once(repo)
+                if not claim_repo_once(repo):
+                    continue
                 revision_guard.verify(f"submit repo worker for {repo}")
                 res = process_one_repo(
                     repo, repo_dir, lengths_code, lengths_commits, args.range_size,
@@ -4694,7 +4750,8 @@ def drain_one_or_more(block: bool = True) -> None:
                     if hasattr(gen, "close"):
                         gen.close()
                     break
-                claim_repo_once(repo)
+                if not claim_repo_once(repo):
+                    continue
                 while len(inflight) >= max_active_repos:
                     drain_one_or_more(block=True)
                     # Worker threads mutate cumulative['valid'] under manifest_lock

scripts/streaming_reindex.py

@@ -91,6 +91,8 @@ def _load_local_symbol_identity() -> ModuleType:
 # Tokenize at the model's full context so packing decides the final lengths.
 TOKENIZE_BUDGET = 65536
 DEFAULT_TARGET_LENGTHS = (1024, 2048, 4096)
+DEFAULT_CODE_INDEX_TIMEOUT_S = 4 * 60 * 60
+DEFAULT_CODE_INDEX_STALL_TIMEOUT_S = 10 * 60
 
 # Directories never worth indexing (VCS / build artifacts). index_project has its
 # own excludes; we additionally avoid extracting .git to keep staging small for
@@ -382,65 +384,6 @@ def _subprocess_env() -> dict:
     return env
 
 
-def _parse_ps_time_seconds(value: str) -> float | None:
-    """Parse ps TIME values like MM:SS.cc, HH:MM:SS.cc, or DD-HH:MM:SS.cc."""
-    raw = value.strip()
-    if not raw:
-        return None
-    days = 0
-    if "-" in raw:
-        day_raw, raw = raw.split("-", 1)
-        try:
-            days = int(day_raw)
-        except ValueError:
-            return None
-    parts = raw.split(":")
-    try:
-        if len(parts) == 2:
-            hours = 0
-            minutes = int(parts[0])
-            seconds = float(parts[1])
-        elif len(parts) == 3:
-            hours = int(parts[0])
-            minutes = int(parts[1])
-            seconds = float(parts[2])
-        else:
-            return None
-    except ValueError:
-        return None
-    return float(days * 86400 + hours * 3600 + minutes * 60) + seconds
-
-
-def _process_group_cpu_seconds(pgid: int) -> float | None:
-    """Return cumulative CPU seconds for a process group, when ps supports it."""
-    try:
-        output = subprocess.check_output(
-            ["ps", "-axo", "pgid=,time="],
-            text=True,
-            stderr=subprocess.DEVNULL,
-        )
-    except (OSError, subprocess.SubprocessError):
-        return None
-    total = 0.0
-    seen = False
-    for line in output.splitlines():
-        fields = line.split()
-        if len(fields) < 2:
-            continue
-        try:
-            row_pgid = int(fields[0])
-        except ValueError:
-            continue
-        if row_pgid != pgid:
-            continue
-        seconds = _parse_ps_time_seconds(fields[1])
-        if seconds is None:
-            continue
-        total += seconds
-        seen = True
-    return total if seen else None
-
-
 def run_checked(
     repo: str,
     stage: str,
@@ -494,7 +437,6 @@ def terminate_process(reason: str) -> None:
             deadline = time.monotonic() + timeout if timeout and timeout > 0 else None
             last_activity = time.monotonic()
             last_signature: tuple[int, int] | None = None
-            last_cpu_seconds: float | None = None
             while proc.poll() is None:
                 now = time.monotonic()
                 if deadline is not None and now > deadline:
@@ -510,18 +452,12 @@ def terminate_process(reason: str) -> None:
                     if signature != last_signature:
                         last_signature = signature
                         last_activity = now
-                cpu_seconds = _process_group_cpu_seconds(proc.pid)
-                if cpu_seconds is not None and cpu_seconds != last_cpu_seconds:
-                    last_cpu_seconds = cpu_seconds
-                    last_activity = now
                 if now - last_activity > stall_timeout:
-                    terminate_process(
-                        f"no log/CPU progress for {stall_timeout}s"
-                    )
+                    terminate_process(f"no log progress for {stall_timeout}s")
                     raise RepoFailure(
                         repo,
                         stage,
-                        f"stalled after {stall_timeout}s without log or CPU progress",
+                        f"stalled after {stall_timeout}s without log progress",
                     )
                 time.sleep(1.0)
             stdout_data, _ = proc.communicate(timeout=1)
@@ -1498,13 +1434,14 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
     p.add_argument("--parse-workers", type=int, default=2,
                    help="Parse workers passed to index_project for the code stage "
                         "(default 2; keep this low when multiple repos run).")
-    p.add_argument("--code-index-timeout-s", type=int, default=0,
-                   help="Optional fail-loud timeout for each index_project code "
-                        "stage. 0 disables the timeout (default).")
-    p.add_argument("--code-index-stall-timeout-s", type=int, default=0,
-                   help="Optional fail-loud stall watchdog for index_project: "
-                        "kill when the log file has no size/mtime progress for "
-                        "this many seconds. 0 disables the watchdog (default).")
+    p.add_argument("--code-index-timeout-s", type=int,
+                   default=DEFAULT_CODE_INDEX_TIMEOUT_S,
+                   help="Fail-loud timeout for each index_project code stage "
+                        f"(default {DEFAULT_CODE_INDEX_TIMEOUT_S}s; 0 disables).")
+    p.add_argument("--code-index-stall-timeout-s", type=int,
+                   default=DEFAULT_CODE_INDEX_STALL_TIMEOUT_S,
+                   help="Fail-loud log-heartbeat watchdog for index_project "
+                        f"(default {DEFAULT_CODE_INDEX_STALL_TIMEOUT_S}s; 0 disables).")
     p.add_argument("--source-cache-dir", default=None,
                    help="Optional repo-level source cache for code-only runs. "
                         "Complete cached repos are reused before opening the "

tests/test_clang_indexer_parse_pool.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+
+def test_parse_executor_recycles_long_lived_libclang_workers() -> None:
+    from tools.clang_indexer.index_project import make_parse_executor
+
+    with make_parse_executor(max_workers=1, max_tasks_per_child=1) as executor:
+        worker_pids = [executor.submit(os.getpid).result(timeout=10) for _ in range(3)]
+
+    assert len(set(worker_pids)) == 3
+
+
+def test_c_file_with_cpp_construct_keeps_cpp_language_mode(tmp_path: Path) -> None:
+    from tools.clang_indexer.index_project import _adapt_args_for_file
+
+    source = tmp_path / "compiler-regression.c"
+    source.write_text(
+        "char *foo(void) { return(::new char[2] ('a', 'b')); }\n",
+        encoding="utf-8",
+    )
+
+    adapted = _adapt_args_for_file(["-std=c++17"], str(source))
+
+    assert adapted[:2] == ["-x", "c++"]
+    assert "-std=c++17" in adapted
+
+    plain_c = tmp_path / "plain.c"
+    plain_c.write_text("int add(int a, int b) { return a + b; }\n", encoding="utf-8")
+    plain_args = _adapt_args_for_file(["-std=c++17"], str(plain_c))
+
+    assert plain_args[:2] == ["-x", "c"]
+    assert "-std=c11" in plain_args
+
+
+def test_training_document_generation_emits_periodic_heartbeat(capsys) -> None:
+    from tools.clang_indexer.index_project import (
+        DOCUMENT_HEARTBEAT_ITEMS,
+        FunctionDef,
+        ProjectIndex,
+        build_training_documents,
+    )
+
+    index = ProjectIndex()
+    total = DOCUMENT_HEARTBEAT_ITEMS + 1
+    for item in range(total):
+        index.add_function(
+            FunctionDef(
+                name=f"function_{item}",
+                qualified_name=f"function_{item}",
+                file=f"file_{item}.cpp",
+                line=1,
+                text=(
+                    f"int function_{item}(int value) {{ int first = value + {item}; "
+                    "int second = first * 2; int third = second - value; "
+                    "return third + first + second; }"
+                ),
+                callees=[],
+            )
+        )
+
+    docs = build_training_documents(index, enriched=False)
+
+    captured = capsys.readouterr()
+    assert len(docs) == total
+    assert (
+        f"Training document generation heartbeat: {DOCUMENT_HEARTBEAT_ITEMS}/{total}"
+        in captured.err
+    )

tests/test_streaming_conveyor_progress.py

@@ -25,6 +25,23 @@ def _repo_identity_map(tmp_path: Path) -> Path:
     return path
 
 
+def test_code_project_identity_claims_deduplicate_aliases() -> None:
+    import streaming_conveyor
+
+    claims: dict[str, str] = {}
+
+    assert streaming_conveyor.claim_code_project_identity(
+        claims,
+        "open-watcom",
+        "open-watcom/open-watcom-v2",
+    ) is None
+    assert streaming_conveyor.claim_code_project_identity(
+        claims,
+        "open-watcom-v2",
+        "open-watcom/open-watcom-v2",
+    ) == "open-watcom"
+
+
 def test_progress_writer_appends_jsonl(tmp_path):
     import streaming_conveyor