REPOMIND v0.1 — repo-scale coding agent demo

Open-source coding agent that ingests an entire git repository and reasons
across the whole codebase with multi-step tool use. Built for the AMD
Developer Hackathon 2026.

This Space ships:
- Ingestion pipeline (clone + tree-sitter + smart chunker + priority budget)
- 5 tools: read_file, grep_codebase, execute_code, run_tests, git_log
- SC-TIR-style agent loop adapted from AIMO3 math reasoning pipeline
- Mock LLM backend for offline demo (default)
- vLLM client ready to wire to MI300X (Qwen3-Coder-Next-FP8, 256K context)

Full source: https://github.com/SRKRZ23/repomind

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+.repomind_cache/
+.benchmarks/
+.pytest_cache/
+.DS_Store

agent/loop.py

@@ -0,0 +1,148 @@
+"""SC-TIR-style agent loop adapted from AIMO3 (math) to coding.
+
+Loop:
+    user → LLM → (tool calls?) → tools → LLM → ... → final answer
+
+Stops when:
+  - LLM emits content with no tool calls, OR
+  - max_steps hit (forces a final response without tools)
+
+The pattern mirrors Sardor's AIMO3 SC-TIR pipeline: the model alternates
+between thinking and tool-augmented action, with deterministic verification
+on the tool side.
+"""
+from __future__ import annotations
+import json
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+from serving.base import LLMClient, LLMResponse, ToolCall
+from tools.base import ToolRegistry, ToolResult
+from agent.prompts import SYSTEM, build_repo_overview, initial_user_prompt
+
+
+@dataclass
+class AgentTurn:
+    role: str
+    content: str
+    tool_calls: List[Dict[str, Any]] = field(default_factory=list)
+    tool_call_id: str | None = None
+
+
+@dataclass
+class AgentRun:
+    answer: str
+    transcript: List[Dict[str, Any]]
+    tool_calls: List[Dict[str, Any]]
+    steps: int
+    finished: bool
+
+
+class Agent:
+    def __init__(
+        self,
+        llm: LLMClient,
+        tools: ToolRegistry,
+        max_steps: int = 6,
+        max_tool_output_chars: int = 6000,
+    ):
+        self.llm = llm
+        self.tools = tools
+        self.max_steps = max_steps
+        self.max_tool_output_chars = max_tool_output_chars
+
+    def run(self, question: str, repo_summary: Dict[str, Any]) -> AgentRun:
+        overview = build_repo_overview(
+            repo=repo_summary.get("repo", ""),
+            n_files=repo_summary.get("n_files", 0),
+            n_chunks=repo_summary.get("n_chunks", 0),
+            total_tokens=repo_summary.get("total_tokens", 0),
+            top_paths=_pick_top_paths(repo_summary),
+        )
+        messages: List[Dict[str, Any]] = [
+            {"role": "system", "content": SYSTEM},
+            {"role": "user", "content": initial_user_prompt(question, overview)},
+        ]
+        tool_schema = self.tools.schema()
+        tool_calls_log: List[Dict[str, Any]] = []
+        step = 0
+        finished = False
+
+        while step < self.max_steps:
+            resp = self.llm.complete(messages, tools=tool_schema)
+            assistant_msg: Dict[str, Any] = {"role": "assistant"}
+            if resp.content:
+                assistant_msg["content"] = resp.content
+            if resp.tool_calls:
+                assistant_msg["tool_calls"] = [self._tool_call_to_msg(tc) for tc in resp.tool_calls]
+            else:
+                assistant_msg.setdefault("content", "")
+            messages.append(assistant_msg)
+
+            if not resp.tool_calls:
+                finished = True
+                break
+
+            for tc in resp.tool_calls:
+                tool_calls_log.append({"name": tc.name, "arguments": tc.arguments})
+                result = self.tools.call(tc.name, tc.arguments)
+                tool_msg = {
+                    "role": "tool",
+                    "tool_call_id": tc.id,
+                    "name": tc.name,
+                    "content": self._format_tool_output(result),
+                }
+                messages.append(tool_msg)
+            step += 1
+
+        # If we hit max_steps without a final answer, force one more text-only call.
+        if not finished:
+            messages.append({
+                "role": "user",
+                "content": "You've used the tool budget. Provide your best final answer now, without tool calls.",
+            })
+            resp = self.llm.complete(messages, tools=[])
+            messages.append({"role": "assistant", "content": resp.content or ""})
+
+        # Final answer = last assistant message with content
+        answer = ""
+        for m in reversed(messages):
+            if m.get("role") == "assistant" and m.get("content"):
+                answer = m["content"]
+                break
+
+        return AgentRun(
+            answer=answer,
+            transcript=messages,
+            tool_calls=tool_calls_log,
+            steps=step,
+            finished=finished,
+        )
+
+    def _tool_call_to_msg(self, tc: ToolCall) -> Dict[str, Any]:
+        return {
+            "id": tc.id,
+            "type": "function",
+            "function": {"name": tc.name, "arguments": json.dumps(tc.arguments)},
+        }
+
+    def _format_tool_output(self, result: ToolResult) -> str:
+        body = result.output if result.ok else f"[error] {result.error}"
+        if len(body) > self.max_tool_output_chars:
+            body = body[: self.max_tool_output_chars] + "\n[... truncated]"
+        return body
+
+
+def _pick_top_paths(summary: Dict[str, Any]) -> List[str]:
+    chunks = summary.get("chunks") or []
+    seen: List[str] = []
+    seen_set = set()
+    # priority 0 first, then 1; keep insertion order
+    for prio in (0, 1, 2):
+        for c in chunks:
+            if c.get("priority") == prio and c.get("path") not in seen_set:
+                seen.append(c["path"])
+                seen_set.add(c["path"])
+                if len(seen) >= 60:
+                    return seen
+    return seen

agent/prompts.py

@@ -0,0 +1,49 @@
+"""System and tool prompt templates for the REPOMIND agent."""
+from __future__ import annotations
+from typing import Iterable
+
+
+SYSTEM = """You are REPOMIND, a coding agent that has the ENTIRE git repository
+loaded in your context window. You have these capabilities:
+
+1. You can see the structure of the repo (tree of files + symbols).
+2. You can call tools to read files, grep, run code, run tests, or inspect git history.
+3. You must reason carefully across multiple files when needed.
+
+## How to answer
+
+- Be precise. When you cite code, give the file path and line numbers.
+- When you don't know, call a tool to find out — never invent.
+- Stop calling tools as soon as you have enough information to answer.
+- After tool calls, respond in plain prose with concrete references.
+
+## Tool-call protocol
+
+Use the standard OpenAI function-calling format. Each tool result will be
+delivered back as a tool-role message; you may then either call another
+tool or write the final answer.
+"""
+
+
+def build_repo_overview(repo: str, n_files: int, n_chunks: int, total_tokens: int, top_paths: Iterable[str]) -> str:
+    paths = "\n".join(f"  - {p}" for p in list(top_paths)[:40])
+    return f"""# Repo: {repo}
+
+- Files indexed: {n_files}
+- Chunks: {n_chunks}
+- Total tokens: {total_tokens:,}
+
+Top-priority files:
+{paths or "  (none)"}
+"""
+
+
+def initial_user_prompt(question: str, overview: str) -> str:
+    return f"""{overview}
+
+# Question
+
+{question}
+
+Answer with concrete code references.
+"""

