fix: normalize logging, error handling, and streaming consistency

.dockerignore

@@ -0,0 +1,17 @@
+.venv/
+venv/
+__pycache__/
+*.pyc
+.git/
+.gitignore
+.env
+qdrant_data/
+data/document_registry.json
+tests/
+reports/
+datasets/
+ui/
+*.md
+.python/
+*.log
+fastembed_cache/

README.md

@@ -21,7 +21,7 @@ pip install -r requirements.txt
 3. Ingest the sample docs:
 
 ```bash
-python app/engine/ingestion.py
+python -m app.engine.ingestion ingest
 ```
 
 4. Start the backend:
@@ -43,7 +43,7 @@ Backend docs run at `http://127.0.0.1:8000/docs`; the Streamlit app runs at `htt
 ```bash
 docker-compose up --build -d
 docker exec -it $(docker-compose ps -q ollama) ollama run llama3
-docker exec -it $(docker-compose ps -q backend) python app/engine/ingestion.py
+docker exec -it $(docker-compose ps -q backend) python -m app.engine.ingestion ingest
 ```
 
 ## Suggested Commit Roadmap

app/auth/security.py

@@ -1,7 +1,8 @@
-from fastapi import Header, HTTPException
+from fastapi import Header
 
 from app.auth.models import UserContext
 from app.core.config import settings
+from app.core.errors import CopilotError
 
 
 def resolve_user(x_api_key: str | None = Header(default=None)) -> UserContext:
@@ -11,9 +12,9 @@ def resolve_user(x_api_key: str | None = Header(default=None)) -> UserContext:
         return UserContext(role="admin", user_id="admin")
     if x_api_key == settings.USER_API_KEY:
         return UserContext(role="user", user_id="user")
-    raise HTTPException(status_code=401, detail="Invalid or missing API key.")
+    raise CopilotError("Invalid or missing API key.", status_code=401)
 
 
 def require_admin(user: UserContext) -> None:
     if user.role != "admin":
-        raise HTTPException(status_code=403, detail="Admin role required.")
+        raise CopilotError("Admin role required.", status_code=403)

app/engine/ingestion.py

@@ -5,6 +5,8 @@ from pathlib import Path
 
 from langchain_core.documents import Document
 
+from app.core.logging import logger
+
 from app.engine.chunking import chunk_documents
 from app.engine.document_registry import (
     file_hash,
@@ -19,7 +21,7 @@ from app.guardrails.document import filter_malicious_documents
 
 
 def ingest_documents(data_dir: str = "data/docs", force: bool = False) -> None:
-    print(f"Loading documents from {data_dir}...")
+    logger.info(f"Loading documents from {data_dir}...")
     root = Path(data_dir)
     root.mkdir(parents=True, exist_ok=True)
 
@@ -36,14 +38,14 @@ def ingest_documents(data_dir: str = "data/docs", force: bool = False) -> None:
 
         content_hash = file_hash(path)
         if not force and hash_exists(content_hash, registry):
-            print(f"Skipping unchanged document: {path.name}")
+            logger.info(f"Skipping unchanged document: {path.name}")
             continue
 
         doc_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{path.name}:{content_hash}"))
         loaded = load_document(path)
         loaded, flagged = filter_malicious_documents(loaded)
         for flagged_source in flagged:
-            print(f"Skipping document with suspicious instructions: {flagged_source}")
+            logger.info(f"Skipping document with suspicious instructions: {flagged_source}")
         if not loaded:
             continue
         for document in loaded:
@@ -53,11 +55,11 @@ def ingest_documents(data_dir: str = "data/docs", force: bool = False) -> None:
         doc_id_by_source[str(path)] = doc_id
 
     if not source_documents:
-        print("No new documents found.")
+        logger.info("No new documents found.")
         return
 
     chunks = chunk_documents(source_documents)
-    print(f"Split documents into {len(chunks)} chunks.")
+    logger.info(f"Split documents into {len(chunks)} chunks.")
 
     index_documents(chunks, force_recreate=force)
 
@@ -73,33 +75,33 @@ def ingest_documents(data_dir: str = "data/docs", force: bool = False) -> None:
             registry=registry,
         )
     save_registry(registry)
-    print("Ingestion complete. Hybrid index is built.")
+    logger.info("Ingestion complete. Hybrid index is built.")
 
 
 def list_documents() -> None:
     registry = load_registry()
     if not registry:
-        print("No indexed documents found.")
+        logger.info("No indexed documents found.")
         return
     for record in registry.values():
-        print(f"{record['doc_id']} | {record['source']} | chunks={record['chunk_count']}")
+        logger.info(f"{record['doc_id']} | {record['source']} | chunks={record['chunk_count']}")
 
 
 def delete_indexed_document(doc_id: str) -> None:
     registry = load_registry()
     if doc_id not in registry:
-        print(f"Document not found: {doc_id}")
+        logger.info(f"Document not found: {doc_id}")
         return
     delete_document(doc_id)
     del registry[doc_id]
     save_registry(registry)
-    print(f"Deleted document: {doc_id}")
+    logger.info(f"Deleted document: {doc_id}")
 
 
 def reset_index() -> None:
     reset_collection()
     save_registry({})
-    print("Vector collection and document registry reset.")
+    logger.info("Vector collection and document registry reset.")
 
 
 def build_parser() -> argparse.ArgumentParser:

app/graph/workflow.py

@@ -6,6 +6,7 @@ from langchain_ollama import ChatOllama
 from langgraph.graph import START, END, StateGraph
 
 from app.core.config import settings
+from app.core.logging import logger
 from app.engine.context_builder import build_context, source_citations
 from app.engine.retriever import retrieve_documents
 
@@ -20,14 +21,14 @@ llm = ChatOllama(model=settings.OLLAMA_MODEL, temperature=0, base_url=settings.O
 llm_json = ChatOllama(model=settings.OLLAMA_MODEL, temperature=0, format="json", base_url=settings.OLLAMA_BASE_URL)
 
 def retrieve(state: GraphState):
-    print("--- NODE: RETRIEVE DOCS ---")
+    logger.info("NODE: RETRIEVE DOCS")
     question = state["question"]
     run_count = state.get("run_count", 0)
     documents = retrieve_documents(question)
     return {"documents": documents, "sources": source_citations(documents), "question": question, "run_count": run_count}
 
 def grade_documents(state: GraphState):
-    print("--- NODE: GRADE DOCUMENT RELEVANCE ---")
+    logger.info("NODE: GRADE DOCUMENT RELEVANCE")
     question = state["question"]
     documents = state.get("documents", [])
     
@@ -54,7 +55,7 @@ def grade_documents(state: GraphState):
     return {"documents": filtered_docs}
 
 def generate(state: GraphState):
-    print("--- NODE: GENERATE ANSWER ---")
+    logger.info("NODE: GENERATE ANSWER")
     question = state["question"]
     documents = state["documents"]
     run_count = state.get("run_count", 0) + 1
@@ -73,9 +74,9 @@ def generate(state: GraphState):
 
 def decide_to_generate(state: GraphState):
     if not state["documents"]:
-        print("--- ROUTE: ALL DOCS IRRELEVANT ---")
+        logger.info("ROUTE: ALL DOCS IRRELEVANT")
         return "end"
-    print("--- ROUTE: RELEVANT DOCS FOUND ---")
+    logger.info("ROUTE: RELEVANT DOCS FOUND")
     return "generate"
 
 def check_hallucinations(state: GraphState):
@@ -84,7 +85,7 @@ def check_hallucinations(state: GraphState):
     run_count = state["run_count"]
     
     if run_count >= 3:
-        print("--- ROUTE: MAX RETRIES REACHED ---")
+        logger.info("ROUTE: MAX RETRIES REACHED")
         return "end"
         
     context = build_context(documents)
@@ -105,9 +106,9 @@ def check_hallucinations(state: GraphState):
         grade = "yes"
         
     if grade.lower() == "yes":
-        print("--- ROUTE: GROUNDED ---")
+        logger.info("ROUTE: GROUNDED")
         return "end"
-    print("--- ROUTE: HALLUCINATION DETECTED ---")
+    logger.info("ROUTE: HALLUCINATION DETECTED")
     return "regenerate"
 
 def compile_workflow():

app/guardrails/input.py

@@ -2,9 +2,8 @@ import re
 import time
 from collections import defaultdict, deque
 
-from fastapi import HTTPException
-
 from app.core.config import settings
+from app.core.errors import CopilotError
 
 
 _requests_by_client: dict[str, deque[float]] = defaultdict(deque)
@@ -24,13 +23,13 @@ PROMPT_INJECTION_PATTERNS = [
 
 def validate_query(query: str) -> None:
     if not query.strip():
-        raise HTTPException(status_code=400, detail="Query cannot be empty.")
+        raise CopilotError("Query cannot be empty.", status_code=400)
     if len(query) > settings.MAX_QUERY_LENGTH:
-        raise HTTPException(status_code=400, detail="Query is too long.")
+        raise CopilotError("Query is too long.", status_code=400)
     normalized = query.lower()
     for pattern in PROMPT_INJECTION_PATTERNS:
         if pattern in normalized:
-            raise HTTPException(status_code=400, detail="Prompt injection attempt detected.")
+            raise CopilotError("Prompt injection attempt detected.", status_code=400)
 
 
 def enforce_rate_limit(client_id: str) -> None:
@@ -41,7 +40,7 @@ def enforce_rate_limit(client_id: str) -> None:
     while bucket and now - bucket[0] > 60:
         bucket.popleft()
     if len(bucket) >= settings.RATE_LIMIT_PER_MINUTE:
-        raise HTTPException(status_code=429, detail="Rate limit exceeded.")
+        raise CopilotError("Rate limit exceeded.", status_code=429)
     bucket.append(now)
 
 

app/main.py

@@ -1,6 +1,9 @@
 import asyncio
-from fastapi import Depends, FastAPI, HTTPException, Request
-from fastapi.responses import StreamingResponse
+from pathlib import Path
+
+from fastapi import Depends, FastAPI, File, Request, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel
 from guardrails import Guard
 from langchain_core.prompts import PromptTemplate
@@ -10,6 +13,7 @@ from app.auth.models import LoginRequest, LoginResponse, UserContext
 from app.auth.security import require_admin, resolve_user
 from app.core.config import settings
 from app.core.dependencies import check_ollama, check_qdrant
+from app.core.errors import CopilotError
 from app.core.logging import configure_logging, logger
 from app.engine.document_registry import load_registry
 from app.engine.ingestion import delete_indexed_document, ingest_documents, reset_index
@@ -22,6 +26,22 @@ from app.observability.metrics import RequestMetrics, log_request_metrics, timed
 
 configure_logging()
 app = FastAPI(title=settings.PROJECT_NAME)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.exception_handler(CopilotError)
+async def copilot_error_handler(request: Request, exc: CopilotError):
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"detail": exc.message},
+    )
+
 rag_agent = compile_workflow()
 input_guard = Guard().use(DetectPromptInjection, on_fail="exception")
 
@@ -64,7 +84,7 @@ async def login_endpoint(request: LoginRequest):
         return LoginResponse(role="admin")
     if request.api_key == settings.USER_API_KEY:
         return LoginResponse(role="user")
-    raise HTTPException(status_code=401, detail="Invalid API key.")
+    raise CopilotError("Invalid API key.", status_code=401)
 
 @app.get("/documents")
 async def documents_endpoint(user: UserContext = Depends(resolve_user)):
@@ -76,6 +96,19 @@ async def admin_ingest_endpoint(request: IngestionRequest, user: UserContext = D
     ingest_documents(data_dir=request.data_dir, force=request.force)
     return {"status": "ok", "message": "Ingestion completed."}
 
+@app.post("/admin/upload")
+async def admin_upload_endpoint(files: list[UploadFile] = File(...), user: UserContext = Depends(resolve_user)):
+    require_admin(user)
+    target_dir = Path(settings.DATA_DIR)
+    target_dir.mkdir(parents=True, exist_ok=True)
+    saved_files = []
+    for uploaded_file in files:
+        filename = Path(uploaded_file.filename or "uploaded.txt").name
+        target_path = target_dir / filename
+        target_path.write_bytes(await uploaded_file.read())
+        saved_files.append(filename)
+    return {"status": "ok", "saved_files": saved_files, "data_dir": str(target_dir)}
+
 @app.delete("/admin/documents/{doc_id}")
 async def admin_delete_document_endpoint(doc_id: str, user: UserContext = Depends(resolve_user)):
     require_admin(user)
@@ -97,7 +130,7 @@ async def chat_endpoint(request: ChatRequest, http_request: Request, user: UserC
         try:
             input_guard.validate(request.query)
         except Exception as e:
-            raise HTTPException(status_code=400, detail=str(getattr(e, "message", e)))
+            raise CopilotError(str(getattr(e, "message", e)), status_code=400)
 
     initial_state = {"question": request.query, "run_count": 0}
     try:
@@ -106,7 +139,7 @@ async def chat_endpoint(request: ChatRequest, http_request: Request, user: UserC
         answer = redact_sensitive_data(final_state.get("generation", "Unable to compile answer."))
         sources = final_state.get("sources", [])
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        raise CopilotError(str(e), status_code=500)
 
     log_request_metrics(metrics, route="/chat", sources=len(sources), model=settings.OLLAMA_MODEL)
     return ChatResponse(query=request.query, answer=answer, sources=sources)
@@ -119,7 +152,7 @@ async def chat_stream_endpoint(request: ChatRequest, http_request: Request, user
         try:
             input_guard.validate(request.query)
         except Exception as e:
-            raise HTTPException(status_code=400, detail=str(getattr(e, "message", e)))
+            raise CopilotError(str(getattr(e, "message", e)), status_code=400)
 
     async def token_generator():
         metrics = RequestMetrics()