fix(conveyor): quarantine unmapped archives and preflight libclang

apstenku123 · apstenku123 · commit e70c220cf8cd · 2026-07-18T15:03:05.000+02:00
diff --git a/scripts/streaming_conveyor.py b/scripts/streaming_conveyor.py
@@ -167,6 +167,12 @@
 EXTRACT_CACHE_MODE = EXTRACT_CACHE_MODE_RUN_LOCAL
 
 _PRINT_LOCK = threading.Lock()
+_LIBCLANG_PREFLIGHT_CODE = """
+import clang.cindex as cindex
+
+cindex.Index.create()
+print(cindex.__file__)
+"""
 
 
 class CodeRevisionError(RuntimeError):
@@ -181,6 +187,39 @@ class CodeRevisionDriftError(CodeRevisionError):
     """The checkout changed after the conveyor captured its revision."""
 
 
+def verify_libclang_preflight(python: Path = VENV_PYTHON) -> str:
+    """Prove the stage interpreter can load and initialize libclang."""
+
+    try:
+        completed = subprocess.run(
+            [str(python), "-c", _LIBCLANG_PREFLIGHT_CODE],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise SystemExit(
+            f"libclang preflight timed out after 30s: python={python}"
+        ) from exc
+    except OSError as exc:
+        raise SystemExit(
+            f"libclang preflight could not start: python={python}: {exc}"
+        ) from exc
+    if completed.returncode != 0:
+        detail = (completed.stderr or completed.stdout).strip()
+        raise SystemExit(
+            "libclang preflight failed: "
+            f"python={python} exit={completed.returncode}: {detail}"
+        )
+    binding_path = completed.stdout.strip()
+    if not binding_path:
+        raise SystemExit(
+            f"libclang preflight returned no binding path: python={python}"
+        )
+    return binding_path
+
+
 def _sha256_bytes(payload: bytes) -> str:
     return hashlib.sha256(payload).hexdigest()
 
@@ -2511,6 +2550,67 @@ def claim_code_project_identity(
     return previous
 
 
+def claim_code_repo_for_submission(
+    *,
+    repo: str,
+    repo_list: Path,
+    project_identity_claims: dict[str, str],
+    manifest: Manifest,
+    manifest_lock: threading.Lock,
+    progress: ProgressWriter,
+) -> bool:
+    """Claim a code repo without letting one unresolved identity stop the run.
+
+    Missing canonical identity is a fail-closed, repo-local data rejection: no
+    indexing subprocess may run, but unrelated repositories must continue.
+    Corrupt repo-list files and other unexpected failures still propagate.
+    """
+
+    try:
+        project_id = sr.resolve_project_identity(repo, repo_list)
+    except SymbolIdentityError as exc:
+        detail = str(exc)
+        unit = code_key(repo)
+        with manifest_lock:
+            manifest.mark_failed(unit, "project_identity", detail)
+        progress.emit(
+            "unit_failed",
+            stream="code",
+            repo=repo,
+            unit=unit,
+            stage="project_identity",
+            detail=detail[:2000],
+        )
+        _log(f"QUARANTINE {unit}: unresolved project identity: {detail}")
+        return False
+
+    previous = claim_code_project_identity(
+        project_identity_claims,
+        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 no_git_key(repo: str) -> str:
     """Checkpoint key proving the .git-preserving stream saw no git metadata."""
     return f"{repo}::no_git"
@@ -4336,6 +4436,12 @@ def _on_signal(signum, _frame):
     for path in required_paths:
         if not Path(path).exists():
             raise SystemExit(f"required path missing: {path}")
+    if not args.source_cache_populate_only:
+        libclang_binding = verify_libclang_preflight(VENV_PYTHON)
+        _log(
+            "libclang preflight: OK "
+            f"python={VENV_PYTHON} binding={libclang_binding}"
+        )
 
     # Pre-create output trees for BOTH streams.
     CONVEYOR_ROOT.mkdir(parents=True, exist_ok=True)
@@ -4599,34 +4705,14 @@ def claim_repo_once(repo: str) -> bool:
         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",
+        return claim_code_repo_for_submission(
             repo=repo,
-            stream="code",
-            **info,
+            repo_list=repo_list,
+            project_identity_claims=submitted_code_project_identities,
+            manifest=manifest,
+            manifest_lock=manifest_lock,
+            progress=progress,
         )
-        return False
 
     def mark_no_git_repo(repo: str) -> None:
         info = {
diff --git a/tests/test_streaming_conveyor_progress.py b/tests/test_streaming_conveyor_progress.py
@@ -42,6 +42,90 @@ def test_code_project_identity_claims_deduplicate_aliases() -> None:
     ) == "open-watcom"
 
 
+def test_libclang_preflight_fails_before_staging_with_child_error(tmp_path) -> None:
+    import streaming_conveyor
+
+    fake_python = tmp_path / "python"
+    fake_python.write_text(
+        "#!/bin/sh\necho 'clang bindings unavailable' >&2\nexit 17\n",
+        encoding="utf-8",
+    )
+    fake_python.chmod(0o755)
+
+    with pytest.raises(
+        SystemExit,
+        match=(
+            r"libclang preflight failed: .*exit=17: "
+            r"clang bindings unavailable"
+        ),
+    ):
+        streaming_conveyor.verify_libclang_preflight(fake_python)
+
+
+def test_unmapped_code_repo_is_quarantined_without_blocking_next_submission(
+    tmp_path,
+) -> None:
+    import threading
+
+    import streaming_conveyor
+
+    repo_list = tmp_path / "repo_list.json"
+    repo_list.write_text(
+        json.dumps(
+            {
+                "repos": [
+                    {
+                        "name": "public-repo",
+                        "owner_repo": "tests/public-repo",
+                    }
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+    manifest = streaming_conveyor.Manifest(tmp_path / "_done.json")
+    progress_path = tmp_path / "progress.jsonl"
+    progress = streaming_conveyor.ProgressWriter(progress_path)
+    claims: dict[str, str] = {}
+    lock = threading.Lock()
+
+    assert not streaming_conveyor.claim_code_repo_for_submission(
+        repo="unmapped archive",
+        repo_list=repo_list,
+        project_identity_claims=claims,
+        manifest=manifest,
+        manifest_lock=lock,
+        progress=progress,
+    )
+    assert manifest.failed["unmapped archive::code"]["stage"] == "project_identity"
+    assert claims == {}
+
+    assert streaming_conveyor.claim_code_repo_for_submission(
+        repo="public-repo",
+        repo_list=repo_list,
+        project_identity_claims=claims,
+        manifest=manifest,
+        manifest_lock=lock,
+        progress=progress,
+    )
+    assert claims == {"tests/public-repo": "public-repo"}
+
+    events = [
+        json.loads(line)
+        for line in progress_path.read_text(encoding="utf-8").splitlines()
+    ]
+    assert len(events) == 1
+    assert events[0]["event"] == "unit_failed"
+    assert events[0]["repo"] == "unmapped archive"
+    assert events[0]["unit"] == "unmapped archive::code"
+    assert events[0]["stream"] == "code"
+    assert events[0]["stage"] == "project_identity"
+    assert events[0]["detail"] == (
+        f"{repo_list}: no canonical project identity for bare repo "
+        "'unmapped archive'"
+    )
+
+
 def test_progress_writer_appends_jsonl(tmp_path):
     import streaming_conveyor
 