app.py

@@ -0,0 +1,175 @@
+"""REPOMIND — HuggingFace Space entry point.
+
+Public demo. Backend defaults to the offline mock LLM so the Space runs
+without GPU credits. Once the AMD MI300X vLLM endpoint is live, switch
+the backend toggle to "vllm" and point at the live URL.
+
+Local repo: https://github.com/SRKRZ23/repomind
+Hackathon:  https://lablab.ai/ai-hackathons/amd-developer
+"""
+from __future__ import annotations
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+# make submodules importable
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import gradio as gr
+
+from ingestion.chunker import ingest_to_json
+from ingestion.cloner import clone
+
+
+HEADER_MD = """
+# REPOMIND
+**Open-source repo-scale coding agent on AMD MI300X.**
+
+Ingest a git repository (up to 256K tokens, FP8) on a single GPU and reason across the whole codebase with multi-step tool use.
+
+> 📦 GitHub: [SRKRZ23/repomind](https://github.com/SRKRZ23/repomind)
+> 🏆 Built for the [AMD Developer Hackathon 2026](https://lablab.ai/ai-hackathons/amd-developer)
+
+### Why MI300X?
+- Qwen3-Coder-Next-FP8 weights ≈ 80 GB
+- 256K KV cache @ FP8 ≈ 38 GB
+- + activations ≈ 25 GB → **~143 GB total on a single GPU**
+- NVIDIA H100 80GB physically OOMs. AMD MI300X 192GB just runs it.
+
+### About this Space
+This is the **frontend** demo. Backend defaults to the **mock LLM** so the Space
+runs on CPU-basic without burning GPU credits. Switch to `vllm` and provide a
+base URL once the MI300X endpoint is live.
+"""
+
+
+# Minimal cap — HF Space CPU-basic gets 16 GB RAM. Don't blow it on giant repos.
+MAX_INGEST_SIZE_MB = 50
+SCRATCH_DIR = Path(tempfile.gettempdir()) / "repomind_hf"
+SCRATCH_DIR.mkdir(exist_ok=True)
+
+
+def ingest(url_or_path: str, chunk_tokens: int) -> str:
+    if not url_or_path or not url_or_path.strip():
+        return "Provide a GitHub URL or `owner/repo` shorthand."
+    out = SCRATCH_DIR / "active.json"
+    try:
+        # Local path mode (rare on HF — usually URL)
+        if Path(url_or_path).is_dir():
+            repo_root = Path(url_or_path)
+            label = repo_root.name
+        else:
+            res = clone(url_or_path, cache_dir=SCRATCH_DIR / "repos")
+            repo_root = res.local_path
+            label = res.url.rsplit("/", 1)[-1].removesuffix(".git")
+        summary = ingest_to_json(
+            repo_root,
+            out,
+            repo_label=label,
+            max_tokens_per_chunk=chunk_tokens,
+        )
+        return json.dumps(summary, indent=2)
+    except Exception as e:
+        return f"❌ {type(e).__name__}: {e}"
+
+
+def ask(question: str, backend: str, base_url: str, model: str):
+    summary_path = SCRATCH_DIR / "active.json"
+    if not summary_path.exists():
+        return "Ingest a repo first.", ""
+    if not question or not question.strip():
+        return "Type a question.", ""
+
+    summary = json.loads(summary_path.read_text())
+    repo_root = Path(summary.get("root", "."))
+
+    # Backend wiring — vLLM only when user explicitly chose it AND a URL is given
+    if backend == "vllm":
+        if not base_url or not base_url.strip():
+            return "vLLM backend selected but no base URL provided.", ""
+        try:
+            from serving.vllm_client import VLLMClient
+            llm = VLLMClient(base_url=base_url.strip(), model=model.strip() or "Qwen/Qwen3-Coder-Next-FP8")
+        except Exception as e:
+            return f"❌ failed to init vLLM client: {e}", ""
+    else:
+        from serving.mock_client import MockClient
+        llm = MockClient(max_tool_turns=2)
+
+    from agent.loop import Agent
+    from tools.registry import default_registry
+
+    try:
+        agent = Agent(llm=llm, tools=default_registry(repo_root, scratch_dir=SCRATCH_DIR / "scratch"), max_steps=4)
+        result = agent.run(question, summary)
+    except Exception as e:
+        return f"❌ agent failed: {type(e).__name__}: {e}", ""
+
+    trace_lines = [f"- {tc['name']} {json.dumps(tc['arguments'], ensure_ascii=False)}" for tc in result.tool_calls]
+    trace = "\n".join(trace_lines) or "(no tool calls)"
+    return result.answer, trace
+
+
+with gr.Blocks(title="REPOMIND — repo-scale coding agent on AMD MI300X", theme=gr.themes.Soft()) as demo:
+    gr.Markdown(HEADER_MD)
+
+    with gr.Tab("1. Ingest"):
+        with gr.Row():
+            url = gr.Textbox(
+                label="GitHub URL or owner/repo",
+                placeholder="https://github.com/torvalds/linux  OR  pallets/flask",
+                scale=4,
+            )
+            chunk_tokens = gr.Slider(256, 4096, value=1024, step=128, label="Tokens / chunk", scale=1)
+        ingest_btn = gr.Button("Ingest", variant="primary")
+        ingest_out = gr.Code(label="Ingestion summary", language="json")
+        ingest_btn.click(ingest, [url, chunk_tokens], ingest_out)
+
+    with gr.Tab("2. Ask"):
+        with gr.Row():
+            backend = gr.Radio(
+                choices=["mock (offline demo)", "vllm (live MI300X)"],
+                value="mock (offline demo)",
+                label="Backend",
+                scale=1,
+            )
+            base_url = gr.Textbox(
+                label="vLLM base URL (only used in `vllm` mode)",
+                value="",
+                placeholder="http://your-mi300x-host:8000/v1",
+                scale=2,
+            )
+            model = gr.Textbox(
+                label="Model id",
+                value="Qwen/Qwen3-Coder-Next-FP8",
+                scale=2,
+            )
+        question = gr.Textbox(
+            label="Question",
+            lines=3,
+            placeholder="What does the chunker prioritize? Where is authentication handled?",
+        )
+        ask_btn = gr.Button("Ask", variant="primary")
+        answer = gr.Markdown(label="Answer")
+        tool_trace = gr.Code(label="Tool trace", language="markdown")
+
+        # normalize backend selector to internal value
+        def _ask(q, b, u, m):
+            internal = "vllm" if b.startswith("vllm") else "mock"
+            return ask(q, internal, u, m)
+
+        ask_btn.click(_ask, [question, backend, base_url, model], [answer, tool_trace])
+
+    gr.Markdown(
+        "---\n"
+        "**Author:** [Sardor Razikov](https://huggingface.co/ZeroR3) · "
+        "[GitHub](https://github.com/SRKRZ23) · "
+        "[lablab.ai](https://lablab.ai/u/@Sardor_R) · "
+        "[Zenodo (ECB)](https://doi.org/10.5281/zenodo.19791329)"
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()

ingestion/chunker.py

@@ -0,0 +1,202 @@
+"""Smart, structure-aware chunking with priority scoring.
+
+Per file:
+  1. Detect language (filesystem extension).
+  2. Extract top-level symbols via tree-sitter (or regex fallback).
+  3. Slice file into chunks aligned to symbol boundaries when possible;
+     otherwise split on paragraph / blank lines / hard cut.
+  4. Tag each chunk with a priority used by the token budgeter:
+        0 = README / top-level docs
+        1 = top-level symbols (functions, classes)
+        2 = nested / private symbols
+        3 = test / vendored / generated code
+        4 = unknown / binary-ish
+
+The agent only sees chunks that fit its context budget — priorities decide
+who gets in first when a 50K-LOC kernel doesn't fit at all.
+"""
+from __future__ import annotations
+import json
+import os
+import re
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Iterable, List, Optional, Sequence
+
+from .parser import Symbol, detect_language, extract_symbols
+from .token_budget import count_tokens
+
+
+SKIP_DIRS = {
+    ".git", "node_modules", ".venv", "venv", "env", "__pycache__",
+    "dist", "build", "target", ".next", ".nuxt", ".cache",
+    "vendor", "third_party", "external", ".gradle", ".idea", ".vscode",
+}
+SKIP_BIN_EXT = {
+    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".ico", ".tiff",
+    ".pdf", ".zip", ".tar", ".gz", ".bz2", ".7z", ".xz", ".whl", ".egg",
+    ".so", ".dylib", ".dll", ".exe", ".o", ".a", ".class", ".jar",
+    ".bin", ".pkl", ".parquet", ".safetensors", ".pt", ".onnx",
+    ".woff", ".woff2", ".ttf", ".otf", ".mp3", ".mp4", ".mov", ".wav",
+}
+README_NAMES = {"README.md", "README.rst", "README.txt", "README"}
+TEST_PATTERNS = (re.compile(r"(?:^|/)tests?/"), re.compile(r"(?:^|/)test_"), re.compile(r"_test\."))
+
+
+@dataclass
+class Chunk:
+    chunk_id: str
+    repo: str
+    path: str
+    section: str            # symbol name or "header"
+    start_line: int
+    end_line: int
+    text: str
+    n_tokens: int
+    priority: int
+
+
+def _is_test_path(rel: str) -> bool:
+    return any(p.search(rel) for p in TEST_PATTERNS)
+
+
+def _file_priority(rel: str, name: str) -> int:
+    if name in README_NAMES or rel.endswith(("README.md", "README.rst")):
+        return 0
+    if _is_test_path(rel):
+        return 3
+    if any(seg in rel.split("/") for seg in ("docs", "doc")):
+        return 0
+    return 1
+
+
+def _chunk_text_by_symbols(
+    text: str, symbols: List[Symbol], max_tokens: int, overlap_lines: int = 4,
+) -> List[tuple[str, str, int, int]]:
+    """Return [(section, text, start_line, end_line)]. Symbols are sorted by start_line."""
+    lines = text.split("\n")
+    n = len(lines)
+    if not symbols:
+        return _chunk_lines("body", lines, 1, n, max_tokens)
+
+    symbols = sorted(symbols, key=lambda s: s.start_line)
+    out: List[tuple[str, str, int, int]] = []
+
+    # Header / preamble before first symbol
+    if symbols[0].start_line > 1:
+        out.extend(_chunk_lines("header", lines, 1, symbols[0].start_line - 1, max_tokens))
+
+    for i, sym in enumerate(symbols):
+        end = symbols[i + 1].start_line - 1 if i + 1 < len(symbols) else n
+        if end < sym.start_line:
+            continue
+        out.extend(_chunk_lines(sym.name or sym.kind, lines, sym.start_line, end, max_tokens))
+    return out
+
+
+def _chunk_lines(section: str, lines: list[str], lo: int, hi: int, max_tokens: int):
+    """Split a slice of [lo..hi] (1-indexed inclusive) into <= max_tokens pieces."""
+    pieces: List[tuple[str, str, int, int]] = []
+    cur: List[str] = []
+    cur_tokens = 0
+    cur_start = lo
+    for idx in range(lo, hi + 1):
+        line = lines[idx - 1] if 0 < idx <= len(lines) else ""
+        line_tokens = count_tokens(line) + 1
+        if cur and cur_tokens + line_tokens > max_tokens:
+            pieces.append((section, "\n".join(cur), cur_start, idx - 1))
+            cur = [line]
+            cur_tokens = line_tokens
+            cur_start = idx
+        else:
+            cur.append(line)
+            cur_tokens += line_tokens
+    if cur:
+        pieces.append((section, "\n".join(cur), cur_start, hi))
+    return pieces
+
+
+def chunk_file(
+    repo: str,
+    path: Path,
+    rel_path: str,
+    max_tokens_per_chunk: int = 1024,
+) -> List[Chunk]:
+    name = path.name
+    if path.suffix.lower() in SKIP_BIN_EXT:
+        return []
+    try:
+        text = path.read_text(encoding="utf-8")
+    except (UnicodeDecodeError, OSError):
+        return []
+    if not text.strip():
+        return []
+
+    lang = detect_language(path)
+    symbols = extract_symbols(text, lang)
+    base_priority = _file_priority(rel_path, name)
+    pieces = _chunk_text_by_symbols(text, symbols, max_tokens_per_chunk)
+
+    chunks: List[Chunk] = []
+    for i, (section, ctext, start, end) in enumerate(pieces):
+        # Nested / very small private fragments get bumped down a tier.
+        prio = base_priority
+        if base_priority == 1 and section.startswith("_"):
+            prio = 2
+        chunks.append(Chunk(
+            chunk_id=f"{rel_path}#{i}",
+            repo=repo,
+            path=rel_path,
+            section=section,
+            start_line=start,
+            end_line=end,
+            text=ctext,
+            n_tokens=count_tokens(ctext),
+            priority=prio,
+        ))
+    return chunks
+
+
+def walk_repo(
+    root: str | Path,
+    repo_label: str,
+    max_tokens_per_chunk: int = 1024,
+    follow_symlinks: bool = False,
+) -> Iterable[Chunk]:
+    root = Path(root).resolve()
+    for dirpath, dirnames, filenames in os.walk(root, followlinks=follow_symlinks):
+        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
+        for fn in filenames:
+            full = Path(dirpath) / fn
+            try:
+                rel = str(full.relative_to(root))
+            except ValueError:
+                continue
+            yield from chunk_file(repo_label, full, rel, max_tokens_per_chunk)
+
+
+def ingest_to_json(
+    root: str | Path,
+    out_path: str | Path,
+    repo_label: Optional[str] = None,
+    max_tokens_per_chunk: int = 1024,
+) -> dict:
+    root = Path(root).resolve()
+    label = repo_label or root.name
+    chunks = list(walk_repo(root, label, max_tokens_per_chunk))
+    summary = {
+        "repo": label,
+        "root": str(root),
+        "n_files": len({c.path for c in chunks}),
+        "n_chunks": len(chunks),
+        "total_tokens": sum(c.n_tokens for c in chunks),
+        "by_priority": {
+            str(p): sum(1 for c in chunks if c.priority == p)
+            for p in sorted({c.priority for c in chunks})
+        },
+        "chunks": [asdict(c) for c in chunks],
+    }
+    out = Path(out_path)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(json.dumps(summary, ensure_ascii=False))
+    return {k: v for k, v in summary.items() if k != "chunks"}

ingestion/cloner.py

@@ -0,0 +1,100 @@
+"""Clone a git repository into a local cache directory.
+
+Uses GitPython if installed, falls back to shelling out to `git`. Always
+shallow-clones (depth=1) by default — for retrieval we don't need history,
+and shallow makes the Linux kernel ingest in seconds instead of minutes.
+"""
+from __future__ import annotations
+import os
+import re
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+
+URL_RE = re.compile(r"^(https?://|git@)([\w./:-]+?)(\.git)?/?$")
+
+
+@dataclass
+class CloneResult:
+    url: str
+    local_path: Path
+    sha: str
+    cached: bool
+
+
+def normalize_url(url_or_path: str) -> str:
+    """Accept https://, git@, owner/repo, or a local path."""
+    s = url_or_path.strip()
+    if os.path.isdir(s):
+        return os.path.abspath(s)
+    if s.startswith("git@") or s.startswith("http"):
+        return s
+    if "/" in s and not s.startswith("/"):
+        # owner/repo shorthand -> github
+        return f"https://github.com/{s}.git"
+    return s
+
+
+def slugify(url: str) -> str:
+    """Stable filesystem-friendly slug from a URL."""
+    if os.path.isdir(url):
+        return Path(url).name
+    m = URL_RE.match(url)
+    if not m:
+        return re.sub(r"[^a-zA-Z0-9._-]+", "_", url)
+    body = m.group(2)
+    return re.sub(r"[^a-zA-Z0-9._-]+", "_", body)
+
+
+def _git(*args: str, cwd: Optional[Path] = None) -> str:
+    proc = subprocess.run(
+        ["git", *args],
+        cwd=str(cwd) if cwd else None,
+        capture_output=True, text=True,
+    )
+    if proc.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")
+    return proc.stdout.strip()
+
+
+def clone(
+    url_or_path: str,
+    cache_dir: str | Path = ".repomind_cache/repos",
+    depth: int = 1,
+    force: bool = False,
+) -> CloneResult:
+    """Clone url to cache_dir/<slug>. If local path is given, just return it."""
+    url = normalize_url(url_or_path)
+    cache_dir = Path(cache_dir)
+    cache_dir.mkdir(parents=True, exist_ok=True)
+
+    # Local-path mode — no clone, just record the SHA if it's a git repo.
+    if os.path.isdir(url):
+        local = Path(url)
+        try:
+            sha = _git("rev-parse", "HEAD", cwd=local)
+        except Exception:
+            sha = "no-git"
+        return CloneResult(url=str(local), local_path=local, sha=sha, cached=True)
+
+    target = cache_dir / slugify(url)
+    if target.exists() and not force:
+        try:
+            sha = _git("rev-parse", "HEAD", cwd=target)
+            return CloneResult(url=url, local_path=target, sha=sha, cached=True)
+        except Exception:
+            shutil.rmtree(target, ignore_errors=True)
+
+    if target.exists() and force:
+        shutil.rmtree(target, ignore_errors=True)
+
+    args = ["clone", "--filter=blob:none"]
+    if depth > 0:
+        args += ["--depth", str(depth)]
+    args += [url, str(target)]
+    _git(*args)
+    sha = _git("rev-parse", "HEAD", cwd=target)
+    return CloneResult(url=url, local_path=target, sha=sha, cached=False)

ingestion/parser.py

@@ -0,0 +1,161 @@
+"""Tree-sitter-aware parsing with graceful fallback.
+
+If `tree-sitter-languages` isn't installed we degrade to a regex-based
+top-level-symbol extractor — good enough for unit tests and for
+languages we don't yet have grammars for.
+"""
+from __future__ import annotations
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, List, Optional
+
+
+# Map filesystem extensions to tree-sitter grammar names.
+LANG_BY_EXT = {
+    ".py": "python", ".pyi": "python",
+    ".rs": "rust",
+    ".go": "go",
+    ".js": "javascript", ".jsx": "javascript",
+    ".ts": "typescript", ".tsx": "tsx",
+    ".c": "c", ".h": "c",
+    ".cc": "cpp", ".cpp": "cpp", ".cxx": "cpp", ".hpp": "cpp",
+    ".java": "java",
+    ".rb": "ruby",
+    ".php": "php",
+    ".cs": "c_sharp",
+    ".swift": "swift",
+    ".kt": "kotlin", ".kts": "kotlin",
+    ".sh": "bash", ".bash": "bash",
+    ".sql": "sql",
+    ".html": "html",
+    ".css": "css",
+    ".json": "json",
+    ".yaml": "yaml", ".yml": "yaml",
+    ".toml": "toml",
+    ".md": "markdown", ".markdown": "markdown",
+}
+
+
+@dataclass
+class Symbol:
+    name: str
+    kind: str          # function / class / method / module / heading
+    start_line: int
+    end_line: int
+
+
+def detect_language(path: str | Path) -> Optional[str]:
+    suffix = Path(path).suffix.lower()
+    return LANG_BY_EXT.get(suffix)
+
+
+def extract_symbols(text: str, language: Optional[str]) -> List[Symbol]:
+    """Top-level structural symbols. Tree-sitter when available; regex fallback."""
+    if not text.strip():
+        return []
+    try:
+        return _ts_symbols(text, language)
+    except Exception:
+        return _regex_symbols(text, language)
+
+
+# ---- tree-sitter path -----------------------------------------------------
+
+def _ts_symbols(text: str, language: Optional[str]) -> List[Symbol]:
+    if not language:
+        raise RuntimeError("no language")
+    try:
+        from tree_sitter_languages import get_parser  # type: ignore
+    except ImportError as e:
+        raise RuntimeError("tree_sitter_languages not installed") from e
+
+    parser = get_parser(language)
+    tree = parser.parse(text.encode("utf-8"))
+    out: List[Symbol] = []
+
+    interesting = {
+        "function_definition", "function_declaration", "method_definition",
+        "class_definition", "class_declaration", "struct_item", "trait_item",
+        "impl_item", "enum_item", "type_alias_declaration",
+        "atx_heading", "setext_heading",
+    }
+
+    def walk(node):
+        if node.type in interesting:
+            name = _node_name(node, text) or node.type
+            kind = _kind_for(node.type)
+            out.append(Symbol(
+                name=name, kind=kind,
+                start_line=node.start_point[0] + 1,
+                end_line=node.end_point[0] + 1,
+            ))
+        for c in node.children:
+            walk(c)
+
+    walk(tree.root_node)
+    return out
+
+
+def _node_name(node, text: str) -> Optional[str]:
+    for c in node.children:
+        if c.type == "identifier" or c.type == "type_identifier":
+            return text[c.start_byte:c.end_byte]
+        for cc in c.children:
+            if cc.type in ("identifier", "type_identifier"):
+                return text[cc.start_byte:cc.end_byte]
+    if node.type in ("atx_heading", "setext_heading"):
+        return text[node.start_byte:node.end_byte].strip().lstrip("#").strip()
+    return None
+
+
+def _kind_for(node_type: str) -> str:
+    if "class" in node_type or "struct" in node_type or "trait" in node_type or "impl" in node_type:
+        return "class"
+    if "method" in node_type:
+        return "method"
+    if "function" in node_type:
+        return "function"
+    if "heading" in node_type:
+        return "heading"
+    return "symbol"
+
+
+# ---- regex fallback -------------------------------------------------------
+
+PY_DEF = re.compile(r"^(?P<indent>\s*)(?:async\s+)?def\s+(?P<name>[A-Za-z_][\w]*)\s*\(", re.MULTILINE)
+PY_CLASS = re.compile(r"^(?P<indent>\s*)class\s+(?P<name>[A-Za-z_][\w]*)", re.MULTILINE)
+RS_FN = re.compile(r"^\s*(?:pub(?:\([^)]*\))?\s+)?fn\s+(?P<name>[A-Za-z_][\w]*)", re.MULTILINE)
+GO_FN = re.compile(r"^\s*func\s+(?:\([^)]*\)\s+)?(?P<name>[A-Za-z_][\w]*)", re.MULTILINE)
+JS_FN = re.compile(r"^\s*(?:export\s+)?(?:async\s+)?function\s+(?P<name>[A-Za-z_$][\w$]*)", re.MULTILINE)
+MD_HEADING = re.compile(r"^(#{1,6})\s+(?P<name>.+)$", re.MULTILINE)
+
+
+def _regex_symbols(text: str, language: Optional[str]) -> List[Symbol]:
+    lines = text.split("\n")
+    out: List[Symbol] = []
+
+    def add(name: str, kind: str, m: re.Match):
+        line = text[:m.start()].count("\n") + 1
+        out.append(Symbol(name=name, kind=kind, start_line=line, end_line=line))
+
+    if language == "python":
+        for m in PY_CLASS.finditer(text):
+            add(m.group("name"), "class", m)
+        for m in PY_DEF.finditer(text):
+            indent = m.group("indent")
+            kind = "method" if indent else "function"
+            add(m.group("name"), kind, m)
+    elif language == "rust":
+        for m in RS_FN.finditer(text):
+            add(m.group("name"), "function", m)
+    elif language == "go":
+        for m in GO_FN.finditer(text):
+            add(m.group("name"), "function", m)
+    elif language in ("javascript", "typescript", "tsx"):
+        for m in JS_FN.finditer(text):
+            add(m.group("name"), "function", m)
+    elif language == "markdown":
+        for m in MD_HEADING.finditer(text):
+            add(m.group("name").strip(), "heading", m)
+    return out

ingestion/token_budget.py

@@ -0,0 +1,71 @@
+"""Token-aware operations: counting + priority-based truncation.
+
+Uses tiktoken when available (cl100k_base encoder approximates Qwen tokens
+within ~3 % on natural code/prose). Falls back to a 3.6 chars-per-token
+estimator otherwise.
+"""
+from __future__ import annotations
+from typing import Iterable, List, Sequence
+
+
+_ENC = None
+
+
+def _get_encoder():
+    global _ENC
+    if _ENC is not None:
+        return _ENC
+    try:
+        import tiktoken
+        _ENC = tiktoken.get_encoding("cl100k_base")
+    except Exception:
+        _ENC = False
+    return _ENC
+
+
+def count_tokens(text: str) -> int:
+    """Best-effort token count for the configured encoder."""
+    if not text:
+        return 0
+    enc = _get_encoder()
+    if enc:
+        return len(enc.encode(text, disallowed_special=()))
+    # Heuristic: ~3.6 chars/token on mixed code+prose
+    return max(1, int(len(text) / 3.6))
+
+
+def truncate_to(text: str, max_tokens: int) -> str:
+    """Drop trailing tokens to fit a budget."""
+    if max_tokens <= 0:
+        return ""
+    enc = _get_encoder()
+    if enc:
+        toks = enc.encode(text, disallowed_special=())
+        if len(toks) <= max_tokens:
+            return text
+        return enc.decode(toks[:max_tokens])
+    # heuristic
+    chars = int(max_tokens * 3.6)
+    return text[:chars]
+
+
+def fit_priority(
+    items: Sequence[tuple[str, int]],   # (text, priority — lower = include first)
+    max_tokens: int,
+) -> str:
+    """Pack texts in priority order until budget exhausted; truncate the last fitting one."""
+    out: List[str] = []
+    used = 0
+    sorted_items = sorted(items, key=lambda t: t[1])
+    for text, _prio in sorted_items:
+        n = count_tokens(text)
+        if used + n <= max_tokens:
+            out.append(text)
+            used += n
+        else:
+            remaining = max_tokens - used
+            if remaining > 32:
+                out.append(truncate_to(text, remaining))
+                used = max_tokens
+            break
+    return "\n\n".join(out)

requirements.txt

@@ -0,0 +1,7 @@
+gradio>=4.40
+gitpython>=3.1
+tree-sitter>=0.22
+tree-sitter-languages>=1.10
+tiktoken>=0.7
+pydantic>=2.6
+openai>=1.40

serving/base.py

@@ -0,0 +1,28 @@
+"""LLM client protocol used by the agent loop."""
+from __future__ import annotations
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Protocol
+
+
+@dataclass
+class ToolCall:
+    id: str
+    name: str
+    arguments: Dict[str, Any]
+
+
+@dataclass
+class LLMResponse:
+    content: str
+    tool_calls: List[ToolCall] = field(default_factory=list)
+    usage: Dict[str, int] = field(default_factory=dict)
+    raw: Any = None
+
+
+class LLMClient(Protocol):
+    def complete(
+        self,
+        messages: List[Dict[str, Any]],
+        tools: List[Dict[str, Any]],
+        **kwargs: Any,
+    ) -> LLMResponse: ...

serving/mock_client.py

@@ -0,0 +1,73 @@
+"""Deterministic offline LLM stub. Drives unit tests without a GPU.
+
+Each call inspects the latest user / tool message and decides what to do
+based on simple heuristics:
+  - Question contains 'grep' or 'search' → emit grep_codebase tool call
+  - Question contains 'read' or 'show' → emit read_file tool call
+  - After two tool turns → emit a final answer
+"""
+from __future__ import annotations
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List
+
+from .base import LLMClient, LLMResponse, ToolCall
+
+
+@dataclass
+class MockClient(LLMClient):
+    max_tool_turns: int = 2
+
+    def complete(self, messages: List[Dict[str, Any]], tools: List[Dict[str, Any]], **kwargs: Any) -> LLMResponse:
+        tool_turns = sum(1 for m in messages if m.get("role") == "tool")
+        last_user = next((m for m in reversed(messages) if m.get("role") == "user"), None)
+        question = (last_user.get("content") if last_user else "") or ""
+
+        if tool_turns >= self.max_tool_turns:
+            tool_outputs = [m.get("content", "") for m in messages if m.get("role") == "tool"]
+            joined = "\n".join(tool_outputs)[:1500]
+            answer = (
+                "Based on the inspected files, here is what I found:\n\n"
+                f"{joined or '(no tool output)'}\n\n"
+                f"Original question: {question.strip()}"
+            )
+            return LLMResponse(content=answer, tool_calls=[], usage={"prompt": 0, "completion": 0})
+
+        # Decide which tool to call next
+        q = question.lower()
+        if any(k in q for k in ("grep", "search", "find", "where", "occurr")):
+            call = ToolCall(
+                id=f"call_{tool_turns}", name="grep_codebase",
+                arguments={"pattern": _extract_term(question), "max_results": 10},
+            )
+        elif any(k in q for k in ("git", "history", "commits", "log")):
+            call = ToolCall(id=f"call_{tool_turns}", name="git_log", arguments={"limit": 10})
+        elif any(k in q for k in ("test", "run", "pytest")):
+            call = ToolCall(id=f"call_{tool_turns}", name="run_tests", arguments={})
+        elif any(k in q for k in ("read", "show", "open", "file")):
+            call = ToolCall(
+                id=f"call_{tool_turns}", name="read_file",
+                arguments={"path": _extract_path(question) or "README.md"},
+            )
+        else:
+            # default to a grep so we always exercise tool path
+            call = ToolCall(
+                id=f"call_{tool_turns}", name="grep_codebase",
+                arguments={"pattern": _extract_term(question), "max_results": 5},
+            )
+        return LLMResponse(content="", tool_calls=[call], usage={"prompt": 0, "completion": 0})
+
+
+def _extract_term(text: str) -> str:
+    # crude term extraction for tests: take first identifier-like token longer than 3 chars
+    import re
+    for m in re.finditer(r"[A-Za-z_][\w]{3,}", text):
+        if m.group(0).lower() not in {"what", "where", "find", "grep", "search", "show", "file", "read"}:
+            return m.group(0)
+    return text.strip()[:32] or "."
+
+
+def _extract_path(text: str) -> str:
+    import re
+    m = re.search(r"[\w./-]+\.[A-Za-z]{1,5}", text)
+    return m.group(0) if m else ""

serving/vllm_client.py

@@ -0,0 +1,69 @@
+"""vLLM-backed client (OpenAI-compatible API).
+
+Targets a local vLLM server running:
+    vllm serve Qwen/Qwen3-Coder-Next-FP8 \\
+        --tool-call-parser qwen3_coder \\
+        --max-model-len 262144 \\
+        --kv-cache-dtype fp8
+
+The server returns tool-calls in the OpenAI function-calling format, which
+we translate to our internal ToolCall dataclass.
+"""
+from __future__ import annotations
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, List
+
+from .base import LLMClient, LLMResponse, ToolCall
+
+
+@dataclass
+class VLLMClient(LLMClient):
+    base_url: str = "http://localhost:8000/v1"
+    model: str = "Qwen/Qwen3-Coder-Next-FP8"
+    api_key: str = "EMPTY"          # vLLM ignores the key but the SDK requires one
+    timeout: float = 300.0
+    temperature: float = 0.2
+    max_tokens: int = 2048
+
+    def __post_init__(self):
+        try:
+            from openai import OpenAI  # type: ignore
+        except ImportError as e:
+            raise ImportError("pip install openai") from e
+        self._client = OpenAI(base_url=self.base_url, api_key=self.api_key, timeout=self.timeout)
+
+    def complete(
+        self,
+        messages: List[Dict[str, Any]],
+        tools: List[Dict[str, Any]],
+        **kwargs: Any,
+    ) -> LLMResponse:
+        kw = {
+            "model": self.model,
+            "messages": messages,
+            "temperature": kwargs.get("temperature", self.temperature),
+            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
+        }
+        if tools:
+            kw["tools"] = tools
+            kw["tool_choice"] = kwargs.get("tool_choice", "auto")
+
+        resp = self._client.chat.completions.create(**kw)
+        choice = resp.choices[0].message
+        content = choice.content or ""
+        tool_calls: List[ToolCall] = []
+        for tc in (choice.tool_calls or []):
+            try:
+                args = json.loads(tc.function.arguments or "{}")
+            except json.JSONDecodeError:
+                args = {}
+            tool_calls.append(ToolCall(id=tc.id, name=tc.function.name, arguments=args))
+
+        usage = {}
+        if getattr(resp, "usage", None):
+            usage = {
+                "prompt": resp.usage.prompt_tokens,
+                "completion": resp.usage.completion_tokens,
+            }
+        return LLMResponse(content=content, tool_calls=tool_calls, usage=usage, raw=resp)

tools/base.py

@@ -0,0 +1,62 @@
+"""Tool-call protocol shared by every tool. Mirrors the qwen3_coder schema
+so vLLM's built-in tool parser can dispatch directly into our registry.
+"""
+from __future__ import annotations
+import json
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List
+
+
+@dataclass
+class ToolResult:
+    ok: bool
+    output: str
+    error: str = ""
+    extra: Dict[str, Any] = field(default_factory=dict)
+
+    def to_message(self) -> Dict[str, Any]:
+        body = self.output if self.ok else f"[error] {self.error}"
+        return {"role": "tool", "content": body}
+
+
+@dataclass
+class ToolSpec:
+    name: str
+    description: str
+    parameters: Dict[str, Any]
+    runner: Callable[..., ToolResult]
+
+
+class ToolRegistry:
+    def __init__(self):
+        self._tools: Dict[str, ToolSpec] = {}
+
+    def register(self, spec: ToolSpec):
+        self._tools[spec.name] = spec
+
+    def names(self) -> List[str]:
+        return list(self._tools)
+
+    def schema(self) -> List[Dict[str, Any]]:
+        """OpenAI / qwen3_coder tool schema (function-calling style)."""
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": s.name,
+                    "description": s.description,
+                    "parameters": s.parameters,
+                },
+            }
+            for s in self._tools.values()
+        ]
+
+    def call(self, name: str, args: Dict[str, Any]) -> ToolResult:
+        if name not in self._tools:
+            return ToolResult(ok=False, output="", error=f"unknown tool: {name}")
+        try:
+            return self._tools[name].runner(**args)
+        except TypeError as e:
+            return ToolResult(ok=False, output="", error=f"bad args: {e}")
+        except Exception as e:
+            return ToolResult(ok=False, output="", error=f"{type(e).__name__}: {e}")

