feat: add support for skipping repositories with no trainable sources and cleanup outputs on deduplication failures.

Author: apstenku123

scripts/streaming_conveyor.py

@@ -68,7 +68,7 @@
     wait,
 )
 from pathlib import Path
-from typing import Callable, Sequence
+from typing import Callable, Iterable, Sequence
 
 # Reuse the proven machinery. streaming_reindex_commits already re-exports the
 # shared primitives from streaming_reindex (MLX_ROOT, Manifest, RepoFailure, ...)
@@ -1362,16 +1362,40 @@ def run_code_half(
     stage_db = sr.code_stage_db(work_root / repo, repo) if dedup_db is not None else None
     promoted = False
     try:
-        info = sr.process_one_repo(
-            repo, repo_dir, lengths_code, work_root, dedup_db, dedup_near,
-            global_symbol_index, memory_limit_gb, parse_workers, index_timeout_s,
-            index_stall_timeout_s,
-            promote_dedup_on_success=False,
-        )
+        try:
+            info = sr.process_one_repo(
+                repo, repo_dir, lengths_code, work_root, dedup_db, dedup_near,
+                global_symbol_index, memory_limit_gb, parse_workers, index_timeout_s,
+                index_stall_timeout_s,
+                promote_dedup_on_success=False,
+            )
+        except RepoFailure as exc:
+            if is_no_trainable_source_failure(exc):
+                promoted = True
+                return {
+                    "source": "code",
+                    "repo": repo,
+                    "skipped": True,
+                    "skip_reason": "no_trainable_source",
+                    "lengths": {},
+                    "stage_timings_s": {},
+                    "detail": exc.detail,
+                }
+            raise
         if info.get("skipped"):
             return info
-        # zstd-max the per-length code parquet files this repo just wrote.
         timings = dict(info.get("stage_timings_s", {}))
+        try:
+            timings.update(sr.promote_dedup_stage(dedup_db, stage_id, stage_db))
+        except Exception as exc:
+            remove_code_outputs(repo, info.get("lengths", {}).keys())
+            raise RepoFailure(
+                repo,
+                "dedup_promote",
+                f"{type(exc).__name__}: {exc}",
+            ) from exc
+        promoted = True
+        # zstd-max the per-length code parquet files this repo just wrote.
         started = time.monotonic()
         deferred = 0
         for L in info.get("lengths", {}):
@@ -1385,9 +1409,7 @@ def run_code_half(
         timings["recompress_s"] = round(time.monotonic() - started, 6)
         if deferred:
             timings["recompress_deferred"] = float(deferred)
-        timings.update(sr.promote_dedup_stage(dedup_db, stage_id, stage_db))
         info["stage_timings_s"] = timings
-        promoted = True
         return info
     finally:
         if not promoted:
@@ -1413,6 +1435,20 @@ def run_code_half(
 ]
 
 
+def remove_code_outputs(repo: str, lengths: Iterable[str | int]) -> None:
+    for length in lengths:
+        dest = sr.OUTPUT_ROOT / str(length) / f"{repo}.parquet"
+        if dest.exists():
+            dest.unlink()
+
+
+def is_no_trainable_source_failure(exc: RepoFailure) -> bool:
+    return (
+        exc.stage == "index_project"
+        and "no training docs (no_trainable_source)" in exc.detail.lower()
+    )
+
+
 def is_retryable_index_project_failure(exc: RepoFailure) -> bool:
     detail = exc.detail.lower()
     return (
@@ -2303,33 +2339,50 @@ def process_one_repo(
                             code_index_stall_timeout_s,
                             recompressor=code_recompressor,
                         )
-                        with manifest_lock:
-                            manifest.mark_done(ck, cinfo)
-                        totals = _length_totals(cinfo)
-                        with manifest_lock:
-                            cumulative["valid"] += totals["valid_tokens"]
-                            cumulative_valid = cumulative["valid"]
-                        if progress is not None:
-                            progress.emit(
-                                "unit_done",
-                                stream="code",
-                                repo=repo,
-                                unit=ck,
-                                rows=totals["rows"],
-                                valid_tokens=totals["valid_tokens"],
-                                capacity_tokens=totals["capacity_tokens"],
-                                **_primary_bucket_progress(cinfo, lengths_code),
-                                lengths=cinfo.get("lengths", {}),
-                                stage_timings_s=cinfo.get("stage_timings_s", {}),
-                                cumulative_valid_tokens=cumulative_valid,
+                        if cinfo.get("skipped"):
+                            with manifest_lock:
+                                manifest.mark_done(ck, cinfo)
+                            if progress is not None:
+                                progress.emit(
+                                    "unit_skipped",
+                                    stream="code",
+                                    repo=repo,
+                                    unit=ck,
+                                    reason=cinfo.get("skip_reason", "skipped"),
+                                    detail=str(cinfo.get("detail", ""))[:2000],
+                                )
+                            result["code"] = cinfo
+                            _log(
+                                f"SKIP {ck}: {cinfo.get('skip_reason', 'skipped')}"
+                            )
+                        else:
+                            with manifest_lock:
+                                manifest.mark_done(ck, cinfo)
+                            totals = _length_totals(cinfo)
+                            with manifest_lock:
+                                cumulative["valid"] += totals["valid_tokens"]
+                                cumulative_valid = cumulative["valid"]
+                            if progress is not None:
+                                progress.emit(
+                                    "unit_done",
+                                    stream="code",
+                                    repo=repo,
+                                    unit=ck,
+                                    rows=totals["rows"],
+                                    valid_tokens=totals["valid_tokens"],
+                                    capacity_tokens=totals["capacity_tokens"],
+                                    **_primary_bucket_progress(cinfo, lengths_code),
+                                    lengths=cinfo.get("lengths", {}),
+                                    stage_timings_s=cinfo.get("stage_timings_s", {}),
+                                    cumulative_valid_tokens=cumulative_valid,
+                                )
+                            if checkpoint is not None:
+                                checkpoint.maybe_checkpoint(cumulative_valid)
+                            result["code"] = cinfo
+                            _log(
+                                f"DONE {ck}: buckets={sorted(cinfo['lengths'].keys())} "
+                                f"(cum_all={cumulative['valid']})"
                             )
-                        if checkpoint is not None:
-                            checkpoint.maybe_checkpoint(cumulative_valid)
-                        result["code"] = cinfo
-                        _log(
-                            f"DONE {ck}: buckets={sorted(cinfo['lengths'].keys())} "
-                            f"(cum_all={cumulative['valid']})"
-                        )
                 except RepoFailure as exc:
                     _log(f"FAIL {ck}: {exc}")
                     with manifest_lock:

tests/test_macro_scan_selection.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def test_macro_scan_roots_are_limited_to_trainable_code_and_indexed_headers(tmp_path: Path) -> None:
+    from tools.clang_indexer import index_project
+
+    src = tmp_path / "src"
+    include = tmp_path / "include"
+    src.mkdir()
+    include.mkdir()
+    used_cpp = src / "used.cpp"
+    unused_cpp = src / "unused.cpp"
+    header = include / "api.hpp"
+    unused_header = include / "unused.hpp"
+    used_cpp.write_text("int used() { return 1; }\n", encoding="utf-8")
+    unused_cpp.write_text("// no trainable declarations here\n", encoding="utf-8")
+    header.write_text("template <class T> struct Box { T value; };\n", encoding="utf-8")
+    unused_header.write_text("#define UNUSED_VALUE 1\n", encoding="utf-8")
+
+    index = index_project.ProjectIndex()
+    index.add_function(
+        index_project.FunctionDef(
+            name="used",
+            qualified_name="used",
+            file="src/used.cpp",
+            line=1,
+            text="int used() { return 1; }",
+            callees=[],
+            is_definition=True,
+        )
+    )
+    index.add_typedef(
+        index_project.TypeDef(
+            name="Box",
+            qualified_name="Box",
+            file="include/api.hpp",
+            line=1,
+            text="template <class T> struct Box { T value; };",
+            kind=4,
+        )
+    )
+
+    roots = index_project.select_macro_scan_files(
+        [str(used_cpp), str(unused_cpp), str(header), str(unused_header)],
+        index,
+        [str(header), str(unused_header)],
+        project_dir=str(tmp_path),
+    )
+
+    rel_roots = {Path(path).relative_to(tmp_path).as_posix() for path in roots}
+    assert rel_roots == {"include/api.hpp", "src/used.cpp"}

tests/test_streaming_conveyor_progress.py

@@ -465,6 +465,73 @@ def runner(
     assert info["lengths"]["1024"]["valid_tokens"] == 1024
 
 
+def test_run_code_half_removes_outputs_when_dedup_promote_fails(tmp_path, monkeypatch):
+    import streaming_conveyor
+
+    repo = "repo"
+    output_root = tmp_path / "out"
+    dest = output_root / "1024" / f"{repo}.parquet"
+    dest.parent.mkdir(parents=True)
+    dest.write_bytes(b"partial parquet already appended")
+    monkeypatch.setattr(streaming_conveyor.sr, "OUTPUT_ROOT", output_root)
+
+    def fake_process_one_repo(*_args, **_kwargs):
+        return {
+            "source": "code",
+            "lengths": {"1024": _stat(rows=1, valid=1024, target_length=1024)},
+            "stage_timings_s": {},
+        }
+
+    def fail_promote(*_args, **_kwargs):
+        raise sqlite3.OperationalError("unable to open database file")
+
+    monkeypatch.setattr(streaming_conveyor.sr, "process_one_repo", fake_process_one_repo)
+    monkeypatch.setattr(streaming_conveyor.sr, "promote_dedup_stage", fail_promote)
+
+    try:
+        streaming_conveyor.run_code_half(
+            repo,
+            tmp_path / "src",
+            (1024,),
+            tmp_path / "work",
+            tmp_path / "dedup.sqlite",
+            True,
+        )
+    except streaming_conveyor.RepoFailure as exc:
+        assert exc.stage == "dedup_promote"
+        assert "OperationalError" in exc.detail
+    else:
+        raise AssertionError("run_code_half should fail when dedup promote fails")
+
+    assert not dest.exists()
+
+
+def test_run_code_half_skips_no_trainable_source(tmp_path, monkeypatch):
+    import streaming_conveyor
+
+    def no_trainable_source(*_args, **_kwargs):
+        raise streaming_conveyor.RepoFailure(
+            "repo",
+            "index_project",
+            "no training docs (no_trainable_source): empty enriched jsonl",
+        )
+
+    monkeypatch.setattr(streaming_conveyor.sr, "process_one_repo", no_trainable_source)
+
+    info = streaming_conveyor.run_code_half(
+        "repo",
+        tmp_path / "src",
+        (1024,),
+        tmp_path / "work",
+        tmp_path / "dedup.sqlite",
+        True,
+    )
+
+    assert info["skipped"] is True
+    assert info["skip_reason"] == "no_trainable_source"
+    assert info["lengths"] == {}
+
+
 def test_failed_code_unit_was_index_memory_reads_manifest_receipt(tmp_path):
     import streaming_conveyor