Stream source cache populate progress

Author: apstenku123

scripts/streaming_conveyor.py

@@ -1224,20 +1224,23 @@ def populate_code_source_cache(
 ) -> dict:
     """Populate the code source cache without running index/tokenize stages."""
     started = time.monotonic()
+
+    def emit_repo_ready(repo: str, repo_dir: Path, repo_count: int) -> None:
+        progress.emit(
+            "source_cache_repo_ready",
+            repo=repo,
+            repo_dir=str(repo_dir),
+            source_cache_dir=str(source_cache_dir),
+            repo_count=repo_count,
+        )
+
     report = sr.populate_source_cache(
         work_root,
         should_process,
         source_cache_dir,
         max_repos=max_repos,
+        on_repo_ready=emit_repo_ready,
     )
-    for idx, item in enumerate(report["repos"], start=1):
-        progress.emit(
-            "source_cache_repo_ready",
-            repo=item["repo"],
-            repo_dir=item["path"],
-            source_cache_dir=str(source_cache_dir),
-            repo_count=idx,
-        )
     report = {
         **report,
         "elapsed_s": round(time.monotonic() - started, 6),

scripts/streaming_reindex.py

@@ -665,6 +665,7 @@ def populate_source_cache(
     source_cache_dir: Path,
     *,
     max_repos: int | None = None,
+    on_repo_ready: Callable[[str, Path, int], None] | None = None,
 ) -> dict:
     """Materialize source-cache repos without running the tokenization pipeline.
 
@@ -684,6 +685,8 @@ def populate_source_cache(
     try:
         for repo, repo_dir in gen:
             repos.append({"repo": repo, "path": str(repo_dir)})
+            if on_repo_ready is not None:
+                on_repo_ready(repo, repo_dir, len(repos))
             if max_repos is not None and len(repos) >= max_repos:
                 break
     finally:

tests/test_streaming_conveyor_progress.py

@@ -66,15 +66,25 @@ def test_populate_code_source_cache_emits_ready_and_summary(tmp_path, monkeypatc
     cache = tmp_path / "source_cache"
     observed = {}
 
-    def fake_populate(work_root, should_process, source_cache_dir, *, max_repos):
+    def fake_populate(
+        work_root,
+        should_process,
+        source_cache_dir,
+        *,
+        max_repos,
+        on_repo_ready,
+    ):
         observed["work_root"] = work_root
         observed["source_cache_dir"] = source_cache_dir
         observed["max_repos"] = max_repos
+        observed["has_callback"] = on_repo_ready is not None
         repos = []
         for repo in ("repo-a", "repo.bare", "repo-b"):
             if should_process(repo):
                 repos.append({"repo": repo, "path": str(source_cache_dir / repo)})
         repos = repos[:max_repos]
+        for idx, item in enumerate(repos, start=1):
+            on_repo_ready(item["repo"], Path(item["path"]), idx)
         return {
             "source_cache_dir": str(source_cache_dir),
             "repos": repos,
@@ -95,6 +105,7 @@ def fake_populate(work_root, should_process, source_cache_dir, *, max_repos):
         "work_root": tmp_path / "work",
         "source_cache_dir": cache,
         "max_repos": 1,
+        "has_callback": True,
     }
     assert report["repo_count"] == 1
     rows = [

tests/test_streaming_reindex_run_checked.py

@@ -170,14 +170,17 @@ def fake_stream(work_root, should_process, *, source_cache_dir, source_cache_onl
 
     monkeypatch.setattr(streaming_reindex, "stream_repo_subtrees", fake_stream)
 
+    ready = []
     report = streaming_reindex.populate_source_cache(
         tmp_path / "work",
         streaming_reindex.is_code_worktree_repo,
         cache,
         max_repos=1,
+        on_repo_ready=lambda repo, path, count: ready.append((repo, path, count)),
     )
 
     assert calls == [(tmp_path / "work", cache, False)]
+    assert ready == [("repo-a", cache / "repo-a", 1)]
     assert report == {
         "source_cache_dir": str(cache),
         "repos": [{"repo": "repo-a", "path": str(cache / "repo-a")}],