Upload 9 files

code_analysis.py

@@ -0,0 +1,53 @@
+import ast
+from llm import get_llm_client, build_prompt
+from rag.rag_chain import build_context
+from rag.retriever import retrieve_relevant_chunks
+
+
+def explain_function(file_path: str, function_name: str) -> str:
+    function_code = extract_function_source(file_path, function_name)
+    related_chunks = retrieve_relevant_chunks(f"usages of {function_name}", k=3)
+    context = build_context(related_chunks)
+
+    prompt = build_prompt(function_code, context, task_type="qa")
+    return get_llm_client().generate(prompt)
+
+
+def extract_function_source(file_path: str, function_name: str) -> str:
+    with open(file_path, "r", encoding="utf-8") as f:
+        source = f.read()
+
+    tree = ast.parse(source)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == function_name:
+            return ast.get_source_segment(source, node)
+
+    raise ValueError(f"Function '{function_name}' not found in {file_path}")
+
+
+def detect_bugs(file_path: str) -> list[dict]:
+    with open(file_path, "r", encoding="utf-8") as f:
+        code = f.read()
+
+    prompt = build_prompt("", code, task_type="bug_finding")
+    raw_response = get_llm_client().generate(prompt)
+
+    import json
+    try:
+        return json.loads(raw_response)
+    except json.JSONDecodeError:
+        return [{"line": 0, "issue": "Could not parse model output", "severity": "unknown", "suggestion": raw_response}]
+
+
+def analyze_complexity(file_path: str) -> dict:
+    from radon.complexity import cc_visit, cc_rank
+    with open(file_path, "r", encoding="utf-8") as f:
+        code = f.read()
+
+    results = cc_visit(code)
+    return {
+        "functions": [
+            {"name": r.name, "complexity": r.complexity, "rank": cc_rank(r.complexity)}
+            for r in results
+        ]
+    }

documentation.py

@@ -0,0 +1,61 @@
+import os
+from llm import get_llm_client, build_prompt
+from rag.repository_loader import load_repository
+
+
+def generate_docstring(function_code: str) -> str:
+    prompt = build_prompt(function_code, "", task_type="docstring")
+    return get_llm_client().generate(prompt)
+
+
+def generate_module_docs(file_path: str) -> str:
+    with open(file_path, "r", encoding="utf-8") as f:
+        code = f.read()
+
+    prompt = build_prompt(
+        f"Generate documentation for this module: {file_path}",
+        code,
+        task_type="qa",
+    )
+    return get_llm_client().generate(prompt)
+
+
+def summarize_repo_structure(root_path: str) -> dict:
+    documents = load_repository(root_path)
+
+    tech_stack = detect_tech_stack(root_path)
+    file_tree = build_file_tree(documents)
+
+    return {
+        "total_files": len(documents),
+        "tech_stack": tech_stack,
+        "file_tree": file_tree,
+    }
+
+
+def detect_tech_stack(root_path: str) -> list[str]:
+    markers = {
+        "requirements.txt": "Python",
+        "package.json": "Node.js",
+        "go.mod": "Go",
+        "pom.xml": "Java (Maven)",
+    }
+    found = []
+    for marker_file, tech_name in markers.items():
+        if os.path.exists(os.path.join(root_path, marker_file)):
+            found.append(tech_name)
+    return found
+
+
+def build_file_tree(documents: list) -> list[str]:
+    return sorted(doc.file_path for doc in documents)
+
+
+def generate_readme(root_path: str) -> str:
+    summary = summarize_repo_structure(root_path)
+    prompt = build_prompt(
+        f"Generate a README.md for a project with this structure: {summary}",
+        "",
+        task_type="qa",
+    )
+    return get_llm_client().generate(prompt)

embedding.py

@@ -0,0 +1,30 @@
+# pyrefly: ignore [missing-import]
+from langchain_google_genai import GoogleGenerativeAIEmbeddings
+from config import get_settings
+from rag.repository_loader import CodeDocument
+
+
+def get_embedding_client() -> GoogleGenerativeAIEmbeddings:
+    settings = get_settings()
+
+    return GoogleGenerativeAIEmbeddings(
+        model=settings.embedding_model,
+        google_api_key=settings.gemini_api_key,
+    )
+
+
+def embed_document(chunks: list[dict]) -> list[dict]:
+    embeddings = get_embedding_client()
+    texts = [chunk["content"] for chunk in chunks]
+    vectors = embeddings.embed_documents(texts)
+
+    for chunk, vector in zip(chunks, vectors):
+        chunk["embedding"] = vector
+    return chunks
+
+
+def embed_query(query: str) -> list[float]:
+    embeddings = get_embedding_client()
+    return embeddings.embed_query(query)
+
+