tools/execute_code.py

@@ -0,0 +1,104 @@
+"""execute_code tool — sandboxed Python runner.
+
+Subprocess with -I -S, no network (best-effort via env), CPU + wall-clock
+limits, file IO restricted to a temp scratch dir. Sandboxing here is
+defense-in-depth, not airtight — REPOMIND's threat model assumes the
+operator trusts the model. The point is preventing accidental damage to
+the repo, not stopping a determined adversary.
+"""
+from __future__ import annotations
+import os
+import resource
+import subprocess
+import sys
+import tempfile
+import textwrap
+from pathlib import Path
+
+from .base import ToolResult, ToolSpec
+
+
+PREAMBLE = textwrap.dedent("""\
+    import sys, os, signal, resource
+    # disable network sockets at python level
+    try:
+        import socket
+        def _block(*_a, **_k):
+            raise RuntimeError("network disabled in sandbox")
+        socket.socket = _block            # type: ignore
+        socket.create_connection = _block # type: ignore
+    except Exception:
+        pass
+""")
+
+
+def _set_limits(cpu_seconds: int = 30, mem_mb: int = 1024):
+    try:
+        resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds + 1))
+    except (ValueError, OSError):
+        pass
+    try:
+        resource.setrlimit(resource.RLIMIT_AS, (mem_mb * 1024 * 1024, mem_mb * 1024 * 1024))
+    except (ValueError, OSError):
+        pass
+
+
+def make_tool(scratch_dir: str | Path = ".repomind_cache/scratch", timeout: int = 30) -> ToolSpec:
+    scratch = Path(scratch_dir)
+    scratch.mkdir(parents=True, exist_ok=True)
+
+    def run(code: str, timeout_seconds: int = 0) -> ToolResult:
+        timeout_s = timeout_seconds if 0 < timeout_seconds <= timeout else timeout
+        with tempfile.NamedTemporaryFile("w", suffix=".py", dir=str(scratch), delete=False) as f:
+            f.write(PREAMBLE + "\n" + code)
+            script_path = f.name
+
+        env = os.environ.copy()
+        env["PYTHONDONTWRITEBYTECODE"] = "1"
+        env["NO_COLOR"] = "1"
+        env["PYTHONIOENCODING"] = "utf-8"
+        # block obvious network env that requests / urllib3 read
+        for k in ("HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY"):
+            env.pop(k, None)
+
+        try:
+            proc = subprocess.run(
+                [sys.executable, "-I", "-S", script_path],
+                capture_output=True, text=True, timeout=timeout_s,
+                cwd=str(scratch), env=env,
+                preexec_fn=lambda: _set_limits(timeout_s, 1024),
+            )
+        except subprocess.TimeoutExpired:
+            return ToolResult(ok=False, output="", error=f"timeout after {timeout_s}s")
+        except Exception as e:
+            return ToolResult(ok=False, output="", error=f"sandbox error: {e}")
+        finally:
+            try:
+                os.unlink(script_path)
+            except OSError:
+                pass
+
+        out = (proc.stdout or "")[-8000:]
+        err = (proc.stderr or "")[-4000:]
+        if proc.returncode == 0:
+            return ToolResult(ok=True, output=out or "(no output)", extra={"returncode": 0})
+        return ToolResult(
+            ok=False,
+            output=out,
+            error=err.strip() or f"non-zero return: {proc.returncode}",
+            extra={"returncode": proc.returncode},
+        )
+
+    return ToolSpec(
+        name="execute_code",
+        description="Run a Python snippet in a sandboxed subprocess. No network, CPU+memory limits, isolated cwd.",
+        parameters={
+            "type": "object",
+            "properties": {
+                "code": {"type": "string", "description": "Python source to execute."},
+                "timeout_seconds": {"type": "integer", "default": 0, "description": "Override default timeout (cap 30s)."},
+            },
+            "required": ["code"],
+        },
+        runner=run,
+    )

