Ghousted/CodeSage
0
1import asyncio2import logging3import os4import re5from typing import List, Literal, Optional6 7import requests8from fastapi import APIRouter, HTTPException, Query9from pydantic import BaseModel, field_validator10 11from concurrent.futures import ThreadPoolExecutor12from core.ingestion import clone_repo, scan_files, cleanup_repo, IngestionError13from core.chunker import chunk_files_parallel14from core.embedder import embed_chunks15from core.vector_store import upsert_chunks, delete_namespace, namespace_exists16from core.retriever import retrieve17from core.llm import generate_answer, rewrite_query_with_history, LLMConfigError, LLMUnavailableError18from core.summarizer import generate_summary19from core.analyzer import analyze_structure20from core.jobs import job_store21 22logger = logging.getLogger(__name__)23router = APIRouter()24 25_GITHUB_URL_RE = re.compile(r"^https://github\.com/[\w.-]+/[\w.-]+(?:\.git)?/?$")26 27 28class AnalyzeRequest(BaseModel):29 repo_url: str30 31 @field_validator("repo_url")32 @classmethod33 def validate_github_url(cls, v: str) -> str:34 v = v.strip()35 if not _GITHUB_URL_RE.match(v):36 raise ValueError("repo_url must be a valid GitHub repository URL")37 return v38 39 40class ChatTurn(BaseModel):41 role: Literal["user", "assistant"]42 content: str43 44 @field_validator("content")45 @classmethod46 def validate_content(cls, v: str) -> str:47 v = v.strip()48 if not v:49 raise ValueError("content must not be empty")50 return v51 52 53class AskRequest(BaseModel):54 question: str55 repo_url: str56 k: int = 557 history: Optional[List[ChatTurn]] = None58 59 @field_validator("question")60 @classmethod61 def validate_question(cls, v: str) -> str:62 v = v.strip()63 if not v:64 raise ValueError("question must not be empty")65 if len(v) > 2000:66 raise ValueError("question must be under 2000 characters")67 return v68 69 @field_validator("k")70 @classmethod71 def validate_k(cls, v: int) -> int:72 if not (1 <= v <= 20):73 raise ValueError("k must be between 1 and 20")74 return v75 76 @field_validator("history")77 @classmethod78 def validate_history(cls, v: Optional[List[ChatTurn]]) -> Optional[List[ChatTurn]]:79 # Cap server-side too — defends against a runaway client.80 if v is not None and len(v) > 20:81 return v[-20:]82 return v83 84 85def _repo_namespace(repo_url: str) -> str:86 """Derive a stable Pinecone namespace from a repo URL."""87 tail = repo_url.rstrip("/").removesuffix(".git").split("github.com/", 1)[-1]88 sanitized = re.sub(r"[^a-zA-Z0-9_-]", "_", tail).strip("_")89 return sanitized or "default"90 91 92def _owner_repo(repo_url: str) -> tuple[str, str]:93 """Extract (owner, repo) from a validated GitHub URL."""94 tail = repo_url.rstrip("/").removesuffix(".git").split("github.com/", 1)[-1]95 parts = tail.split("/")96 if len(parts) < 2:97 raise ValueError("Invalid GitHub URL")98 return parts[0], parts[1]99 100 101# -------------------- Background indexing --------------------102 103def _safe_summary(repo_path: str, file_paths: list[str]) -> str:104 """Generate the project summary, swallowing failures so they don't sink the job."""105 try:106 return generate_summary(repo_path, file_paths)["summary"]107 except Exception as exc:108 logger.warning("Summary generation failed: %s", exc)109 return "(Summary unavailable — LLM call did not return in time.)"110 111 112def _run_indexing(job_id: str, repo_url: str) -> None:113 """Synchronous indexing pipeline. Executed in a worker thread."""114 namespace = _repo_namespace(repo_url)115 repo_path = None116 indexed_branch: str | None = None117 try:118 job_store.update(job_id, status="running", stage="cloning")119 repo_path = clone_repo(repo_url)120 # Capture which branch was actually checked out for the indexed snapshot.121 try:122 import git as _git123 indexed_branch = _git.Repo(repo_path).active_branch.name124 except Exception:125 indexed_branch = None126 127 job_store.update(job_id, stage="scanning")128 file_paths = scan_files(repo_path)129 if not file_paths:130 raise IngestionError("No supported source files found in repository.")131 132 job_store.update(job_id, stage="chunking")133 all_chunks = chunk_files_parallel(file_paths)134 if not all_chunks:135 raise IngestionError("No chunkable code found in repository.")136 137 # Kick off the LLM-bound summary in parallel — it has nothing to do with138 # embedding or indexing and is usually the longest single network call.139 with ThreadPoolExecutor(max_workers=2) as bg:140 summary_future = bg.submit(_safe_summary, repo_path, file_paths)141 structure_future = bg.submit(analyze_structure, repo_path, file_paths)142 143 job_store.update(job_id, stage="embedding")144 embed_chunks(all_chunks)145 146 job_store.update(job_id, stage="indexing")147 delete_namespace(namespace) # safe even if missing148 stored_count = upsert_chunks(all_chunks, namespace)149 150 job_store.update(job_id, stage="finalizing")151 structure_data = structure_future.result()152 summary = summary_future.result()153 154 job_store.update(155 job_id,156 status="completed",157 stage="done",158 result={159 "namespace": namespace,160 "repo_url": repo_url,161 "indexed_branch": indexed_branch,162 "files_indexed": len(file_paths),163 "chunks_stored": stored_count,164 "summary": summary,165 "structure": structure_data,166 },167 )168 169 except IngestionError as exc:170 logger.info("Indexing failed for %s: %s", repo_url, exc)171 job_store.update(job_id, status="failed", stage="error", error=str(exc))172 except Exception as exc:173 logger.exception("Unexpected indexing failure for %s", repo_url)174 job_store.update(job_id, status="failed", stage="error", error=f"{type(exc).__name__}: {exc}")175 finally:176 if repo_path:177 try:178 cleanup_repo(repo_path)179 except Exception as exc:180 logger.warning("Cleanup failed for %s: %s", repo_path, exc)181 182 183# -------------------- Endpoints --------------------184 185@router.post("/analyze")186async def analyze(request: AnalyzeRequest):187 """Kick off a background indexing job. Returns a job_id for polling."""188 # Fail-fast on private / nonexistent repos before spending minutes indexing.189 try:190 owner, repo = _owner_repo(request.repo_url)191 except ValueError:192 raise HTTPException(status_code=422, detail="Could not parse repo URL")193 try:194 check = await asyncio.to_thread(195 _gh_get, f"https://api.github.com/repos/{owner}/{repo}"196 )197 except requests.RequestException:198 check = None199 if check is not None and check.status_code == 404:200 raise HTTPException(status_code=404, detail=_private_repo_message(owner, repo))201 # 403 (rate-limited) → don't block analyze; clone may still succeed for public repos.202 203 job = job_store.create()204 # Run blocking pipeline in a worker thread so the event loop stays free205 asyncio.create_task(asyncio.to_thread(_run_indexing, job.id, request.repo_url))206 return {"job_id": job.id, "status": job.status}207 208 209@router.get("/jobs/{job_id}")210async def get_job(job_id: str):211 job = job_store.get(job_id)212 if job is None:213 raise HTTPException(status_code=404, detail="Job not found or expired.")214 return job.to_dict()215 216 217@router.post("/ask")218async def ask(request: AskRequest):219 """Retrieve relevant code chunks and generate a grounded answer."""220 namespace = _repo_namespace(request.repo_url)221 222 if not namespace_exists(namespace):223 raise HTTPException(224 status_code=404,225 detail="This repository has not been indexed yet. Run /analyze first.",226 )227 228 history_payload = (229 [{"role": t.role, "content": t.content} for t in request.history]230 if request.history else None231 )232 233 # If there's prior turns, condense the follow-up into a self-contained query234 # so retrieval doesn't get a context-less "tell me more" or "show me that".235 # Rewrite is best-effort — a failure falls back to the literal question.236 search_query = request.question237 if history_payload:238 try:239 search_query = await asyncio.to_thread(240 rewrite_query_with_history, request.question, history_payload241 )242 if search_query != request.question:243 logger.info("Rewrote follow-up for retrieval: %r → %r", request.question, search_query)244 except Exception as exc:245 logger.warning("Query rewrite raised %s — using literal question", exc)246 search_query = request.question247 248 try:249 chunks = await asyncio.to_thread(retrieve, search_query, namespace, request.k)250 except Exception as exc:251 logger.exception("Retrieval failed")252 raise HTTPException(status_code=502, detail=f"Retrieval failed: {exc}") from exc253 254 if not chunks:255 raise HTTPException(256 status_code=404,257 detail="No relevant code found for this question.",258 )259 260 try:261 result = await asyncio.to_thread(generate_answer, request.question, chunks, history_payload)262 except LLMConfigError as exc:263 raise HTTPException(status_code=500, detail=str(exc)) from exc264 except LLMUnavailableError as exc:265 raise HTTPException(status_code=503, detail=str(exc)) from exc266 except Exception as exc:267 logger.exception("Answer generation failed")268 raise HTTPException(status_code=502, detail=f"Answer generation failed: {exc}") from exc269 270 return result271 272 273# Allow file paths with most reasonable filename characters; block traversal.274_SAFE_PATH_RE = re.compile(r"^[\w./\- ()+@,'\[\]&]+$")275_SAFE_BRANCH_RE = re.compile(r"^[\w./\-]+$")276_MAX_FILE_BYTES = 1_000_000 # 1 MB cap on remotely-fetched file277_RAW_TIMEOUT_SECONDS = 15278_API_TIMEOUT_SECONDS = 10279 280# Caches keyed by (owner, repo). Tree cache also keyed by branch.281_default_branch_cache: dict[tuple[str, str], str] = {}282_branches_cache: dict[tuple[str, str], list[str]] = {}283_tree_cache: dict[tuple[str, str, str], dict] = {}284 285 286def _gh_headers() -> dict[str, str]:287 headers = {"Accept": "application/vnd.github+json"}288 token = os.environ.get("GITHUB_TOKEN")289 if token:290 # Optional — raises rate limit from 60/hour to 5000/hour291 headers["Authorization"] = f"Bearer {token}"292 return headers293 294 295def _gh_get(url: str) -> requests.Response:296 return requests.get(url, headers=_gh_headers(), timeout=_API_TIMEOUT_SECONDS)297 298 299def _rate_limit_message() -> str:300 return (301 "Rate-limited by GitHub. Set GITHUB_TOKEN in backend/.env to raise the limit "302 "from 60/hour to 5000/hour, then restart the server."303 )304 305 306def _private_repo_message(owner: str, repo: str) -> str:307 return (308 f"Repository {owner}/{repo} is not publicly accessible. "309 f"CodeSage only supports public GitHub repositories. "310 f"If the repo URL is correct, make the repository public on GitHub and try again."311 )312 313 314def _resolve_default_branch(owner: str, repo: str) -> str | None:315 key = (owner, repo)316 if key in _default_branch_cache:317 return _default_branch_cache[key]318 try:319 res = _gh_get(f"https://api.github.com/repos/{owner}/{repo}")320 if res.status_code == 200:321 branch = res.json().get("default_branch")322 if branch:323 _default_branch_cache[key] = branch324 return branch325 except requests.RequestException as exc:326 logger.warning("GitHub API lookup failed for %s/%s: %s", owner, repo, exc)327 return None328 329 330def _list_branches(owner: str, repo: str) -> list[str]:331 """Return all branches for a repo (paginated, capped at 300)."""332 key = (owner, repo)333 if key in _branches_cache:334 return _branches_cache[key]335 336 branches: list[str] = []337 for page in range(1, 4): # up to 3 pages × 100 = 300 branches338 try:339 res = _gh_get(340 f"https://api.github.com/repos/{owner}/{repo}/branches"341 f"?per_page=100&page={page}"342 )343 except requests.RequestException as exc:344 logger.warning("Branches lookup failed for %s/%s: %s", owner, repo, exc)345 break346 if res.status_code == 403:347 raise HTTPException(status_code=429, detail=_rate_limit_message())348 if res.status_code == 404:349 raise HTTPException(status_code=404, detail=_private_repo_message(owner, repo))350 if res.status_code != 200:351 raise HTTPException(352 status_code=res.status_code,353 detail=f"GitHub API returned {res.status_code} when listing branches.",354 )355 page_branches = [b["name"] for b in res.json()]356 branches.extend(page_branches)357 if len(page_branches) < 100:358 break359 360 _branches_cache[key] = branches361 return branches362 363 364def _fetch_tree(owner: str, repo: str, branch: str) -> dict:365 key = (owner, repo, branch)366 if key in _tree_cache:367 return _tree_cache[key]368 369 res = _gh_get(370 f"https://api.github.com/repos/{owner}/{repo}/git/trees/{branch}?recursive=1"371 )372 if res.status_code == 403:373 raise HTTPException(status_code=429, detail=_rate_limit_message())374 if res.status_code == 404:375 # Could be: branch doesn't exist, or repo is private and we can't see it.376 # Distinguish by probing the repo root via the same auth context.377 repo_check = _gh_get(f"https://api.github.com/repos/{owner}/{repo}")378 if repo_check.status_code == 404:379 raise HTTPException(status_code=404, detail=_private_repo_message(owner, repo))380 raise HTTPException(status_code=404, detail=f"Branch {branch!r} not found in {owner}/{repo}.")381 if res.status_code != 200:382 raise HTTPException(383 status_code=res.status_code,384 detail=f"GitHub API returned {res.status_code} when fetching tree.",385 )386 387 data = res.json()388 entries = [389 {"path": e["path"], "type": e["type"], "size": e.get("size", 0)}390 for e in data.get("tree", [])391 ]392 result = {393 "branch": branch,394 "entries": entries,395 "truncated": data.get("truncated", False),396 }397 _tree_cache[key] = result398 return result399 400 401@router.get("/branches")402async def list_branches(repo_url: str = Query(...)):403 """Return all branches for a repo plus the default branch name."""404 if not _GITHUB_URL_RE.match(repo_url.strip()):405 raise HTTPException(status_code=422, detail="Invalid repo_url")406 try:407 owner, repo = _owner_repo(repo_url.strip())408 except ValueError:409 raise HTTPException(status_code=422, detail="Could not parse repo URL")410 411 branches = await asyncio.to_thread(_list_branches, owner, repo)412 default_branch = await asyncio.to_thread(_resolve_default_branch, owner, repo)413 return {"default_branch": default_branch, "branches": branches}414 415 416@router.get("/tree")417async def get_tree(418 repo_url: str = Query(...),419 branch: str = Query(..., description="Branch name to load the tree for"),420):421 """Return the recursive file tree for a given branch."""422 if not _GITHUB_URL_RE.match(repo_url.strip()):423 raise HTTPException(status_code=422, detail="Invalid repo_url")424 if not _SAFE_BRANCH_RE.match(branch):425 raise HTTPException(status_code=422, detail="Invalid branch name")426 try:427 owner, repo = _owner_repo(repo_url.strip())428 except ValueError:429 raise HTTPException(status_code=422, detail="Could not parse repo URL")430 return await asyncio.to_thread(_fetch_tree, owner, repo, branch)431 432 433@router.get("/file")434async def get_file(435 repo_url: str = Query(..., description="GitHub repository URL"),436 path: str = Query(..., description="File path within the repo"),437 branch: str | None = Query(None, description="Branch name; falls back to default if omitted"),438):439 """Stream a single file's contents from GitHub raw."""440 if not _GITHUB_URL_RE.match(repo_url.strip()):441 raise HTTPException(status_code=422, detail="Invalid repo_url")442 if ".." in path or path.startswith("/") or not _SAFE_PATH_RE.match(path):443 raise HTTPException(status_code=422, detail="Invalid file path")444 if branch is not None and not _SAFE_BRANCH_RE.match(branch):445 raise HTTPException(status_code=422, detail="Invalid branch name")446 447 try:448 owner, repo = _owner_repo(repo_url.strip())449 except ValueError:450 raise HTTPException(status_code=422, detail="Could not parse repo URL")451 452 # Build candidate branches: explicit > default > main > master453 branches: list[str] = []454 if branch:455 branches.append(branch)456 else:457 default_branch = await asyncio.to_thread(_resolve_default_branch, owner, repo)458 if default_branch:459 branches.append(default_branch)460 for fallback in ("main", "master"):461 if fallback not in branches:462 branches.append(fallback)463 464 last_status = 404465 for b in branches:466 url = f"https://raw.githubusercontent.com/{owner}/{repo}/{b}/{path}"467 try:468 res = await asyncio.to_thread(469 requests.get, url, timeout=_RAW_TIMEOUT_SECONDS,470 )471 except requests.RequestException as exc:472 logger.warning("Raw fetch failed for %s: %s", url, exc)473 continue474 if res.status_code == 200:475 content = res.content[:_MAX_FILE_BYTES]476 try:477 text = content.decode("utf-8")478 except UnicodeDecodeError:479 raise HTTPException(status_code=415, detail="File is not UTF-8 text")480 truncated = len(res.content) > _MAX_FILE_BYTES481 return {482 "path": path,483 "branch": b,484 "content": text,485 "truncated": truncated,486 "size_bytes": len(content),487 }488 last_status = res.status_code489 490 if branch:491 raise HTTPException(492 status_code=last_status,493 detail=f"File {path!r} not found on branch {branch!r}.",494 )495 raise HTTPException(496 status_code=last_status,497 detail=(498 f"File {path!r} not found on any tried branch ({', '.join(branches)}). "499 f"{_rate_limit_message()}"500 ),501 )502