file_creator.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass, asdict
+from llm import get_llm_client, build_prompt
+from rag.rag_chain import build_context
+from rag.retriever import retrieve_relevant_chunks
+
+
+@dataclass
+class FileProposal:
+    path: str
+    content: str
+    reason: str
+    approved: bool = False
+
+
+def propose_new_file(description: str, context_query: str | None = None) -> FileProposal:
+    context = ""
+    if context_query:
+        related_chunks = retrieve_relevant_chunks(context_query, k=5)
+        context = build_context(related_chunks)
+
+    prompt = build_prompt(description, context, task_type="file_creation")
+    content = get_llm_client().generate(prompt)
+
+    proposed_path = infer_file_path(description)
+
+    return FileProposal(path=proposed_path, content=content, reason=description)
+
+
+def infer_file_path(description: str) -> str:
+    prompt = f"Given this file creation request: '{description}', respond with ONLY a suitable relative file path, nothing else."
+    path = get_llm_client().generate(prompt).strip()
+    return path
+
+
+async def apply_approved_file(proposal: FileProposal, user_confirmed: bool, base_dir: str = ".") -> dict:
+    """Write the proposed file to local disk under *base_dir*.
+
+    The LLM-suggested path (e.g. /src/geometry/rectangle.js) is treated as
+    relative to *base_dir*, so leading slashes/backslashes are stripped before
+    joining to avoid accidental absolute-path writes.
+    """
+    if not user_confirmed:
+        return {"status": "rejected", "proposal": asdict(proposal)}
+
+    import os
+
+    # Strip any leading separators so the path is always relative to base_dir
+    relative_path = proposal.path.lstrip("/\\")
+    abs_path = os.path.join(base_dir, relative_path)
+
+    # Create parent directories if they don't exist
+    os.makedirs(os.path.dirname(abs_path), exist_ok=True)
+
+    with open(abs_path, "w", encoding="utf-8") as f:
+        f.write(proposal.content)
+
+    proposal.path = abs_path  # update so the caller can display the real path
+    return {"status": "written", "path": abs_path}

rag_chain.py

@@ -0,0 +1,25 @@
+from rag.retriever import retrieve_relevant_chunks
+from llm import get_llm_client, build_prompt
+
+
+def build_context(chunks: list[dict]) -> str:
+    parts = []
+    for c in chunks:
+        meta = c["metadata"]
+        header = f"# {meta['file_path']} (lines {meta['start_line']}-{meta['end_line']})"
+        parts.append(f"{header}\n{c['content']}")
+    return "\n\n---\n\n".join(parts)
+
+
+def run_rag_query(query: str, k: int = 5) -> dict:
+    chunks = retrieve_relevant_chunks(query, k)
+    context = build_context(chunks)
+
+    prompt = build_prompt(query, context, task_type="qa")
+    llm = get_llm_client()
+    answer = llm.generate(prompt)
+
+    return {
+        "answer": answer,
+        "sources": [c["metadata"] for c in chunks],
+    }

repository_loader.py