tools/git_log.py

@@ -0,0 +1,50 @@
+"""git_log tool — read-only commit history queries."""
+from __future__ import annotations
+import subprocess
+from pathlib import Path
+
+from .base import ToolResult, ToolSpec
+
+
+def make_tool(repo_root: str | Path, max_commits: int = 50) -> ToolSpec:
+    root = Path(repo_root).resolve()
+
+    def run(path: str = "", limit: int = 20, since: str = "") -> ToolResult:
+        if not (root / ".git").exists():
+            return ToolResult(ok=False, output="", error="not a git repository")
+        n = max(1, min(limit, max_commits))
+        cmd = ["git", "log", f"-n{n}", "--pretty=format:%h|%an|%ai|%s"]
+        if since:
+            cmd += [f"--since={since}"]
+        if path:
+            cmd += ["--", path]
+        try:
+            proc = subprocess.run(cmd, cwd=str(root), capture_output=True, text=True, timeout=30)
+        except subprocess.TimeoutExpired:
+            return ToolResult(ok=False, output="", error="git log timeout")
+        if proc.returncode != 0:
+            return ToolResult(ok=False, output="", error=proc.stderr.strip())
+        rows = [_format_row(line) for line in proc.stdout.splitlines() if line]
+        return ToolResult(ok=True, output="\n".join(rows), extra={"commits": len(rows)})
+
+    return ToolSpec(
+        name="git_log",
+        description="Read commit history. Optionally filter by path or --since.",
+        parameters={
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "default": ""},
+                "limit": {"type": "integer", "default": 20},
+                "since": {"type": "string", "default": "", "description": "git --since (e.g. '2 weeks ago')"},
+            },
+        },
+        runner=run,
+    )
+
+
+def _format_row(line: str) -> str:
+    parts = line.split("|", 3)
+    if len(parts) != 4:
+        return line
+    sha, author, date, subject = parts
+    return f"{sha[:9]}  {date[:10]}  {author[:24]:<24}  {subject}"

