Refactor local git layers into domains

Author: rudironsoni

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "typer>=0.9.0",
     "msgspec>=0.19.0",
     "pathspec>=0.12.1",
+    "pygit2>=1.19.0",
     "pyyaml>=6.0",
     "questionary>=2.0.0",
 ]

src/cocoindex_code/cli.py

@@ -516,6 +516,42 @@ def _setup_user_settings_interactive(litellm_model_flag: str | None) -> None:
     _typer.echo()
 
 
+def _register_overlay_policy(project_root: Path, base_ref: str) -> None:
+    from ._daemon_paths import daemon_state_dir
+    from .embedder_params import resolve_embedder_params
+    from .layered_project import build_index_config_hash
+    from .layers import LayerStore
+    from .settings import load_user_settings
+    from .version_control import GitContextError, resolve_worktree
+
+    user_settings = load_user_settings()
+    params = resolve_embedder_params(user_settings.embedding)
+    config_hash = build_index_config_hash(
+        project_root,
+        indexing_params=params.indexing,
+        query_params=params.query,
+    )
+    try:
+        worktree = resolve_worktree(
+            project_root, base_ref=base_ref, index_config_hash=config_hash
+        )
+    except GitContextError as e:
+        _typer.echo(f"Warning: could not register Git overlay policy: {e}", err=True)
+        return
+
+    store = LayerStore(daemon_state_dir() / "daemon.db")
+    store.upsert_repository(
+        repo_id=worktree.repository.id,
+        repo_name=worktree.repository.repo_name,
+        remote_url=worktree.repository.remote_url,
+        normalized_remote_url=worktree.repository.normalized_remote_url,
+        repo_relative_root=worktree.repository.repo_relative_root,
+        last_seen_root=worktree.repository.last_seen_root,
+    )
+    store.upsert_overlay_policy(repo_id=worktree.repository.id, base_ref=worktree.branch.base_ref)
+    _typer.echo(f"Registered Git overlay base: {worktree.branch.base_ref}")
+
+
 @app.command()
 def init(
     litellm_model: str | None = _typer.Option(
@@ -524,6 +560,7 @@ def init(
         help="Use the given LiteLLM model and skip provider/model prompts.",
     ),
     force: bool = _typer.Option(False, "-f", "--force", help="Skip parent directory warning"),
+    base_ref: str | None = _typer.Option(None, "--base", help="Git base ref for overlays"),
 ) -> None:
     """Initialize a project for cocoindex-code."""
     cwd = Path.cwd().resolve()
@@ -545,6 +582,8 @@ def init(
     # Check if already initialized
     if settings_file.is_file():
         _typer.echo("Project already initialized.")
+        if base_ref is not None:
+            _register_overlay_policy(cwd, base_ref)
         return
 
     # Check parent directories for markers
@@ -563,6 +602,9 @@ def init(
     save_project_settings(cwd, default_project_settings())
     _typer.echo(f"Created project settings: {format_path_for_display(settings_file)}")
 
+    if base_ref is not None:
+        _register_overlay_policy(cwd, base_ref)
+
     # Add to .gitignore
     add_to_gitignore(cwd)
 

src/cocoindex_code/git_context.py

@@ -1,178 +1,13 @@
-"""Git worktree identity resolution for shared daemon layers."""
-
 from __future__ import annotations
 
-import hashlib
-import os
-import subprocess
-from dataclasses import dataclass
-from pathlib import Path
-from urllib.parse import urlparse
-
-
-class GitContextError(RuntimeError):
-    """Raised when a directory cannot be resolved as a usable Git worktree."""
-
-
-@dataclass(frozen=True)
-class GitStatusEntry:
-    index_status: str
-    worktree_status: str
-    path: str
-    original_path: str | None = None
-
-
-@dataclass(frozen=True)
-class WorktreeContext:
-    repo_id: str
-    worktree_id: str
-    repo_root: Path
-    git_common_dir: Path
-    remote_url: str
-    normalized_remote_url: str
-    branch_name: str
-    head_commit: str
-    base_ref: str
-    base_commit: str
-    merge_base: str
-    dirty_snapshot_hash: str | None
-    status_entries: tuple[GitStatusEntry, ...]
-
-
-def _git(cwd: Path, *args: str) -> str:
-    try:
-        return subprocess.check_output(
-            ["git", "-C", str(cwd), *args],
-            stderr=subprocess.DEVNULL,
-            text=True,
-        ).strip()
-    except subprocess.CalledProcessError as e:
-        raise GitContextError(f"git {' '.join(args)} failed in {cwd}") from e
-
-
-def _git_bytes(cwd: Path, *args: str) -> bytes:
-    try:
-        return subprocess.check_output(
-            ["git", "-C", str(cwd), *args],
-            stderr=subprocess.DEVNULL,
-        )
-    except subprocess.CalledProcessError as e:
-        raise GitContextError(f"git {' '.join(args)} failed in {cwd}") from e
-
-
-def normalize_remote_url(url: str) -> str:
-    """Normalize common Git remote URL forms into a stable lowercase identity."""
-    raw = url.strip()
-    if raw.endswith(".git"):
-        raw = raw[:-4]
-    if raw.startswith("git@") and ":" in raw:
-        host, path = raw[4:].split(":", 1)
-        return f"{host.lower()}/{path.strip('/').lower()}"
-    parsed = urlparse(raw)
-    if parsed.scheme and parsed.netloc:
-        path = parsed.path.strip("/")
-        return f"{parsed.netloc.lower()}/{path.lower()}"
-    return raw.strip("/").lower()
-
-
-def _parse_status(raw: bytes) -> tuple[GitStatusEntry, ...]:
-    if not raw:
-        return ()
-    parts = raw.decode("utf-8", errors="surrogateescape").split("\0")
-    entries: list[GitStatusEntry] = []
-    i = 0
-    while i < len(parts):
-        item = parts[i]
-        i += 1
-        if not item:
-            continue
-        status = item[:2]
-        path = item[3:]
-        original_path: str | None = None
-        if status[0] in {"R", "C"} or status[1] in {"R", "C"}:
-            if i < len(parts):
-                original_path = parts[i] or None
-                i += 1
-        entries.append(
-            GitStatusEntry(
-                index_status=status[0],
-                worktree_status=status[1],
-                path=path,
-                original_path=original_path,
-            )
-        )
-    return tuple(entries)
-
-
-def _dirty_snapshot_hash(repo_root: Path, entries: tuple[GitStatusEntry, ...]) -> str | None:
-    if not entries:
-        return None
-    digest = hashlib.sha256()
-    for entry in sorted(entries, key=lambda e: (e.path, e.original_path or "")):
-        digest.update(entry.index_status.encode())
-        digest.update(entry.worktree_status.encode())
-        digest.update(entry.path.encode())
-        if entry.original_path is not None:
-            digest.update(entry.original_path.encode())
-        path = repo_root / entry.path
-        if path.is_file():
-            digest.update(hashlib.sha256(path.read_bytes()).digest())
-    return digest.hexdigest()[:24]
-
-
-def _resolve_base_ref(repo_root: Path, requested: str | None) -> str:
-    candidates = [requested] if requested else ["origin/main", "main", "master", "HEAD"]
-    for candidate in candidates:
-        if candidate is None:
-            continue
-        try:
-            _git(repo_root, "rev-parse", "--verify", candidate)
-            return candidate
-        except GitContextError:
-            continue
-    raise GitContextError("No usable base ref found")
-
-
-def resolve_worktree_context(
-    cwd: str | os.PathLike[str] | Path,
-    *,
-    base_ref: str | None = None,
-    index_config_hash: str,
-) -> WorktreeContext:
-    """Resolve Git identity and dirty state for *cwd*."""
-    start = Path(cwd).resolve()
-    repo_root = Path(_git(start, "rev-parse", "--show-toplevel")).resolve()
-    git_common_raw = _git(repo_root, "rev-parse", "--git-common-dir")
-    git_common_dir = Path(git_common_raw)
-    if not git_common_dir.is_absolute():
-        git_common_dir = (repo_root / git_common_dir).resolve()
-    remote_url = _git(repo_root, "remote", "get-url", "origin")
-    normalized_remote = normalize_remote_url(remote_url)
-    branch_name = _git(repo_root, "branch", "--show-current") or "HEAD"
-    head_commit = _git(repo_root, "rev-parse", "HEAD")
-    resolved_base_ref = _resolve_base_ref(repo_root, base_ref)
-    base_commit = _git(repo_root, "rev-parse", resolved_base_ref)
-    merge_base = _git(repo_root, "merge-base", "HEAD", resolved_base_ref)
-    status_entries = _parse_status(_git_bytes(repo_root, "status", "--porcelain=v1", "-z"))
-    dirty_hash = _dirty_snapshot_hash(repo_root, status_entries)
-    # The current project model indexes the Git worktree root.  Keep this as a
-    # stable logical root instead of deriving it from the physical worktree
-    # path, otherwise linked worktrees would get different repo IDs.
-    repo_relative_root = "."
-    repo_seed = f"{normalized_remote}\0{repo_relative_root}\0{index_config_hash}"
-    worktree_seed = str(repo_root)
-    return WorktreeContext(
-        repo_id=hashlib.sha256(repo_seed.encode()).hexdigest()[:24],
-        worktree_id=hashlib.sha256(worktree_seed.encode()).hexdigest()[:24],
-        repo_root=repo_root,
-        git_common_dir=git_common_dir,
-        remote_url=remote_url,
-        normalized_remote_url=normalized_remote,
-        branch_name=branch_name,
-        head_commit=head_commit,
-        base_ref=resolved_base_ref,
-        base_commit=base_commit,
-        merge_base=merge_base,
-        dirty_snapshot_hash=dirty_hash,
-        status_entries=status_entries,
-    )
+from .version_control import GitContextError, GitStatusEntry, normalize_remote_url
+from .version_control import Worktree as WorktreeContext
+from .version_control import resolve_worktree as resolve_worktree_context
+
+__all__ = [
+    "GitContextError",
+    "GitStatusEntry",
+    "WorktreeContext",
+    "normalize_remote_url",
+    "resolve_worktree_context",
+]