@@ -0,0 +1,152 @@
+from typing import List
+from config import get_settings
+import os
+from dataclasses import dataclass
+
+
+@dataclass
+class CodeDocument:
+    content: str
+    file_path: str
+    language: str
+    size_bytes: int
+
+
+IGNORE_DIRS = {".git", "node_modules", "__pycache__", "venv", ".venv", "dist", "build"}
+LANGUAGE_BY_EXT = {
+    ".py": "python",
+    ".js": "javascript",
+    ".ts": "typescript",
+    ".java": "java",
+    ".go": "go",
+    ".md": "markdown",
+}
+
+
+def load_repository(root_path: str) -> List[CodeDocument]:
+    documents: List[CodeDocument] = []
+    for dirpath, dirnames, filenames in os.walk(root_path):
+        dirnames[:] = [d for d in dirnames if d not in IGNORE_DIRS]
+        for filename in filenames:
+            fpath = os.path.join(dirpath, filename)
+            if should_include(fpath):
+                documents.append(read_file_with_metadata(fpath))
+    return documents
+
+
+def should_include(fpath: str) -> bool:
+    settings = get_settings()
+    _, ext = os.path.splitext(fpath)
+    if ext not in settings.allowed_extensions:
+        return False
+    try:
+        size_kb = os.path.getsize(fpath)/1024
+    except OSError:
+        return False
+    return size_kb <= settings.max_file_size_kb
+
+
+def read_file_with_metadata(filepath: str) -> CodeDocument:
+    with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
+        content = f.read()
+    return CodeDocument(
+        content=content,
+        file_path=filepath,
+        language=detect_language(filepath),
+        size_bytes=len(content.encode("utf-8")),
+    )
+
+
+def detect_language(filepath: str) -> str:
+    _, ext = os.path.splitext(filepath)
+    return LANGUAGE_BY_EXT.get(ext, "unknown")
+
+
+
+
+
+"""
+import os
+from dataclasses import dataclass
+from langchain_community.document_loaders import DirectoryLoader, TextLoader
+from config import get_settings
+
+
+@dataclass
+class CodeDocument:
+    content: str
+    file_path: str
+    language: str
+    size_bytes: int
+
+LANGUAGE_BY_EXT = {
+    ".py": "python",
+    ".js": "javascript",
+    ".ts": "typescript",
+    ".java": "java",
+    ".go": "go",
+    ".cpp": "cpp",
+    ".c": "c",
+    ".cs": "csharp",
+    ".html": "html",
+    ".css": "css",
+    ".json": "json",
+    ".yaml": "yaml",
+    ".yml": "yaml",
+    ".md": "markdown",
+}
+def detect_language(filepath: str) -> str:
+    _, ext = os.path.splitext(filepath)
+    return LANGUAGE_BY_EXT.get(ext.lower(), "unknown")
+
+def load_repository(root_path: str) -> list[CodeDocument]:
+    settings = get_settings()
+    loader = DirectoryLoader(
+        path=root_path,
+        glob="**/*",
+        recursive=True,
+        silent_errors=True,
+        loader_cls=TextLoader,
+        loader_kwargs={
+            "encoding": "utf-8",
+            "autodetect_encoding": True,
+        },
+        exclude=[
+            "**/.git/**",
+            "**/.venv/**",
+            "**/venv/**",
+            "**/__pycache__/**",
+            "**/node_modules/**",
+            "**/dist/**",
+            "**/build/**",
+        ],
+    )
+    documents = loader.load()
+    code_documents: list[CodeDocument] = []
+
+    for doc in documents:
+        filepath = doc.metadata["source"]
+        _, ext = os.path.splitext(filepath)
+
+        if ext.lower() not in settings.allowed_extensions:
+            continue
+
+        try:
+            size_bytes = os.path.getsize(filepath)
+        except OSError:
+            continue
+
+        if size_bytes > settings.max_file_size_kb * 1024:
+            continue
+
+        code_documents.append(
+            CodeDocument(
+                content=doc.page_content,
+                file_path=filepath,
+                language=detect_language(filepath),
+                size_bytes=size_bytes,
+            )
+        )
+
+    return code_documents
+"""

retriever.py

@@ -0,0 +1,59 @@
+import chromadb
+from config import get_settings
+from rag.embedding import embed_query
+
+
+def get_chroma_client():
+    settings = get_settings()
+    return chromadb.PersistentClient(path=settings.vector_db_path)
+
+
+def build_vector_store(embedded_chunks: list[dict]):
+    client = get_chroma_client()
+
+    # Always start fresh — delete any data from the previously ingested repo
+    # so that stale chunks never bleed into the current analysis session.
+    try:
+        client.delete_collection("codebase")
+    except Exception:
+        pass  # collection didn't exist yet — that's fine
+
+    collection = client.create_collection("codebase")
+
+    ids        = [f"{c['file_path']}::{c['chunk_index']}" for c in embedded_chunks]
+    embeddings = [c["embedding"] for c in embedded_chunks]
+    documents  = [c["content"]   for c in embedded_chunks]
+    metadatas  = [
+        {
+            "file_path":  c["file_path"],
+            "language":   c["language"],
+            "start_line": c["start_line"],
+            "end_line":   c["end_line"],
+        }
+        for c in embedded_chunks
+    ]
+    collection.add(ids=ids, embeddings=embeddings, documents=documents, metadatas=metadatas)
+    return collection
+
+
+def load_vector_store():
+    client = get_chroma_client()
+    return client.get_or_create_collection("codebase")
+
+
+def retrieve_relevant_chunks(query: str, k: int = 5) -> list[dict]:
+    collection   = load_vector_store()
+    query_vector = embed_query(query)
+    results      = collection.query(query_embeddings=[query_vector], n_results=k)
+    return format_results(results)
+
+
+def format_results(results: dict) -> list[dict]:
+    formatted = []
+    documents = results.get("documents", [[]])[0]
+    metadatas = results.get("metadatas", [[]])[0]
+
+    for doc_text, meta in zip(documents, metadatas):
+        formatted.append({"content": doc_text, "metadata": meta})
+
+    return formatted