tools/grep.py

@@ -0,0 +1,95 @@
+"""grep_codebase tool — ripgrep-style search inside the ingested repo.
+
+Uses Python's `re` so we don't depend on rg being installed; that lets the
+tool run identically in tests, in the local sandbox, and on AMD Cloud.
+"""
+from __future__ import annotations
+import os
+import re
+from pathlib import Path
+from typing import List
+
+from .base import ToolResult, ToolSpec
+
+
+SKIP_DIRS = {".git", "node_modules", ".venv", "venv", "__pycache__", "target", "build", "dist"}
+
+
+def make_tool(
+    repo_root: str | Path,
+    max_matches: int = 200,
+    max_file_size: int = 2_000_000,
+) -> ToolSpec:
+    root = Path(repo_root).resolve()
+
+    def run(pattern: str, path: str = "", case_sensitive: bool = False, max_results: int = 50) -> ToolResult:
+        try:
+            flags = 0 if case_sensitive else re.IGNORECASE
+            rx = re.compile(pattern, flags)
+        except re.error as e:
+            return ToolResult(ok=False, output="", error=f"invalid regex: {e}")
+
+        scope = (root / path).resolve() if path else root
+        try:
+            scope.relative_to(root)
+        except ValueError:
+            return ToolResult(ok=False, output="", error=f"path outside repo: {path}")
+        if not scope.exists():
+            return ToolResult(ok=False, output="", error=f"not found: {path}")
+
+        hits: List[str] = []
+        n = 0
+        cap = min(max_results, max_matches)
+
+        def consider(filepath: Path):
+            nonlocal n
+            if n >= cap:
+                return
+            try:
+                if filepath.stat().st_size > max_file_size:
+                    return
+            except OSError:
+                return
+            try:
+                text = filepath.read_text(encoding="utf-8", errors="replace")
+            except OSError:
+                return
+            for ln, line in enumerate(text.split("\n"), start=1):
+                if rx.search(line):
+                    rel = str(filepath.relative_to(root))
+                    hits.append(f"{rel}:{ln}: {line.rstrip()}")
+                    n += 1
+                    if n >= cap:
+                        return
+
+        if scope.is_file():
+            consider(scope)
+        else:
+            for dirpath, dirnames, filenames in os.walk(scope):
+                dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
+                for fn in filenames:
+                    consider(Path(dirpath) / fn)
+                    if n >= cap:
+                        break
+                if n >= cap:
+                    break
+
+        if not hits:
+            return ToolResult(ok=True, output="(no matches)", extra={"matches": 0})
+        return ToolResult(ok=True, output="\n".join(hits), extra={"matches": n, "capped": n >= cap})
+
+    return ToolSpec(
+        name="grep_codebase",
+        description="Search regular expression across files in the ingested repo. Returns path:line:match.",
+        parameters={
+            "type": "object",
+            "properties": {
+                "pattern": {"type": "string"},
+                "path": {"type": "string", "description": "Limit search to this subpath. Empty = whole repo.", "default": ""},
+                "case_sensitive": {"type": "boolean", "default": False},
+                "max_results": {"type": "integer", "default": 50},
+            },
+            "required": ["pattern"],
+        },
+        runner=run,
+    )

tools/read_file.py

@@ -0,0 +1,50 @@
+"""read_file tool — read a slice of a file inside the ingested repo root."""
+from __future__ import annotations
+from pathlib import Path
+
+from .base import ToolResult, ToolSpec
+
+
+def make_tool(repo_root: str | Path, max_bytes: int = 200_000) -> ToolSpec:
+    root = Path(repo_root).resolve()
+
+    def run(path: str, start_line: int = 1, end_line: int = -1) -> ToolResult:
+        # security: only allow paths inside the ingested root
+        target = (root / path).resolve()
+        try:
+            target.relative_to(root)
+        except ValueError:
+            return ToolResult(ok=False, output="", error=f"path outside repo: {path}")
+        if not target.exists():
+            return ToolResult(ok=False, output="", error=f"not found: {path}")
+        try:
+            text = target.read_text(encoding="utf-8", errors="replace")
+        except OSError as e:
+            return ToolResult(ok=False, output="", error=str(e))
+        if len(text) > max_bytes:
+            text = text[:max_bytes] + f"\n[... truncated at {max_bytes} bytes]"
+        lines = text.split("\n")
+        start = max(1, start_line)
+        end = len(lines) if end_line in (-1, 0) else min(len(lines), end_line)
+        slice_text = "\n".join(lines[start - 1:end])
+        numbered = "\n".join(f"{i:>5}  {l}" for i, l in enumerate(lines[start - 1:end], start=start))
+        return ToolResult(
+            ok=True,
+            output=numbered,
+            extra={"path": path, "lines": (start, end), "total_lines": len(lines)},
+        )
+
+    return ToolSpec(
+        name="read_file",
+        description="Read a file from the ingested repo. Optionally restrict to a line range.",
+        parameters={
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "Path relative to repo root."},
+                "start_line": {"type": "integer", "description": "1-indexed inclusive start.", "default": 1},
+                "end_line": {"type": "integer", "description": "1-indexed inclusive end. -1 = end of file.", "default": -1},
+            },
+            "required": ["path"],
+        },
+        runner=run,
+    )