routes.py

@@ -0,0 +1,71 @@
+from fastapi import APIRouter
+from pydantic import BaseModel
+
+from rag.rag_chain import run_rag_query
+from services.code_analysis import detect_bugs, explain_function, analyze_complexity
+from services.documentation import generate_module_docs, generate_readme
+from services.file_creator import propose_new_file, apply_approved_file, FileProposal
+
+router = APIRouter()
+
+
+class QueryRequest(BaseModel):
+    query: str
+    k: int = 5
+
+
+class ExplainRequest(BaseModel):
+    file_path: str
+    function_name: str
+
+
+class FileProposalRequest(BaseModel):
+    description: str
+    context_query: str | None = None
+
+
+class FileApprovalRequest(BaseModel):
+    proposal: dict
+    approved: bool
+
+
+@router.post("/query")
+def query(req: QueryRequest):
+    return run_rag_query(req.query, req.k)
+
+
+@router.post("/analyze/bugs")
+def bugs(file_path: str):
+    return {"bugs": detect_bugs(file_path)}
+
+
+@router.post("/analyze/complexity")
+def complexity(file_path: str):
+    return analyze_complexity(file_path)
+
+
+@router.post("/analyze/explain")
+def explain(req: ExplainRequest):
+    return {"explanation": explain_function(req.file_path, req.function_name)}
+
+
+@router.post("/docs/module")
+def module_docs(file_path: str):
+    return {"docs": generate_module_docs(file_path)}
+
+
+@router.post("/docs/readme")
+def readme(root_path: str):
+    return {"readme": generate_readme(root_path)}
+
+
+@router.post("/files/propose")
+def propose(req: FileProposalRequest):
+    proposal = propose_new_file(req.description, req.context_query)
+    return proposal
+
+
+@router.post("/files/approve")
+async def approve(req: FileApprovalRequest):
+    proposal = FileProposal(**req.proposal)
+    return await apply_approved_file(proposal, req.approved)

splitter.py

@@ -0,0 +1,61 @@
+from langchain_text_splitters import RecursiveCharacterTextSplitter, Language
+from rag.repository_loader import CodeDocument
+from config import get_settings
+
+
+LANGUAGE_MAP = {
+    "python": Language.PYTHON,
+    "javascript": Language.JS,
+    "typescript": Language.TS,
+    "java": Language.JAVA,
+    "go": Language.GO,
+}
+
+
+def split_code(doc : CodeDocument) -> list[dict]:
+
+    settings = get_settings()
+    lang_enum = LANGUAGE_MAP.get(doc.language.lower())
+
+    if lang_enum:
+        splitter = RecursiveCharacterTextSplitter.from_language(
+            language=lang_enum,
+            chunk_size=settings.chunk_size,
+            chunk_overlap=settings.chunk_overlap,
+        )
+    else:
+        splitter = RecursiveCharacterTextSplitter(
+            chunk_size=settings.chunk_size,
+            chunk_overlap=settings.chunk_overlap
+        )
+
+    raw_chunks = splitter.split_text(doc.content)
+    return [
+        attach_metadata(chunk, doc, idx)
+        for idx, chunk in enumerate(raw_chunks)
+    ]
+
+
+def attach_metadata(chunk_text: str, doc: CodeDocument, index: int) -> dict:
+    start_line, end_line = estimate_line_range(doc.content, chunk_text)
+    return {
+        "content": chunk_text,
+        "file_path": doc.file_path,
+        "language": doc.language,
+        "chunk_index": index,
+        "start_line": start_line,
+        "end_line": end_line,
+    }
+
+
+def estimate_line_range(full_text: str, chunk_text: str) -> tuple[int, int]:
+    offset = full_text.find(chunk_text)
+    if offset == -1:
+        return (0, 0)
+    start_line = full_text[:offset].count("\n") + 1
+    end_line = start_line + chunk_text.count("\n")
+    return (start_line, end_line)
+
+
+
+