tools/registry.py

@@ -0,0 +1,20 @@
+"""Build the default tool registry for a given repo root."""
+from __future__ import annotations
+from pathlib import Path
+
+from .base import ToolRegistry
+from .execute_code import make_tool as make_execute
+from .git_log import make_tool as make_git_log
+from .grep import make_tool as make_grep
+from .read_file import make_tool as make_read_file
+from .run_tests import make_tool as make_run_tests
+
+
+def default_registry(repo_root: str | Path, scratch_dir: str | Path = ".repomind_cache/scratch") -> ToolRegistry:
+    reg = ToolRegistry()
+    reg.register(make_read_file(repo_root))
+    reg.register(make_grep(repo_root))
+    reg.register(make_execute(scratch_dir))
+    reg.register(make_run_tests(repo_root))
+    reg.register(make_git_log(repo_root))
+    return reg

tools/run_tests.py

@@ -0,0 +1,54 @@
+"""run_tests tool — pytest invocation inside the ingested repo (read-only)."""
+from __future__ import annotations
+import subprocess
+import sys
+from pathlib import Path
+
+from .base import ToolResult, ToolSpec
+
+
+def make_tool(repo_root: str | Path, timeout: int = 120) -> ToolSpec:
+    root = Path(repo_root).resolve()
+
+    def run(test_path: str = "", k_expression: str = "", max_lines: int = 200) -> ToolResult:
+        target = (root / test_path).resolve() if test_path else root
+        try:
+            target.relative_to(root)
+        except ValueError:
+            return ToolResult(ok=False, output="", error=f"path outside repo: {test_path}")
+
+        cmd = [sys.executable, "-m", "pytest", "-x", "--tb=short", "-q", str(target)]
+        if k_expression:
+            cmd += ["-k", k_expression]
+        try:
+            proc = subprocess.run(
+                cmd, capture_output=True, text=True, cwd=str(root), timeout=timeout,
+            )
+        except subprocess.TimeoutExpired:
+            return ToolResult(ok=False, output="", error=f"pytest timeout after {timeout}s")
+        except FileNotFoundError:
+            return ToolResult(ok=False, output="", error="pytest not installed")
+
+        lines = (proc.stdout or "").splitlines()[-max_lines:]
+        out = "\n".join(lines)
+        err = (proc.stderr or "").strip()
+        if proc.returncode == 0:
+            return ToolResult(ok=True, output=out or "(all passed)", extra={"returncode": 0})
+        return ToolResult(
+            ok=False, output=out, error=err or f"pytest exit {proc.returncode}",
+            extra={"returncode": proc.returncode},
+        )
+
+    return ToolSpec(
+        name="run_tests",
+        description="Run pytest on the ingested repo (or a sub-path). Read-only.",
+        parameters={
+            "type": "object",
+            "properties": {
+                "test_path": {"type": "string", "default": ""},
+                "k_expression": {"type": "string", "default": "", "description": "pytest -k expression"},
+                "max_lines": {"type": "integer", "default": 200},
+            },
+        },
+        runner=run,
+    )