Enhance environment configuration and API documentation for Milestone 11

- Update `.env.example` to reflect new application name, version, and additional configuration options for LLM providers.
- Revise `README.md` to improve architecture overview, quick start instructions, and use cases for the DocuAudit AI application.
- Add detailed specifications for API endpoints in `milestones.md`, including new features for multi-file uploads, query parameters, and audit logging.
- Refactor API routes to support enhanced query and ingestion functionalities, including user ID tracking and improved response structures.
- Update request and response models to align with new API specifications and ensure comprehensive coverage of expected behaviors.

.env.example

@@ -1,12 +1,48 @@
-APP_NAME=doc-audi-ai
-APP_VERSION=0.1.0
+# DocuAudit AI — environment template (see docs/DOCUAUDIT_AI_REQUIREMENTS.md)
+
+# LLM Provider: ollama | anthropic | openai | huggingface
 LLM_PROVIDER=ollama
-EMBEDDING_MODEL_NAME=nomic-embed-text
-OLLAMA_BASE_URL=http://localhost:11434
+
+# OpenAI (optional)
 OPENAI_API_KEY=
+OPENAI_MODEL=gpt-4o
+OPENAI_EMBEDDING_MODEL=text-embedding-3-small
+
+# Anthropic (optional)
+ANTHROPIC_API_KEY=
+ANTHROPIC_MODEL=claude-3-5-sonnet-20241022
+
+# Ollama (recommended local default)
+OLLAMA_BASE_URL=http://localhost:11434
+OLLAMA_CHAT_MODEL=llama3.1:8b
+OLLAMA_EMBEDDING_MODEL=nomic-embed-text
+
+# App
+APP_NAME=DocuAudit AI
+APP_VERSION=1.0.0
+DEBUG=false
+MAX_FILE_SIZE_MB=50
+# Spec name alias (optional; mapped to MAX_FILE_SIZE_MB in settings)
+MAX_UPLOAD_SIZE_MB=
+
+# ChromaDB
 CHROMA_PERSIST_DIRECTORY=./data/chroma
-JOBS_DB_PATH=./data/jobs.db
+CHROMA_PERSIST_DIR=
+CHROMA_COLLECTION_NAME=docuaudit_docs
+
+# Chunking
 CHUNK_SIZE=1000
-CHUNK_OVERLAP=150
-RETRIEVAL_K=4
-MAX_FILE_SIZE_MB=25
+CHUNK_OVERLAP=200
+
+# Retrieval default (overridable per request on /query/ask via top_k)
+TOP_K_RESULTS=5
+
+# Audit + jobs SQLite
+AUDIT_DB_PATH=./audit.db
+JOBS_DB_PATH=./data/jobs.db
+
+# Limits
+MAX_DOCUMENTS_PER_BATCH=100
+
+# Streamlit → API
+STREAMLIT_BACKEND_URL=http://localhost:8000

README.md

@@ -1,42 +1,138 @@
-# doc-Audi-ai
+# DocuAudit AI
 
+**DocuAudit AI** is a production-oriented FastAPI backend plus optional Streamlit UI for **multi-document RAG**: upload documents, build a Chroma vector index, ask grounded questions with citations, and retain a **SQLite audit trail** of every query.
 
-create requirements.txt & .env
+## Architecture
 
+```mermaid
+flowchart LR
+  subgraph ingest [Ingestion]
+    A[PDF / TXT / MD] --> B[Loader]
+    B --> C[Chunker]
+    C --> D[Embedder]
+    D --> E[(ChromaDB)]
+  end
+  subgraph query [Query path]
+    Q[User question] --> R[Semantic search]
+    R --> E
+    R --> T[Top-K chunks]
+    T --> L[LLM]
+    L --> U[Answer + citations]
+  end
+  U --> V[(SQLite audit)]
+```
 
-# 1. Setup environment
-uv venv --python 3.11.14
-uv init --python 3.11.14
-uv add requirements.txt
-uv pip install -r requirements.txt
-copy .env.example .env
+ASCII equivalent:
 
-Note for Intel Macs (x86_64):
-- `pyproject.toml` includes platform-specific pins for `onnxruntime` and `torch` to ensure `uv` resolves versions that have compatible wheels.
+```
+PDF Upload → Parser → Chunker → Embedder → ChromaDB
+                                              ↓
+User Query → Semantic Search → Top-K Chunks → LLM → Answer + Citations
+                                              ↓
+                                       Audit Log (SQLite)
+```
 
-Install Ollama
-curl -fsSL https://ollama.com/install.sh | sh
+## Use cases
 
-Pull required models
-ollama pull llama3.1:8b # LLM for answer generation (~2 GB) 
-ollama pull nomic-embed-text # Embedding model (~274 MB)
+- **Litigation document analysis** — trace claims to exact pages and filenames.
+- **Corporate finance review** — compare disclosures and filings under a consistent audit log.
+- **Investigation support** — bulk ingest, async jobs, and reproducible query history.
 
-Start Ollama server
-ollama serve &
+## Quick start (local, without Docker)
 
-Verify models are running
-curl http://localhost:11434/api/tags
+Docker and Compose are planned under **Milestone 12**. Until then, run the API with **uv** (or your preferred tool):
 
-### Start backend server- fastapi
 ```bash
-uv run uvicorn api.main:app --host 0.0.0.0 --port 8000
+git clone <repository-url> doc-Audi-ai
+cd doc-Audi-ai
+copy .env.example .env
+uv sync
+ollama pull llama3.1:8b
+ollama pull nomic-embed-text
+uv run uvicorn api.main:app --host 0.0.0.0 --port 8000 --reload
 ```
-### Start frontend server- Streamlit
+
+Optional UI:
+
 ```bash
-uv run streamlit run streamlit_app.py --server.address=0.0.0.0 --server.port=8501
+uv run streamlit run streamlit_app.py --server.port 8501 --server.address 0.0.0.0
 ```
 
-git diff --name-status --diff-filter=AM <from_commit_hash> <to_commit_hash>
+After **Milestone 12**, the intended one-command experience will be `docker compose up` for API (`localhost:8000`) and UI (`localhost:8501`).
+
+## API overview
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/health` | Liveness; returns configured app name and version |
+| POST | `/ingest/upload` | Multipart **`files`** (one or more); queues background ingest job |
+| POST | `/ingest/url` | JSON **`urls`** array (1–100); download and queue ingest |
+| GET | `/ingest/collections` | Lists collections with **`document_count`** and optional **`created_at`** |
+| DELETE | `/ingest/collection/{collection_name}` | Drops a collection; returns **`documents_removed`** |
+| GET | `/jobs` | Lists jobs with **`total`** count |
+| GET | `/jobs/{job_id}` | Job status with **`progress_percent`**, file counters, timestamps, **`errors`** |
+| POST | `/query/ask` | Grounded answer; request includes **`top_k`**, **`user_id`** |
+| POST | `/query/summarise` | Collection summary; distinct response shape (`summary`, `document_count`, …) |
+| POST | `/query` | Legacy alias of **`/query/ask`** |
+| GET | `/audit/logs` | Filterable audit index (`user_id`, `from_date`, `to_date`, pagination) |
+| GET | `/audit/logs/{query_id}` | Full stored answer and citations for one query |
+
+Interactive docs: `http://localhost:8000/docs`.
+
+## Sample request and response (`POST /query/ask`)
+
+Request:
+
+```json
+{
+  "question": "What were the key risk factors identified in the Q3 2023 financial report?",
+  "collection_name": "default",
+  "top_k": 5,
+  "user_id": "analyst_001"
+}
+```
+
+Response (shape; values depend on your documents and model):
+
+```json
+{
+  "query_id": "uuid-string",
+  "question": "What were the key risk factors identified in the Q3 2023 financial report?",
+  "answer": "… grounded text with citations …",
+  "sources": [
+    {
+      "document_name": "q3_financial_report.pdf",
+      "page_number": 12,
+      "chunk_text": "Key risk factors include …",
+      "relevance_score": 0.91
+    }
+  ],
+  "model_used": "llama3.1:8b",
+  "tokens_used": 0,
+  "response_time_ms": 1820,
+  "timestamp": "2026-05-03T12:00:00Z"
+}
+```
+
+## Design decisions
+
+- **Source citations** — High-stakes review requires every substantive claim to be tied to **document name** and **page** (where available), not a free-floating model monologue.
+- **Auditability** — Each ask/summarise persists **query id**, **user id**, timing, model id, token usage (when the provider exposes it), and serialized sources so regulators or counsel can reconstruct what the system returned.
+
+## Scale note
+
+Architecture is designed for **high-volume document ingestion** via **async background jobs** (FastAPI `BackgroundTasks`), persistent Chroma collections, and a stateless API tier that can be replicated once you add a shared vector store and job queue.
+
+## Tests
+
+```bash
+uv run pytest tests/ -q
+```
+
+## Configuration
+
+See **`.env.example`**. Common variables include `LLM_PROVIDER`, Ollama/OpenAI/Anthropic keys and models, `CHROMA_PERSIST_DIRECTORY`, `AUDIT_DB_PATH`, `JOBS_DB_PATH`, and upload limits (`MAX_FILE_SIZE_MB`; **`MAX_UPLOAD_SIZE_MB`** is accepted as an alias via settings normalization).
 
-git diff --name-status --diff-filter=AM  18ad0e6c94d041b1fd902e7f9b60113738eee1fa 0f2ee3afa124348adece82df0ff0e5a0943a7b8b
+## Specification
 
+Authoritative product and API shapes: **`docs/DOCUAUDIT_AI_REQUIREMENTS.md`**. Gap tracking: **`docs/REQUIREMENTS_IMPLEMENTATION_GAPS.md`**.

api/config.py

@@ -1,5 +1,7 @@
 from functools import lru_cache
-from pydantic import Field
+from typing import Any
+
+from pydantic import Field, model_validator
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -10,10 +12,30 @@ class Settings(BaseSettings):
         env_file_encoding="utf-8",
         extra="ignore",
         case_sensitive=False,
+        populate_by_name=True,
     )
 
-    app_name: str = Field(default="doc-audi-ai", description="The name of the application")
-    app_version: str = Field(default="0.1.0", description="The version of the application")
+    @model_validator(mode="before")
+    @classmethod
+    def _map_max_upload_env_alias(cls, data: Any) -> Any:
+        if not isinstance(data, dict):
+            return data
+        out = dict(data)
+        if out.get("max_file_size_mb") in (None, "") and out.get("max_upload_size_mb") not in (None, ""):
+            out["max_file_size_mb"] = out.pop("max_upload_size_mb")
+        elif "max_upload_size_mb" in out and "max_file_size_mb" not in out:
+            out["max_file_size_mb"] = out.pop("max_upload_size_mb")
+        return out
+
+    app_name: str = Field(default="DocuAudit AI", description="FastAPI title and product name")
+    app_version: str = Field(default="1.0.0", description="Application version")
+    app_description: str = Field(
+        default=(
+            "Multi-document RAG API for high-stakes consulting environments. "
+            "Every answer is grounded in source documents with full audit trails."
+        ),
+        description="OpenAPI /docs description",
+    )
     llm_provider: str = Field(default="ollama", description="Embedding provider")
     
     openai_api_key: str | None = Field(default=None, description="OpenAI API key")
@@ -38,12 +60,12 @@ class Settings(BaseSettings):
 
     chunk_size: int = Field(default=1000, ge=100, le=8000, description="Chunk size for splitting")
     chunk_overlap: int = Field(default=200, ge=0, le=2000, description="Chunk overlap for splitting")
-    top_k_results: int = Field(default=4, ge=1, le=20, description="Number of chunks to retrieve")
+    top_k_results: int = Field(default=5, ge=1, le=20, description="Default number of chunks to retrieve")
 
     audit_db_path: str = "./audit.db"
     jobs_db_path: str = Field(default="./data/jobs.db", description="SQLite path for ingest job tracking")
 
-    max_file_size_mb: int = Field(default=50, ge=1, le=200, description="Max upload file size")
+    max_file_size_mb: int = Field(default=50, ge=1, le=200, description="Max upload file size (MB)")
     max_documents_per_batch: int = Field(default=100, ge=1, le=1000, description="Max documents per batch")
 
 

api/main.py

@@ -4,13 +4,27 @@ import os
 os.environ.setdefault("ANONYMIZED_TELEMETRY", "FALSE")
 
 from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 
 from api.config import get_settings
 from storage.audit_store import init_audit_db
 from storage.job_store import init_jobs_db
 from .routes import audit, ingest, jobs, query
 
-app = FastAPI()
+_settings = get_settings()
+app = FastAPI(
+    title=_settings.app_name,
+    version=_settings.app_version,
+    description=_settings.app_description,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
 
 app.include_router(audit.router)
 app.include_router(ingest.router)
@@ -25,6 +39,12 @@ async def startup() -> None:
     await init_audit_db(settings.audit_db_path)
     await init_jobs_db(settings.jobs_db_path)
 
+
 @app.get("/health", tags=["Health"])
 def health() -> dict[str, str]:
-    return {"status": "ok","app_name": "doc-audi-ai", "version": "0.1.0"}
+    settings = get_settings()
+    return {
+        "status": "ok",
+        "app": settings.app_name,
+        "version": settings.app_version,
+    }

api/routes/audit.py

@@ -4,42 +4,54 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from api.config import get_settings
 from models.requests import AuditListParams
-from models.responses import AuditDetailResponse, AuditEvent, AuditListResponse
+from models.responses import AuditLogDetailResponse, AuditLogsResponse
 from storage.audit_store import get_audit_event, list_audit_events
 
 
 def _audit_list_params(
-    limit: Annotated[int, Query(ge=1, le=100)] = 10,
+    limit: Annotated[int, Query(ge=1, le=100)] = 50,
     offset: Annotated[int, Query(ge=0)] = 0,
+    user_id: Annotated[str | None, Query(max_length=256)] = None,
+    from_date: Annotated[str | None, Query(description="ISO 8601 lower bound")] = None,
+    to_date: Annotated[str | None, Query(description="ISO 8601 upper bound")] = None,
 ) -> AuditListParams:
-    return AuditListParams(limit=limit, offset=offset)
+    return AuditListParams(
+        limit=limit,
+        offset=offset,
+        user_id=user_id,
+        from_date=from_date,
+        to_date=to_date,
+    )
 
 
 router = APIRouter(prefix="/audit", tags=["audit"])
 
 
-@router.get("/logs", response_model=AuditListResponse)
+@router.get("/logs", response_model=AuditLogsResponse)
 async def audit_logs(
     params: Annotated[AuditListParams, Depends(_audit_list_params)],
-) -> AuditListResponse:
+) -> AuditLogsResponse:
     settings = get_settings()
-    rows = await list_audit_events(settings.audit_db_path, limit=params.limit, offset=params.offset)
-    events = [AuditEvent.model_validate(row) for row in rows]
-    return AuditListResponse(
-        status="success",
-        message=f"Returned {len(events)} audit event(s).",
-        events=events,
+    logs, total = await list_audit_events(
+        settings.audit_db_path,
+        limit=params.limit,
+        offset=params.offset,
+        user_id=params.user_id,
+        from_date=params.from_date,
+        to_date=params.to_date,
+    )
+    return AuditLogsResponse(
+        logs=logs,
+        total=total,
+        limit=params.limit,
+        offset=params.offset,
     )
 
 
-@router.get("/logs/{query_id}", response_model=AuditDetailResponse)
-async def audit_log_detail(query_id: str) -> AuditDetailResponse:
+@router.get("/logs/{query_id}", response_model=AuditLogDetailResponse)
+async def audit_log_detail(query_id: str) -> AuditLogDetailResponse:
     settings = get_settings()
     event = await get_audit_event(settings.audit_db_path, query_id)
     if event is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Audit event not found.")
-    return AuditDetailResponse(
-        status="success",
-        message="Audit event retrieved.",
-        event=event,
-    )
+    return event

api/routes/ingest.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import Annotated
@@ -7,14 +8,20 @@ import httpx
 from fastapi import APIRouter, BackgroundTasks, File, Form, HTTPException, UploadFile, status
 
 from api.config import get_settings
-from models.requests import IngestUrlRequest
+from models.requests import URLIngestRequest
 from models.responses import (
+    CollectionItem,
     IngestCollectionsResponse,
     IngestDeleteCollectionResponse,
     IngestUploadResponse,
-    CollectionItem,
+    UrlIngestResponse,
+)
+from rag.vector_store import (
+    collection_created_at,
+    collection_document_count,
+    delete_collection,
+    list_collection_names,
 )
-from rag.vector_store import delete_collection, list_collection_names
 from storage.job_store import create_ingest_job
 from workers.ingest_worker import run_ingest_job
 
@@ -86,7 +93,7 @@ async def _download_url_to_temp(url: str, max_bytes: int) -> tuple[str, str]:
 
     timeout = httpx.Timeout(60.0, connect=10.0)
     limits = httpx.Limits(max_keepalive_connections=5, max_connections=5)
-    headers = {"User-Agent": "doc-audi-ai/ingest"}
+    headers = {"User-Agent": "docuaudit-ai/ingest"}
 
     try:
         async with httpx.AsyncClient(timeout=timeout, limits=limits, follow_redirects=True) as client:
@@ -132,97 +139,131 @@ async def _download_url_to_temp(url: str, max_bytes: int) -> tuple[str, str]:
     return temp_path, display_name
 
 
+def _parse_created_at(raw: str | None) -> datetime | None:
+    if not raw:
+        return None
+    s = raw.strip()
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    try:
+        dt = datetime.fromisoformat(s)
+        if dt.tzinfo is None:
+            return dt.replace(tzinfo=timezone.utc)
+        return dt
+    except ValueError:
+        return None
+
+
 @router.post("/upload", response_model=IngestUploadResponse)
 async def upload_endpoint(
     background_tasks: BackgroundTasks,
-    file: UploadFile = File(..., description="PDF/TXT/MD document to ingest"),
+    files: list[UploadFile] = File(..., description="One or more PDF, TXT, or MD files"),
     collection_name: Annotated[str, Form(min_length=1, max_length=256)] = "default",
 ) -> IngestUploadResponse:
     settings = get_settings()
-    max_bytes = settings.max_file_size_mb * 1024 * 1024
-    suffix = _validate_file(file, max_bytes)
-    display_name = (file.filename or "upload").strip()
+    if not files:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="At least one file is required.")
+    if len(files) > settings.max_documents_per_batch:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Too many files in one request (max {settings.max_documents_per_batch}).",
+        )
 
-    temp_path = ""
+    max_bytes = settings.max_file_size_mb * 1024 * 1024
+    temp_paths: list[tuple[str, str]] = []
+    filenames: list[str] = []
     try:
-        file_bytes = await file.read()
-        with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
-            temp_path = tmp.name
-            tmp.write(file_bytes)
+        for file in files:
+            suffix = _validate_file(file, max_bytes)
+            display_name = (file.filename or "upload").strip()
+            file_bytes = await file.read()
+            with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+                tmp.write(file_bytes)
+                temp_paths.append((tmp.name, display_name))
+            filenames.append(display_name)
+            await file.close()
 
         job_id = await create_ingest_job(
             settings.jobs_db_path,
-            collection_name=collection_name,
-            filename=display_name,
+            collection_name=collection_name.strip(),
+            filenames=filenames,
         )
 
         background_tasks.add_task(
             run_ingest_job,
             job_id,
-            temp_path,
-            collection_name,
+            temp_paths,
+            collection_name.strip(),
             settings.jobs_db_path,
             settings.chroma_persist_directory,
         )
 
         return IngestUploadResponse(
-            status="queued",
-            message=f"Ingestion job accepted. Poll GET /jobs/{job_id} for status.",
             job_id=job_id,
-            document_ids=[],
+            status="queued",
+            total_files=len(filenames),
+            filenames=filenames,
+            message=f"Documents queued for processing. Poll /jobs/{job_id} for status.",
         )
     except HTTPException:
-        if temp_path:
-            Path(temp_path).unlink(missing_ok=True)
+        for path, _ in temp_paths:
+            Path(path).unlink(missing_ok=True)
         raise
     except Exception as exc:
-        if temp_path:
-            Path(temp_path).unlink(missing_ok=True)
+        for path, _ in temp_paths:
+            Path(path).unlink(missing_ok=True)
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
-    finally:
-        await file.close()
 
 
-@router.post("/url", response_model=IngestUploadResponse)
+@router.post("/url", response_model=UrlIngestResponse)
 async def ingest_url_endpoint(
     background_tasks: BackgroundTasks,
-    payload: IngestUrlRequest,
-) -> IngestUploadResponse:
+    payload: URLIngestRequest,
+) -> UrlIngestResponse:
     settings = get_settings()
     max_bytes = settings.max_file_size_mb * 1024 * 1024
-    url_str = str(payload.url).strip()
-    temp_path = ""
+    url_strings = [str(u).strip() for u in payload.urls]
+    if len(url_strings) > settings.max_documents_per_batch:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Too many URLs in one request (max {settings.max_documents_per_batch}).",
+        )
+
+    downloaded: list[tuple[str, str]] = []
     try:
-        temp_path, display_name = await _download_url_to_temp(url_str, max_bytes)
+        for url_str in url_strings:
+            temp_path, display_name = await _download_url_to_temp(url_str, max_bytes)
+            downloaded.append((temp_path, display_name))
 
+        coll = (payload.collection_name or "default").strip()
         job_id = await create_ingest_job(
             settings.jobs_db_path,
-            collection_name=payload.collection_name,
-            filename=display_name,
+            collection_name=coll,
+            filenames=[name for _, name in downloaded],
         )
 
         background_tasks.add_task(
             run_ingest_job,
             job_id,
-            temp_path,
-            payload.collection_name,
+            downloaded,
+            coll,
             settings.jobs_db_path,
             settings.chroma_persist_directory,
         )
 
-        return IngestUploadResponse(
-            status="queued",
-            message=f"Ingestion job accepted. Poll GET /jobs/{job_id} for status.",
+        return UrlIngestResponse(
             job_id=job_id,
-            document_ids=[],
+            status="queued",
+            total_urls=len(downloaded),
+            message="URLs queued for download and processing.",
         )
     except HTTPException:
-        if temp_path:
-            Path(temp_path).unlink(missing_ok=True)
+        for path, _ in downloaded:
+            Path(path).unlink(missing_ok=True)
         raise
     except Exception as exc:
-        if temp_path:
-            Path(temp_path).unlink(missing_ok=True)
+        for path, _ in downloaded:
+            Path(path).unlink(missing_ok=True)
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
 
 
@@ -233,12 +274,18 @@ async def list_collections_endpoint() -> IngestCollectionsResponse:
         names = list_collection_names(settings.chroma_persist_directory)
     except Exception as exc:
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
-    items = [CollectionItem(name=n) for n in names]
-    return IngestCollectionsResponse(
-        status="success",
-        message=f"Found {len(items)} collection(s).",
-        collections=items,
-    )
+    items: list[CollectionItem] = []
+    for n in names:
+        cnt = collection_document_count(settings.chroma_persist_directory, n)
+        raw_created = collection_created_at(settings.chroma_persist_directory, n)
+        items.append(
+            CollectionItem(
+                name=n,
+                document_count=cnt,
+                created_at=_parse_created_at(raw_created),
+            )
+        )
+    return IngestCollectionsResponse(collections=items, total=len(items))
 
 
 @router.delete("/collection/{collection_name}", response_model=IngestDeleteCollectionResponse)
@@ -251,13 +298,12 @@ async def delete_collection_endpoint(collection_name: str) -> IngestDeleteCollec
         existing = list_collection_names(settings.chroma_persist_directory)
         if name not in existing:
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Collection not found.")
-        delete_collection(settings.chroma_persist_directory, name)
+        removed = delete_collection(settings.chroma_persist_directory, name)
     except HTTPException:
         raise
     except Exception as exc:
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
     return IngestDeleteCollectionResponse(
-        status="success",
-        message=f"Deleted collection '{name}'.",
-        collection_name=name,
+        message=f"Collection '{name}' deleted successfully.",
+        documents_removed=removed,
     )

api/routes/jobs.py

@@ -4,8 +4,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from api.config import get_settings
 from models.requests import JobsListParams
-from models.responses import IngestJobDetailResponse, JobListResponse, JobSummary
-from storage.job_store import get_ingest_job, list_ingest_jobs
+from models.responses import JobListResponse, JobStatusResponse
+from storage.job_store import get_job_status, list_ingest_jobs
 
 
 def _jobs_list_params(
@@ -23,27 +23,18 @@ async def list_jobs(
     params: Annotated[JobsListParams, Depends(_jobs_list_params)],
 ) -> JobListResponse:
     settings = get_settings()
-    rows = await list_ingest_jobs(
+    jobs, total = await list_ingest_jobs(
         settings.jobs_db_path,
         limit=params.limit,
         offset=params.offset,
     )
-    jobs = [JobSummary.model_validate(row) for row in rows]
-    return JobListResponse(
-        status="success",
-        message=f"Returned {len(jobs)} job(s).",
-        jobs=jobs,
-    )
+    return JobListResponse(jobs=jobs, total=total)
 
 
-@router.get("/jobs/{job_id}", response_model=IngestJobDetailResponse)
-async def get_job(job_id: str) -> IngestJobDetailResponse:
+@router.get("/jobs/{job_id}", response_model=JobStatusResponse)
+async def get_job(job_id: str) -> JobStatusResponse:
     settings = get_settings()
-    job = await get_ingest_job(settings.jobs_db_path, job_id)
+    job = await get_job_status(settings.jobs_db_path, job_id)
     if job is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found.")
-    return IngestJobDetailResponse(
-        status="success",
-        message="Job found.",
-        job=job,
-    )
+    return job

api/routes/query.py

@@ -1,8 +1,12 @@
+import time
+from datetime import datetime, timezone
+from uuid import uuid4
+
 from fastapi import APIRouter, HTTPException, status
 
-from api.config import get_settings
+from api.config import Settings, get_settings
 from models.requests import QueryRequest, SummariseRequest
-from models.responses import QueryResponse, QueryResultItem, QuerySourceItem
+from models.responses import AskQueryResponse, SourceCitation, SummariseQueryResponse
 from rag.embedder import create_embedding_function
 from rag.retriever import (
     SUMMARY_RETRIEVAL_QUERY,
@@ -11,117 +15,153 @@ from rag.retriever import (
     retrieve_chunks,
     summarise_with_grounding,
 )
-from rag.vector_store import get_vector_store
+from rag.vector_store import collection_document_count, get_vector_store
 from storage.audit_store import persist_query_audit
 
 router = APIRouter(prefix="/query", tags=["query"])
 
 
-def _response_from_chunks(
-    *,
-    collection_name: str,
-    chunks: list[RetrievedChunk],
-    answer: str,
-    message: str,
-) -> QueryResponse:
-    results = [QueryResultItem(text=chunk.text, score=chunk.score) for chunk in chunks]
-    sources = [
-        QuerySourceItem(
-            source=chunk.source,
-            page=chunk.page,
-            chunk_index=chunk.chunk_index,
-            score=chunk.score,
-            excerpt=chunk.text[:280],
-        )
-        for chunk in chunks
-    ]
-    return QueryResponse(
-        status="success",
-        message=message,
-        answer=answer,
-        sources=sources,
-        results=results,
-    )
+def _model_used_label(settings: Settings) -> str:
+    provider = settings.llm_provider.lower()
+    if provider == "openai":
+        return settings.openai_model
+    if provider == "ollama":
+        return settings.ollama_chat_model
+    if provider == "anthropic":
+        return settings.anthropic_model
+    if provider == "huggingface":
+        return settings.huggingface_model
+    return f"{provider}:unknown"
 
 
-@router.post("/ask", response_model=QueryResponse)
-async def ask_endpoint(payload: QueryRequest) -> QueryResponse:
-    settings = get_settings()
-    try:
-        embedding_function = create_embedding_function()
-        vector_store = get_vector_store(
-            persist_directory=settings.chroma_persist_directory,
-            collection_name=payload.collection_name,
-            embedding_function=embedding_function,
+def _chunks_to_citations(chunks: list[RetrievedChunk]) -> list[SourceCitation]:
+    citations: list[SourceCitation] = []
+    for chunk in chunks:
+        page = chunk.page if chunk.page is not None else 0
+        score = float(chunk.score) if chunk.score is not None else 0.0
+        citations.append(
+            SourceCitation(
+                document_name=chunk.source or "unknown",
+                page_number=page,
+                chunk_text=chunk.text,
+                relevance_score=score,
+            )
         )
-        chunks = retrieve_chunks(vector_store, payload.question, settings.top_k_results)
-        answer = answer_with_grounding(settings, payload.question, chunks)
-    except Exception as exc:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
+    return citations
 
-    response = _response_from_chunks(
-        collection_name=payload.collection_name,
-        chunks=chunks,
+
+async def _run_ask(
+    settings: Settings,
+    payload: QueryRequest,
+) -> AskQueryResponse:
+    top_k = payload.top_k
+    t0 = time.perf_counter()
+    embedding_function = create_embedding_function()
+    vector_store = get_vector_store(
+        persist_directory=settings.chroma_persist_directory,
+        collection_name=payload.collection_name or "default",
+        embedding_function=embedding_function,
+    )
+    chunks = retrieve_chunks(vector_store, payload.question, top_k)
+    answer, tokens_used = answer_with_grounding(settings, payload.question, chunks)
+    elapsed_ms = int((time.perf_counter() - t0) * 1000)
+    citations = _chunks_to_citations(chunks)
+    query_id = str(uuid4())
+    ts = datetime.now(timezone.utc)
+    response = AskQueryResponse(
+        query_id=query_id,
+        question=payload.question,
         answer=answer,
-        message=(
-            f"Retrieved {len(chunks)} chunks from '{payload.collection_name}' and generated a grounded answer."
-        ),
+        sources=citations,
+        model_used=_model_used_label(settings),
+        tokens_used=tokens_used,
+        response_time_ms=elapsed_ms,
+        timestamp=ts,
+    )
+    await persist_query_audit(
+        settings.audit_db_path,
+        query_id=query_id,
+        action="query",
+        user_id=payload.user_id,
+        question=payload.question,
+        collection_name=payload.collection_name or "default",
+        answer=answer,
+        sources=citations,
+        model_used=response.model_used,
+        tokens_used=tokens_used,
+        response_time_ms=elapsed_ms,
+        kind="ask",
     )
-    try:
-        await persist_query_audit(
-            settings.audit_db_path,
-            action="query",
-            question=payload.question,
-            collection_name=payload.collection_name,
-            response=response,
-        )
-    except Exception as exc:
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
     return response
 
 
-@router.post("/summarise", response_model=QueryResponse)
-async def summarise_endpoint(payload: SummariseRequest) -> QueryResponse:
-    settings = get_settings()
+async def _run_summarise(
+    settings: Settings,
+    payload: SummariseRequest,
+) -> SummariseQueryResponse:
+    top_k = settings.top_k_results
     retrieval_query = (payload.focus or "").strip() or SUMMARY_RETRIEVAL_QUERY
     audit_question = payload.focus.strip() if payload.focus and payload.focus.strip() else "Summarise collection"
+    t0 = time.perf_counter()
+    embedding_function = create_embedding_function()
+    vector_store = get_vector_store(
+        persist_directory=settings.chroma_persist_directory,
+        collection_name=payload.collection_name,
+        embedding_function=embedding_function,
+    )
+    chunks = retrieve_chunks(vector_store, retrieval_query, top_k)
+    summary, tokens_used = summarise_with_grounding(settings, focus=payload.focus, chunks=chunks)
+    elapsed_ms = int((time.perf_counter() - t0) * 1000)
+    citations = _chunks_to_citations(chunks)
+    doc_count = collection_document_count(settings.chroma_persist_directory, payload.collection_name)
+    query_id = str(uuid4())
+    ts = datetime.now(timezone.utc)
+    response = SummariseQueryResponse(
+        query_id=query_id,
+        summary=summary,
+        document_count=doc_count,
+        sources=citations,
+        timestamp=ts,
+    )
+    await persist_query_audit(
+        settings.audit_db_path,
+        query_id=query_id,
+        action="summarise",
+        user_id=payload.user_id,
+        question=audit_question,
+        collection_name=payload.collection_name,
+        answer=summary,
+        sources=citations,
+        model_used=_model_used_label(settings),
+        tokens_used=tokens_used,
+        response_time_ms=elapsed_ms,
+        kind="summarise",
+    )
+    return response
+
+
+@router.post("/ask", response_model=AskQueryResponse)
+async def ask_endpoint(payload: QueryRequest) -> AskQueryResponse:
+    settings = get_settings()
     try:
-        embedding_function = create_embedding_function()
-        vector_store = get_vector_store(
-            persist_directory=settings.chroma_persist_directory,
-            collection_name=payload.collection_name,
-            embedding_function=embedding_function,
-        )
-        chunks = retrieve_chunks(vector_store, retrieval_query, settings.top_k_results)
-        answer = summarise_with_grounding(settings, focus=payload.focus, chunks=chunks)
+        return await _run_ask(settings, payload)
     except Exception as exc:
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
 
-    response = _response_from_chunks(
-        collection_name=payload.collection_name,
-        chunks=chunks,
-        answer=answer,
-        message=(
-            f"Retrieved {len(chunks)} chunks from '{payload.collection_name}' and generated a grounded summary."
-        ),
-    )
+
+@router.post("/summarise", response_model=SummariseQueryResponse)
+async def summarise_endpoint(payload: SummariseRequest) -> SummariseQueryResponse:
+    settings = get_settings()
     try:
-        await persist_query_audit(
-            settings.audit_db_path,
-            action="summarise",
-            question=audit_question,
-            collection_name=payload.collection_name,
-            response=response,
-        )
+        return await _run_summarise(settings, payload)
     except Exception as exc:
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc
-    return response
 
 
 legacy_query_router = APIRouter(tags=["query"])
 
 
-@legacy_query_router.post("/query", response_model=QueryResponse)
-async def query_post_compat(payload: QueryRequest) -> QueryResponse:
+@legacy_query_router.post("/query", response_model=AskQueryResponse)
+async def query_post_compat(payload: QueryRequest) -> AskQueryResponse:
     """Same behavior as POST /query/ask; kept for older clients and docs that used POST /query."""
     return await ask_endpoint(payload)

models/requests.py

@@ -1,16 +1,20 @@
+from typing import Optional
+
 from pydantic import BaseModel, ConfigDict, Field, HttpUrl
 
 
 class QueryRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    question: str = Field(min_length=1, max_length=8000, description="The question to ask the document")
-    collection_name: str = Field(
+    question: str = Field(min_length=5, max_length=2000, description="Natural language question")
+    collection_name: Optional[str] = Field(
         default="default",
         min_length=1,
         max_length=256,
-        description="The name of the collection to ask the question from",
+        description="Chroma collection to search",
     )
+    top_k: int = Field(default=5, ge=1, le=20, description="Number of chunks to retrieve")
+    user_id: str = Field(default="anonymous", max_length=256, description="Caller id for audit filtering")
 
 
 class SummariseRequest(BaseModel):
@@ -20,37 +24,50 @@ class SummariseRequest(BaseModel):
         default="default",
         min_length=1,
         max_length=256,
-        description="Chroma collection to summarise from",
+        description="Chroma collection to summarise",
     )
     focus: str | None = Field(
         default=None,
         max_length=8000,
-        description="Optional angle or scope for retrieval and the summary (e.g. 'contract payment terms')",
+        description="Optional angle or scope for retrieval and the summary",
     )
+    user_id: str = Field(default="anonymous", max_length=256, description="Caller id for audit filtering")
 
 
-class IngestUrlRequest(BaseModel):
+class URLIngestRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    url: HttpUrl = Field(description="HTTP(S) URL to a PDF, TXT, or Markdown document")
-    collection_name: str = Field(
+    urls: list[HttpUrl] = Field(
+        min_length=1,
+        max_length=100,
+        description="One or more HTTP(S) URLs to PDF, TXT, or Markdown documents",
+    )
+    collection_name: Optional[str] = Field(
         default="default",
         min_length=1,
         max_length=256,
         description="Target Chroma collection name",
     )
 
-class IngestUploadRequest(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-    collection_name: str = Field(default="default", min_length=1, max_length=256, description="The name of the collection to upload the document to")
-    filename: str = Field(min_length=1, max_length=1024, description="The name of the file to upload")
 
 class JobsListParams(BaseModel):
     model_config = ConfigDict(extra="forbid")
-    limit: int = Field(default=10, ge=1, le=100, description="The limit of the jobs to list")
-    offset: int = Field(default=0, ge=0, description="The offset of the jobs to list")
+
+    limit: int = Field(default=10, ge=1, le=100, description="Max jobs to return")
+    offset: int = Field(default=0, ge=0, description="Offset for pagination")
+
 
 class AuditListParams(BaseModel):
     model_config = ConfigDict(extra="forbid")
-    limit: int = Field(default=10, ge=1, le=100, description="The limit of the audit to list")
-    offset: int = Field(default=0, ge=0, description="The offset of the audit to list")
+
+    limit: int = Field(default=50, ge=1, le=100, description="Max log entries to return")
+    offset: int = Field(default=0, ge=0, description="Offset for pagination")
+    user_id: str | None = Field(default=None, max_length=256, description="Filter by user id")
+    from_date: str | None = Field(
+        default=None,
+        description="ISO 8601 datetime lower bound (inclusive) on timestamp",
+    )
+    to_date: str | None = Field(
+        default=None,
+        description="ISO 8601 datetime upper bound (inclusive) on timestamp",
+    )

models/responses.py

@@ -1,102 +1,130 @@
+from datetime import datetime
+
 from pydantic import BaseModel, Field
 
 
-class QueryResultItem(BaseModel):
-    text: str | None = None
-    score: float | None = None
+# --- Shared citations (spec-shaped) ---
 
-class QuerySourceItem(BaseModel):
-    source: str
-    page: int | None = None
-    chunk_index: int | None = None
-    score: float | None = None
-    excerpt: str | None = None
 
-class QueryResponse(BaseModel):
-    status: str
-    message: str
-    answer: str | None = None
-    sources: list[QuerySourceItem] = Field(default_factory=list)
-    results: list[QueryResultItem] = Field(default_factory=list)
+class SourceCitation(BaseModel):
+    document_name: str
+    page_number: int
+    chunk_text: str
+    relevance_score: float
+
+
+# --- Query: ask ---
+
+
+class AskQueryResponse(BaseModel):
+    query_id: str
+    question: str
+    answer: str
+    sources: list[SourceCitation] = Field(default_factory=list)
+    model_used: str
+    tokens_used: int
+    response_time_ms: int
+    timestamp: datetime
+
+
+# --- Query: summarise ---
+
+
+class SummariseQueryResponse(BaseModel):
+    query_id: str
+    summary: str
+    document_count: int
+    sources: list[SourceCitation] = Field(default_factory=list)
+    timestamp: datetime
+
+
+# --- Ingest ---
+
 
 class IngestUploadResponse(BaseModel):
+    job_id: str
     status: str
+    total_files: int
+    filenames: list[str]
     message: str
+
+
+class UrlIngestResponse(BaseModel):
     job_id: str
-    document_ids: list[str] = Field(default_factory=list)
+    status: str
+    total_urls: int
+    message: str
 
 
 class CollectionItem(BaseModel):
     name: str
+    document_count: int
+    created_at: datetime | None = None
 
 
 class IngestCollectionsResponse(BaseModel):
-    status: str
-    message: str
     collections: list[CollectionItem] = Field(default_factory=list)
+    total: int
 
 
 class IngestDeleteCollectionResponse(BaseModel):
-    status: str
     message: str
-    collection_name: str
+    documents_removed: int
 
-class JobSummary(BaseModel):
-    job_id: str
-    status: str
-    collection_name: str | None = None
-    filename: str | None = None
-    created_at: str | None = None
 
-class JobListResponse(BaseModel):
-    status: str
-    message: str
-    jobs: list[JobSummary] = Field(default_factory=list)
+# --- Jobs ---
 
 
-class IngestJobDetail(BaseModel):
+class JobStatusResponse(BaseModel):
     job_id: str
     status: str
-    collection_name: str
-    filename: str
-    message: str
-    document_ids: list[str] = Field(default_factory=list)
-    created_at: str
-    updated_at: str
+    total_files: int
+    processed_files: int
+    failed_files: int
+    progress_percent: int
+    started_at: datetime | None
+    completed_at: datetime | None
+    errors: list[str] = Field(default_factory=list)
 
 
-class IngestJobDetailResponse(BaseModel):
+class JobListItem(BaseModel):
+    job_id: str
     status: str
-    message: str
-    job: IngestJobDetail | None = None
+    total_files: int
+    completed_at: datetime | None = None
 
-class AuditEvent(BaseModel):
-    event_id: str
-    action: str
-    question: str | None = None
-    collection_name: str | None = None
-    created_at: str | None = None
 
-class AuditListResponse(BaseModel):
-    status: str
-    message: str
-    events: list[AuditEvent] = Field(default_factory=list)
+class JobListResponse(BaseModel):
+    jobs: list[JobListItem] = Field(default_factory=list)
+    total: int
+
+
+# --- Audit ---
 
 
-class AuditDetail(BaseModel):
-    event_id: str
-    action: str
+class AuditLogEntry(BaseModel):
+    query_id: str
+    user_id: str
     question: str
-    collection_name: str
-    answer: str | None = None
-    status: str
-    message: str
-    sources: list[QuerySourceItem] = Field(default_factory=list)
-    results: list[QueryResultItem] = Field(default_factory=list)
-    created_at: str
+    answer_summary: str
+    sources_count: int
+    model_used: str | None
+    timestamp: datetime
 
 
-class AuditDetailResponse(BaseModel):
-    status: str
-    message: str
-    event: AuditDetail | None = None
+class AuditLogsResponse(BaseModel):
+    logs: list[AuditLogEntry] = Field(default_factory=list)
+    total: int
+    limit: int
+    offset: int
+
+
+class AuditLogDetailResponse(BaseModel):
+    query_id: str
+    user_id: str
+    question: str
+    full_answer: str
+    sources: list[SourceCitation] = Field(default_factory=list)
+    model_used: str | None
+    tokens_used: int | None
+    timestamp: datetime

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py

rag/retriever.py

@@ -27,9 +27,27 @@ except ImportError:
 
 from api.config import Settings
 
-NO_MATCH_ANSWER = "I could not find this information in the uploaded documents."
+NO_MATCH_ANSWER = "I cannot find this information in the uploaded documents."
 MIN_RELEVANCE_SCORE = 0.15
 
+# Verbatim from DOCUAUDIT_AI_REQUIREMENTS.md (placeholders filled at runtime).
+DOCUAUDIT_ASK_TEMPLATE = """You are DocuAudit AI, an expert document analyst for consulting environments.
+
+RULES:
+1. Answer ONLY based on the provided document excerpts below.
+2. If the answer is not in the documents, say: "I cannot find this information in the uploaded documents."
+3. ALWAYS cite your sources: mention the document name and page number for every claim.
+4. Be precise and professional. This is a high-stakes consulting environment.
+5. Do not speculate or add information not present in the documents.
+
+DOCUMENT EXCERPTS:
+{context}
+
+QUESTION: {question}
+
+ANSWER (with source citations):
+"""
+
 
 @dataclass
 class RetrievedChunk:
@@ -62,31 +80,19 @@ SUMMARY_RETRIEVAL_QUERY = (
 )
 
 
-def answer_with_grounding(settings: Settings, question: str, chunks: list[RetrievedChunk]) -> str:
+def answer_with_grounding(settings: Settings, question: str, chunks: list[RetrievedChunk]) -> tuple[str, int]:
     ranked_chunks = [chunk for chunk in chunks if chunk.score is None or chunk.score >= MIN_RELEVANCE_SCORE]
     if not ranked_chunks:
-        return NO_MATCH_ANSWER
+        return NO_MATCH_ANSWER, 0
 
     llm = _create_chat_model(settings)
     prompt_context = _format_context(ranked_chunks)
-    messages = [
-        SystemMessage(
-            content=(
-                "You answer questions using only the provided context from uploaded documents. "
-                "If the answer is not in context, say you do not know."
-            )
-        ),
-        HumanMessage(
-            content=(
-                f"Question: {question}\n\n"
-                f"Context:\n{prompt_context}\n\n"
-                "Return a concise grounded answer."
-            )
-        ),
-    ]
+    user_content = DOCUAUDIT_ASK_TEMPLATE.format(context=prompt_context, question=question)
+    messages = [HumanMessage(content=user_content)]
     response = llm.invoke(messages)
     answer = _extract_message_text(response).strip()
-    return answer or NO_MATCH_ANSWER
+    tokens = _extract_usage_tokens(response)
+    return (answer or NO_MATCH_ANSWER), tokens
 
 
 def summarise_with_grounding(
@@ -94,10 +100,10 @@ def summarise_with_grounding(
     *,
     focus: str | None,
     chunks: list[RetrievedChunk],
-) -> str:
+) -> tuple[str, int]:
     ranked_chunks = [chunk for chunk in chunks if chunk.score is None or chunk.score >= MIN_RELEVANCE_SCORE]
     if not ranked_chunks:
-        return NO_MATCH_ANSWER
+        return NO_MATCH_ANSWER, 0
 
     llm = _create_chat_model(settings)
     prompt_context = _format_context(ranked_chunks)
@@ -123,7 +129,8 @@ def summarise_with_grounding(
     ]
     response = llm.invoke(messages)
     answer = _extract_message_text(response).strip()
-    return answer or NO_MATCH_ANSWER
+    tokens = _extract_usage_tokens(response)
+    return (answer or NO_MATCH_ANSWER), tokens
 
 
 def _create_chat_model(settings: Settings) -> BaseChatModel:
@@ -186,6 +193,25 @@ def _to_int_or_none(value: object) -> int | None:
         return None
 
 
+def _extract_usage_tokens(response: object) -> int:
+    um = getattr(response, "usage_metadata", None)
+    if isinstance(um, dict):
+        total = um.get("total_tokens")
+        if total is not None:
+            return int(total)
+        inp = int(um.get("input_tokens", 0) or 0)
+        out = int(um.get("output_tokens", 0) or 0)
+        return inp + out
+    rm = getattr(response, "response_metadata", None) or {}
+    if isinstance(rm, dict):
+        tu = rm.get("token_usage")
+        if isinstance(tu, dict):
+            if tu.get("total_tokens") is not None:
+                return int(tu["total_tokens"])
+            return int(tu.get("prompt_tokens", 0) or 0) + int(tu.get("completion_tokens", 0) or 0)
+    return 0
+
+
 def _extract_message_text(response: object) -> str:
     content = getattr(response, "content", "")
     if isinstance(content, str):

rag/vector_store.py

@@ -36,8 +36,42 @@ def list_collection_names(persist_directory: str) -> list[str]:
     return sorted(c.name for c in client.list_collections())
 
 
-def delete_collection(persist_directory: str, collection_name: str) -> None:
+def delete_collection(persist_directory: str, collection_name: str) -> int:
+    """Delete a collection and return the number of documents that were removed (best effort)."""
     Path(persist_directory).mkdir(parents=True, exist_ok=True)
     client = chromadb.PersistentClient(path=persist_directory, settings=_CHROMA_CLIENT_SETTINGS)
+    removed = 0
+    try:
+        col = client.get_collection(name=collection_name)
+        removed = int(col.count())
+    except Exception:
+        removed = 0
     client.delete_collection(name=collection_name)
+    return removed
+
+
+def collection_document_count(persist_directory: str, collection_name: str) -> int:
+    Path(persist_directory).mkdir(parents=True, exist_ok=True)
+    client = chromadb.PersistentClient(path=persist_directory, settings=_CHROMA_CLIENT_SETTINGS)
+    try:
+        col = client.get_collection(name=collection_name)
+        return int(col.count())
+    except Exception:
+        return 0
+
+
+def collection_created_at(persist_directory: str, collection_name: str) -> str | None:
+    """Return collection metadata ``created_at`` if present (Chroma-specific)."""
+    Path(persist_directory).mkdir(parents=True, exist_ok=True)
+    client = chromadb.PersistentClient(path=persist_directory, settings=_CHROMA_CLIENT_SETTINGS)
+    try:
+        col = client.get_collection(name=collection_name)
+        meta = getattr(col, "metadata", None) or {}
+        if isinstance(meta, dict):
+            raw = meta.get("created_at") or meta.get("created")
+            if raw is not None:
+                return str(raw)
+    except Exception:
+        pass
+    return None
 

storage/audit_store.py

@@ -1,11 +1,54 @@
 import json
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 from uuid import uuid4
 
 import aiosqlite
 
-from models.responses import AuditDetail, QueryResponse
+from models.responses import AuditLogDetailResponse, AuditLogEntry, SourceCitation
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _parse_ts(value: object) -> datetime:
+    if value is None or value == "":
+        return datetime.now(timezone.utc)
+    s = str(value).strip()
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    try:
+        dt = datetime.fromisoformat(s)
+        if dt.tzinfo is None:
+            return dt.replace(tzinfo=timezone.utc)
+        return dt
+    except ValueError:
+        return datetime.now(timezone.utc)
+
+
+async def _migrate_audit_columns(conn: aiosqlite.Connection) -> None:
+    cursor = await conn.execute("PRAGMA table_info(audit_events)")
+    rows = await cursor.fetchall()
+    col_names = {str(r[1]) for r in rows}
+    alters: list[str] = []
+    if "user_id" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN user_id TEXT NOT NULL DEFAULT 'anonymous'")
+    if "model_used" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN model_used TEXT")
+    if "tokens_used" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN tokens_used INTEGER")
+    if "response_time_ms" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN response_time_ms INTEGER")
+    if "answer_summary" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN answer_summary TEXT")
+    if "kind" not in col_names:
+        alters.append("ALTER TABLE audit_events ADD COLUMN kind TEXT NOT NULL DEFAULT 'ask'")
+    for stmt in alters:
+        await conn.execute(stmt)
+    if alters:
+        await conn.commit()
 
 
 async def init_audit_db(db_path: str) -> None:
@@ -24,80 +67,219 @@ async def init_audit_db(db_path: str) -> None:
                 message TEXT NOT NULL,
                 sources_json TEXT NOT NULL,
                 results_json TEXT NOT NULL,
-                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                user_id TEXT NOT NULL DEFAULT 'anonymous',
+                model_used TEXT,
+                tokens_used INTEGER,
+                response_time_ms INTEGER,
+                answer_summary TEXT,
+                kind TEXT NOT NULL DEFAULT 'ask'
             )
             """
         )
         await conn.commit()
+        await _migrate_audit_columns(conn)
+
+
+def _summary_from_answer(answer: str, max_len: int = 280) -> str:
+    text = (answer or "").strip()
+    if len(text) <= max_len:
+        return text
+    return text[: max_len - 1].rstrip() + "…"
+
+
+def _sources_to_citations(raw: list[dict[str, Any]]) -> list[SourceCitation]:
+    out: list[SourceCitation] = []
+    for item in raw:
+        if not isinstance(item, dict):
+            continue
+        if "document_name" in item:
+            doc = str(item.get("document_name", ""))
+            page = int(item.get("page_number", 0) or 0)
+            chunk = str(item.get("chunk_text", ""))
+            score = float(item.get("relevance_score", 0.0) or 0.0)
+        else:
+            doc = str(item.get("source", item.get("document_name", "")))
+            p = item.get("page_number", item.get("page"))
+            try:
+                page = int(p) if p is not None else 0
+            except (TypeError, ValueError):
+                page = 0
+            chunk = str(item.get("chunk_text", item.get("excerpt", item.get("text", ""))))
+            s = item.get("relevance_score", item.get("score"))
+            try:
+                score = float(s) if s is not None else 0.0
+            except (TypeError, ValueError):
+                score = 0.0
+        out.append(
+            SourceCitation(
+                document_name=doc or "unknown",
+                page_number=page,
+                chunk_text=chunk,
+                relevance_score=score,
+            )
+        )
+    return out
 
 
 async def persist_query_audit(
     db_path: str,
     *,
+    query_id: str,
     action: str,
+    user_id: str,
     question: str,
     collection_name: str,
-    response: QueryResponse,
+    answer: str,
+    sources: list[SourceCitation],
+    model_used: str,
+    tokens_used: int,
+    response_time_ms: int,
+    status: str = "success",
+    message: str = "ok",
+    kind: str = "ask",
 ) -> str:
-    event_id = str(uuid4())
     await init_audit_db(db_path)
+    sources_payload = [s.model_dump(mode="json") for s in sources]
+    summary = _summary_from_answer(answer)
+    created = _utc_now_iso()
     async with aiosqlite.connect(db_path) as conn:
         await conn.execute(
             """
             INSERT INTO audit_events (
-                event_id, action, question, collection_name, answer, status, message, sources_json, results_json
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                event_id, action, question, collection_name, answer, status, message,
+                sources_json, results_json, created_at, user_id, model_used, tokens_used,
+                response_time_ms, answer_summary, kind
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, '[]', ?, ?, ?, ?, ?, ?, ?)
             """,
             (
-                event_id,
+                query_id,
                 action,
                 question,
                 collection_name,
-                response.answer,
-                response.status,
-                response.message,
-                json.dumps([item.model_dump() for item in response.sources]),
-                json.dumps([item.model_dump() for item in response.results]),
+                answer,
+                status,
+                message,
+                json.dumps(sources_payload),
+                created,
+                user_id,
+                model_used,
+                tokens_used,
+                response_time_ms,
+                summary,
+                kind,
             ),
         )
         await conn.commit()
-    return event_id
+    return query_id
 
 
-async def list_audit_events(db_path: str, *, limit: int, offset: int) -> list[dict[str, Any]]:
+async def count_audit_events(
+    db_path: str,
+    *,
+    user_id: str | None = None,
+    from_date: str | None = None,
+    to_date: str | None = None,
+) -> int:
     await init_audit_db(db_path)
+    where, params = _audit_filters(user_id, from_date, to_date)
+    async with aiosqlite.connect(db_path) as conn:
+        cur = await conn.execute(f"SELECT COUNT(*) AS c FROM audit_events {where}", params)
+        row = await cur.fetchone()
+    return int(row[0]) if row else 0
+
+
+def _audit_filters(user_id: str | None, from_date: str | None, to_date: str | None) -> tuple[str, list[Any]]:
+    clauses: list[str] = []
+    params: list[Any] = []
+    if user_id:
+        clauses.append("user_id = ?")
+        params.append(user_id)
+    if from_date:
+        clauses.append("datetime(created_at) >= datetime(?)")
+        params.append(from_date)
+    if to_date:
+        clauses.append("datetime(created_at) <= datetime(?)")
+        params.append(to_date)
+    if not clauses:
+        return "", []
+    return "WHERE " + " AND ".join(clauses), params
+
+
+async def list_audit_events(
+    db_path: str,
+    *,
+    limit: int,
+    offset: int,
+    user_id: str | None = None,
+    from_date: str | None = None,
+    to_date: str | None = None,
+) -> tuple[list[AuditLogEntry], int]:
+    await init_audit_db(db_path)
+    where, fparams = _audit_filters(user_id, from_date, to_date)
+    total = await count_audit_events(db_path, user_id=user_id, from_date=from_date, to_date=to_date)
     async with aiosqlite.connect(db_path) as conn:
         conn.row_factory = aiosqlite.Row
         cursor = await conn.execute(
-            """
-            SELECT event_id, action, question, collection_name, created_at
+            f"""
+            SELECT event_id, user_id, question, answer, answer_summary, sources_json, model_used, created_at
             FROM audit_events
+            {where}
             ORDER BY datetime(created_at) DESC, rowid DESC
             LIMIT ? OFFSET ?
             """,
-            (limit, offset),
+            [*fparams, limit, offset],
         )
         rows = await cursor.fetchall()
-    return [dict(row) for row in rows]
+    logs: list[AuditLogEntry] = []
+    for row in rows:
+        src_raw = json.loads(row["sources_json"] or "[]")
+        if not isinstance(src_raw, list):
+            src_raw = []
+        summary_cell = row["answer_summary"]
+        summary_text = str(summary_cell).strip() if summary_cell else ""
+        if not summary_text:
+            summary_text = _summary_from_answer(str(row["answer"] or ""))
+        logs.append(
+            AuditLogEntry(
+                query_id=str(row["event_id"]),
+                user_id=str(row["user_id"] or "anonymous"),
+                question=str(row["question"]),
+                answer_summary=summary_text,
+                sources_count=len(src_raw),
+                model_used=row["model_used"],
+                timestamp=_parse_ts(row["created_at"]),
+            )
+        )
+    return logs, total
 
 
-async def get_audit_event(db_path: str, event_id: str) -> AuditDetail | None:
+async def get_audit_event(db_path: str, query_id: str) -> AuditLogDetailResponse | None:
     await init_audit_db(db_path)
     async with aiosqlite.connect(db_path) as conn:
         conn.row_factory = aiosqlite.Row
         cursor = await conn.execute(
             """
-            SELECT event_id, action, question, collection_name, answer, status, message, sources_json, results_json, created_at
+            SELECT event_id, user_id, question, answer, sources_json, model_used, tokens_used, created_at
             FROM audit_events
             WHERE event_id = ?
             """,
-            (event_id,),
+            (query_id,),
         )
         row = await cursor.fetchone()
     if row is None:
         return None
-
-    payload = dict(row)
-    payload["sources"] = json.loads(payload.pop("sources_json") or "[]")
-    payload["results"] = json.loads(payload.pop("results_json") or "[]")
-    return AuditDetail.model_validate(payload)
+    src_raw = json.loads(row["sources_json"] or "[]")
+    if not isinstance(src_raw, list):
+        src_raw = []
+    citations = _sources_to_citations(src_raw)
+    return AuditLogDetailResponse(
+        query_id=str(row["event_id"]),
+        user_id=str(row["user_id"] or "anonymous"),
+        question=str(row["question"]),
+        full_answer=str(row["answer"] or ""),
+        sources=citations,
+        model_used=row["model_used"],
+        tokens_used=row["tokens_used"],
+        timestamp=_parse_ts(row["created_at"]),
+    )

storage/job_store.py

@@ -1,11 +1,64 @@
 import json
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 from uuid import uuid4
 
 import aiosqlite
 
-from models.responses import IngestJobDetail
+from models.responses import JobListItem, JobStatusResponse
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+async def _migrate_jobs_columns(conn: aiosqlite.Connection) -> None:
+    cursor = await conn.execute("PRAGMA table_info(ingest_jobs)")
+    rows = await cursor.fetchall()
+    col_names = {str(r[1]) for r in rows}
+    alters: list[str] = []
+    if "total_files" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN total_files INTEGER NOT NULL DEFAULT 1")
+    if "processed_files" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN processed_files INTEGER NOT NULL DEFAULT 0")
+    if "failed_files" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN failed_files INTEGER NOT NULL DEFAULT 0")
+    if "filenames_json" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN filenames_json TEXT NOT NULL DEFAULT '[]'")
+    if "errors_json" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN errors_json TEXT NOT NULL DEFAULT '[]'")
+    if "started_at" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN started_at TEXT")
+    if "completed_at" not in col_names:
+        alters.append("ALTER TABLE ingest_jobs ADD COLUMN completed_at TEXT")
+    for stmt in alters:
+        await conn.execute(stmt)
+    if alters:
+        await conn.commit()
+    await _backfill_job_filenames(conn)
+
+
+async def _backfill_job_filenames(conn: aiosqlite.Connection) -> None:
+    conn.row_factory = aiosqlite.Row
+    cursor = await conn.execute("SELECT job_id, filename, filenames_json, total_files FROM ingest_jobs")
+    rows = await cursor.fetchall()
+    for row in rows:
+        raw = row["filenames_json"] or "[]"
+        try:
+            parsed: Any = json.loads(raw)
+        except json.JSONDecodeError:
+            parsed = []
+        if not parsed and row["filename"]:
+            await conn.execute(
+                """
+                UPDATE ingest_jobs
+                SET filenames_json = ?, total_files = CASE WHEN total_files IS NULL OR total_files < 1 THEN 1 ELSE total_files END
+                WHERE job_id = ?
+                """,
+                (json.dumps([row["filename"]]), row["job_id"]),
+            )
+    await conn.commit()
 
 
 async def init_jobs_db(db_path: str) -> None:
@@ -22,74 +75,134 @@ async def init_jobs_db(db_path: str) -> None:
                 message TEXT NOT NULL DEFAULT '',
                 document_ids_json TEXT NOT NULL DEFAULT '[]',
                 created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+                updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                total_files INTEGER NOT NULL DEFAULT 1,
+                processed_files INTEGER NOT NULL DEFAULT 0,
+                failed_files INTEGER NOT NULL DEFAULT 0,
+                filenames_json TEXT NOT NULL DEFAULT '[]',
+                errors_json TEXT NOT NULL DEFAULT '[]',
+                started_at TEXT,
+                completed_at TEXT
             )
             """
         )
         await conn.commit()
+        await _migrate_jobs_columns(conn)
 
 
 async def create_ingest_job(
     db_path: str,
     *,
     collection_name: str,
-    filename: str,
+    filenames: list[str],
 ) -> str:
+    if not filenames:
+        raise ValueError("filenames must not be empty")
     job_id = str(uuid4())
+    primary = filenames[0]
+    names_json = json.dumps(filenames)
+    total = len(filenames)
     await init_jobs_db(db_path)
     async with aiosqlite.connect(db_path) as conn:
         await conn.execute(
             """
             INSERT INTO ingest_jobs (
-                job_id, status, collection_name, filename, message, document_ids_json
-            ) VALUES (?, 'queued', ?, ?, '', '[]')
+                job_id, status, collection_name, filename, message, document_ids_json,
+                total_files, processed_files, failed_files, filenames_json, errors_json
+            ) VALUES (?, 'queued', ?, ?, '', '[]', ?, 0, 0, ?, '[]')
             """,
-            (job_id, collection_name, filename),
+            (job_id, collection_name, primary, total, names_json),
         )
         await conn.commit()
     return job_id
 
 
-async def update_ingest_job(
+async def mark_job_processing(db_path: str, job_id: str) -> None:
+    await init_jobs_db(db_path)
+    started = _utc_now_iso()
+    async with aiosqlite.connect(db_path) as conn:
+        await conn.execute(
+            """
+            UPDATE ingest_jobs
+            SET status = 'processing', message = 'Ingestion in progress.', started_at = COALESCE(started_at, ?),
+                updated_at = CURRENT_TIMESTAMP
+            WHERE job_id = ?
+            """,
+            (started, job_id),
+        )
+        await conn.commit()
+
+
+async def update_job_progress(
     db_path: str,
     job_id: str,
     *,
-    status: str,
+    processed_files: int,
+    failed_files: int,
+    errors: list[str],
     message: str | None = None,
-    document_ids: list[str] | None = None,
 ) -> None:
     await init_jobs_db(db_path)
     async with aiosqlite.connect(db_path) as conn:
-        if document_ids is not None:
-            await conn.execute(
-                """
-                UPDATE ingest_jobs
-                SET status = ?, message = COALESCE(?, message), document_ids_json = ?,
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE job_id = ?
-                """,
-                (status, message, json.dumps(document_ids), job_id),
-            )
-        else:
-            await conn.execute(
-                """
-                UPDATE ingest_jobs
-                SET status = ?, message = COALESCE(?, message),
-                    updated_at = CURRENT_TIMESTAMP
-                WHERE job_id = ?
-                """,
-                (status, message, job_id),
-            )
+        await conn.execute(
+            """
+            UPDATE ingest_jobs
+            SET processed_files = ?, failed_files = ?, errors_json = ?,
+                message = COALESCE(?, message), updated_at = CURRENT_TIMESTAMP
+            WHERE job_id = ?
+            """,
+            (processed_files, failed_files, json.dumps(errors), message, job_id),
+        )
         await conn.commit()
 
 
-async def get_ingest_job(db_path: str, job_id: str) -> IngestJobDetail | None:
+async def complete_ingest_job(
+    db_path: str,
+    job_id: str,
+    *,
+    document_ids: list[str],
+    message: str,
+) -> None:
+    await init_jobs_db(db_path)
+    completed = _utc_now_iso()
+    async with aiosqlite.connect(db_path) as conn:
+        await conn.execute(
+            """
+            UPDATE ingest_jobs
+            SET status = 'completed', message = ?, document_ids_json = ?,
+                completed_at = ?, updated_at = CURRENT_TIMESTAMP
+            WHERE job_id = ?
+            """,
+            (message, json.dumps(document_ids), completed, job_id),
+        )
+        await conn.commit()
+
+
+async def fail_ingest_job(db_path: str, job_id: str, *, message: str, errors: list[str] | None = None) -> None:
+    await init_jobs_db(db_path)
+    completed = _utc_now_iso()
+    err_json = json.dumps(errors or [message])
+    async with aiosqlite.connect(db_path) as conn:
+        await conn.execute(
+            """
+            UPDATE ingest_jobs
+            SET status = 'failed', message = ?, errors_json = ?, completed_at = ?,
+                updated_at = CURRENT_TIMESTAMP
+            WHERE job_id = ?
+            """,
+            (message, err_json, completed, job_id),
+        )
+        await conn.commit()
+
+
+async def get_job_status(db_path: str, job_id: str) -> JobStatusResponse | None:
     await init_jobs_db(db_path)
     async with aiosqlite.connect(db_path) as conn:
         conn.row_factory = aiosqlite.Row
         cursor = await conn.execute(
             """
-            SELECT job_id, status, collection_name, filename, message, document_ids_json, created_at, updated_at
+            SELECT job_id, status, total_files, processed_files, failed_files, errors_json,
+                   started_at, completed_at, message
             FROM ingest_jobs
             WHERE job_id = ?
             """,
@@ -98,18 +211,39 @@ async def get_ingest_job(db_path: str, job_id: str) -> IngestJobDetail | None:
         row = await cursor.fetchone()
     if row is None:
         return None
-    payload = dict(row)
-    payload["document_ids"] = json.loads(payload.pop("document_ids_json") or "[]")
-    return IngestJobDetail.model_validate(payload)
+    data = dict(row)
+    total = int(data["total_files"] or 0)
+    processed = int(data["processed_files"] or 0)
+    failed = int(data["failed_files"] or 0)
+    denom = total if total > 0 else 1
+    progress = int(min(100, max(0, round((processed + failed) / denom * 100))))
+    errors = json.loads(data.get("errors_json") or "[]")
+    if not isinstance(errors, list):
+        errors = [str(errors)]
+    errors_str = [str(e) for e in errors]
+    return JobStatusResponse(
+        job_id=str(data["job_id"]),
+        status=str(data["status"]),
+        total_files=total,
+        processed_files=processed,
+        failed_files=failed,
+        progress_percent=progress,
+        started_at=_parse_dt(data.get("started_at")),
+        completed_at=_parse_dt(data.get("completed_at")),
+        errors=errors_str,
+    )
 
 
-async def list_ingest_jobs(db_path: str, *, limit: int, offset: int) -> list[dict[str, Any]]:
+async def list_ingest_jobs(db_path: str, *, limit: int, offset: int) -> tuple[list[JobListItem], int]:
     await init_jobs_db(db_path)
     async with aiosqlite.connect(db_path) as conn:
         conn.row_factory = aiosqlite.Row
+        cur_total = await conn.execute("SELECT COUNT(*) AS c FROM ingest_jobs")
+        total_row = await cur_total.fetchone()
+        total = int(total_row["c"]) if total_row else 0
         cursor = await conn.execute(
             """
-            SELECT job_id, status, collection_name, filename, created_at
+            SELECT job_id, status, total_files, completed_at
             FROM ingest_jobs
             ORDER BY datetime(updated_at) DESC, rowid DESC
             LIMIT ? OFFSET ?
@@ -117,4 +251,30 @@ async def list_ingest_jobs(db_path: str, *, limit: int, offset: int) -> list[dic
             (limit, offset),
         )
         rows = await cursor.fetchall()
-    return [dict(row) for row in rows]
+    items = [
+        JobListItem(
+            job_id=str(r["job_id"]),
+            status=str(r["status"]),
+            total_files=int(r["total_files"] or 0),
+            completed_at=_parse_dt(r["completed_at"]),
+        )
+        for r in rows
+    ]
+    return items, total
+
+
+def _parse_dt(value: object) -> datetime | None:
+    if value is None or value == "":
+        return None
+    s = str(value).strip()
+    if not s:
+        return None
+    if s.endswith("Z"):
+        s = s[:-1] + "+00:00"
+    try:
+        dt = datetime.fromisoformat(s)
+        if dt.tzinfo is None:
+            return dt.replace(tzinfo=timezone.utc)
+        return dt
+    except ValueError:
+        return None

streamlit_app.py

@@ -71,17 +71,43 @@ def _fmt_api_error(exc: httpx.HTTPStatusError) -> str:
     return f"HTTP {exc.response.status_code}"
 
 
-def _post_query_ask(client: httpx.Client, *, question: str, collection_name: str) -> httpx.Response:
-    """Milestone 8 uses POST /query/ask; older servers only expose POST /query."""
-    body = {"question": question.strip(), "collection_name": collection_name}
+def _post_query_ask(
+    client: httpx.Client,
+    *,
+    question: str,
+    collection_name: str,
+    top_k: int = 5,
+    user_id: str = "anonymous",
+) -> httpx.Response:
+    """POST /query/ask (falls back to POST /query on older servers)."""
+    body: dict[str, object] = {
+        "question": question.strip(),
+        "collection_name": collection_name,
+        "top_k": top_k,
+        "user_id": user_id,
+    }
     r = client.post("/query/ask", json=body)
     if r.status_code == 404:
         r = client.post("/query", json=body)
     return r
 
 
-def _get_audit_logs(client: httpx.Client, *, limit: int, offset: int) -> httpx.Response:
-    params = {"limit": limit, "offset": offset}
+def _get_audit_logs(
+    client: httpx.Client,
+    *,
+    limit: int,
+    offset: int,
+    user_id: str | None = None,
+    from_date: str | None = None,
+    to_date: str | None = None,
+) -> httpx.Response:
+    params: dict[str, object] = {"limit": limit, "offset": offset}
+    if user_id:
+        params["user_id"] = user_id
+    if from_date:
+        params["from_date"] = from_date
+    if to_date:
+        params["to_date"] = to_date
     r = client.get("/audit/logs", params=params)
     if r.status_code == 404:
         r = client.get("/audit", params=params)
@@ -156,7 +182,7 @@ def main() -> None:
                     st.warning("Choose a file first.")
                 else:
                     try:
-                        files = {"file": (uploaded.name, uploaded.getvalue(), uploaded.type or "application/octet-stream")}
+                        files = {"files": (uploaded.name, uploaded.getvalue(), uploaded.type or "application/octet-stream")}
                         data = {"collection_name": up_collection}
                         with _client() as c:
                             r = c.post("/ingest/upload", files=files, data=data)
@@ -185,7 +211,10 @@ def main() -> None:
             else:
                 try:
                     with _client() as c:
-                        r = c.post("/ingest/url", json={"url": ingest_url.strip(), "collection_name": url_collection})
+                        r = c.post(
+                            "/ingest/url",
+                            json={"urls": [ingest_url.strip()], "collection_name": url_collection},
+                        )
                         r.raise_for_status()
                         out = r.json()
                     st.success(out.get("message", "Queued"))
@@ -206,10 +235,10 @@ def main() -> None:
                     r = c.get("/ingest/collections")
                     r.raise_for_status()
                     cols = r.json()
-                names = [x["name"] for x in cols.get("collections", [])]
-                st.write(cols.get("message", ""))
-                if names:
-                    st.dataframe({"name": names}, hide_index=True, use_container_width=True)
+                rows = cols.get("collections", [])
+                st.write(f"{cols.get('total', len(rows))} collection(s).")
+                if rows:
+                    st.dataframe(rows, hide_index=True, use_container_width=True)
                 else:
                     st.info("No collections yet.")
             except httpx.HTTPStatusError as e:
@@ -228,7 +257,10 @@ def main() -> None:
                     with _client() as c:
                         r = c.delete(f"/ingest/collection/{del_name.strip()}")
                         r.raise_for_status()
-                    st.success(r.json().get("message", "Deleted"))
+                    del_body = r.json()
+                    st.success(del_body.get("message", "Deleted"))
+                    if "documents_removed" in del_body:
+                        st.caption(f"Documents removed: **{del_body['documents_removed']}**")
                 except httpx.HTTPStatusError as e:
                     st.error(_fmt_api_error(e))
                 except httpx.ConnectError as e:
@@ -250,7 +282,7 @@ def main() -> None:
                     r.raise_for_status()
                     payload = r.json()
                 jobs: list[dict[str, Any]] = payload.get("jobs", [])
-                st.caption(payload.get("message", ""))
+                st.caption(f"Total jobs (matching filters): **{payload.get('total', len(jobs))}**")
                 if jobs:
                     st.dataframe(jobs, hide_index=True, use_container_width=True)
                 else:
@@ -293,9 +325,8 @@ def main() -> None:
                         r = c.get(f"/jobs/{job_id.strip()}")
                         r.raise_for_status()
                         body = r.json()
-                        job = body.get("job") or {}
-                        st_ = job.get("status", "")
-                        status_ph.write(f"Poll {i + 1}: **{st_}** — {job.get('message', '')}")
+                        st_ = body.get("status", "")
+                        status_ph.write(f"Poll {i + 1}: **{st_}** — {body.get('progress_percent', 0)}%")
                         if st_ in ("completed", "failed"):
                             st.json(body)
                             break
@@ -331,8 +362,7 @@ def main() -> None:
                             )
                             r.raise_for_status()
                             ans = r.json()
-                    msg = ans.get("message") or ""
-                    st.success(msg if msg else "Request completed.")
+                    st.success(f"Query id: `{ans.get('query_id', '')}`")
                     if ans.get("answer"):
                         st.markdown("### Answer")
                         st.markdown(ans["answer"])
@@ -370,11 +400,11 @@ def main() -> None:
                         r = c.post("/query/summarise", json=body)
                         r.raise_for_status()
                         ans = r.json()
-                msg = ans.get("message") or ""
-                st.success(msg if msg else "Request completed.")
-                if ans.get("answer"):
+                st.success(f"Query id: `{ans.get('query_id', '')}` · documents: **{ans.get('document_count', '')}**")
+                summary_text = ans.get("summary") or ans.get("answer")
+                if summary_text:
                     st.markdown("### Summary")
-                    st.markdown(ans["answer"])
+                    st.markdown(summary_text)
                 else:
                     st.warning("No summary text in the response; see **Raw response** below.")
                 src = ans.get("sources") or []
@@ -411,11 +441,15 @@ def main() -> None:
                     )
                     r.raise_for_status()
                     payload = r.json()
-                events = payload.get("events", [])
-                st.caption(payload.get("message", ""))
+                events = payload.get("logs", payload.get("events", []))
+                st.caption(f"Total matching: **{payload.get('total', len(events))}**")
                 if events:
                     st.dataframe(events, hide_index=True, use_container_width=True)
-                    ids = [e["event_id"] for e in events if isinstance(e, dict) and "event_id" in e]
+                    ids = [
+                        e.get("query_id") or e.get("event_id")
+                        for e in events
+                        if isinstance(e, dict) and (e.get("query_id") or e.get("event_id"))
+                    ]
                     if ids:
                         st.session_state["_audit_ids"] = ids
                 else:
@@ -432,7 +466,7 @@ def main() -> None:
         pick = ""
         if ids_for_select:
             pick = st.selectbox("Event ID", options=[""] + list(ids_for_select), key="audit_pick")
-        manual_id = st.text_input("Or enter event ID", key="audit_manual")
+        manual_id = st.text_input("Or enter query / event ID", key="audit_manual")
         ev_id = (manual_id.strip() or (pick or "").strip()).strip()
         if st.button("Load detail", key="btn_audit_detail") and ev_id:
             try:

tests/test_audit.py

@@ -1,12 +1,13 @@
 import asyncio
 from unittest.mock import AsyncMock
+from uuid import uuid4
 
 import pytest
 from fastapi.testclient import TestClient
 
 from api.config import Settings
 from api.main import app
-from models.responses import QueryResponse
+from models.responses import SourceCitation
 from storage.audit_store import persist_query_audit
 
 
@@ -32,39 +33,89 @@ def client(settings, monkeypatch):
         yield test_client
 
 
-def _seed_audit(settings: Settings, question: str = "What are key risks?") -> str:
-    return asyncio.run(
+def _seed_audit(settings: Settings, question: str = "What are key risks?", user_id: str = "analyst_001") -> str:
+    query_id = str(uuid4())
+    asyncio.run(
         persist_query_audit(
             settings.audit_db_path,
+            query_id=query_id,
             action="query",
+            user_id=user_id,
             question=question,
             collection_name="default",
-            response=QueryResponse(
-                status="success",
-                message="ok",
-                answer="Grounded answer",
-                sources=[],
-                results=[],
-            ),
+            answer="Grounded answer text for audit trail.",
+            sources=[
+                SourceCitation(
+                    document_name="report.pdf",
+                    page_number=3,
+                    chunk_text="Risk disclosure excerpt.",
+                    relevance_score=0.9,
+                )
+            ],
+            model_used="ollama:llama3.1:8b",
+            tokens_used=120,
+            response_time_ms=50,
+            kind="ask",
         )
     )
+    return query_id
 
 
 def test_audit_logs_and_detail_success(client, settings):
-    event_id = _seed_audit(settings)
+    query_id = _seed_audit(settings)
 
     list_response = client.get("/audit/logs?limit=10&offset=0")
     assert list_response.status_code == 200
     body = list_response.json()
-    assert body["status"] == "success"
-    assert len(body["events"]) >= 1
-    assert any(event["event_id"] == event_id for event in body["events"])
+    assert "logs" in body
+    assert body["total"] >= 1
+    assert any(entry["query_id"] == query_id for entry in body["logs"])
 
-    detail_response = client.get(f"/audit/logs/{event_id}")
+    detail_response = client.get(f"/audit/logs/{query_id}")
     assert detail_response.status_code == 200
     detail = detail_response.json()
-    assert detail["status"] == "success"
-    assert detail["event"]["question"] == "What are key risks?"
+    assert detail["query_id"] == query_id
+    assert detail["question"] == "What are key risks?"
+    assert detail["full_answer"] == "Grounded answer text for audit trail."
+    assert len(detail["sources"]) == 1
+    assert detail["sources"][0]["document_name"] == "report.pdf"
+
+
+def test_audit_logs_filter_by_user_id(client, settings):
+    q1 = _seed_audit(settings, question="Q one", user_id="user_a")
+    _seed_audit(settings, question="Q two", user_id="user_b")
+
+    r = client.get("/audit/logs", params={"user_id": "user_a", "limit": 50, "offset": 0})
+    assert r.status_code == 200
+    body = r.json()
+    ids = {e["query_id"] for e in body["logs"]}
+    assert q1 in ids
+    assert all(e["user_id"] == "user_a" for e in body["logs"])
+
+
+def test_audit_logs_filter_by_from_date(client, settings):
+    query_id = str(uuid4())
+    future = "2099-01-01T00:00:00Z"
+    asyncio.run(
+        persist_query_audit(
+            settings.audit_db_path,
+            query_id=query_id,
+            action="query",
+            user_id="u",
+            question="Future dated row",
+            collection_name="default",
+            answer="A",
+            sources=[],
+            model_used="m",
+            tokens_used=0,
+            response_time_ms=1,
+            kind="ask",
+        )
+    )
+    r = client.get("/audit/logs", params={"from_date": future, "limit": 50, "offset": 0})
+    assert r.status_code == 200
+    body = r.json()
+    assert query_id not in {e["query_id"] for e in body["logs"]}
 
 
 def test_audit_logs_validation_error_for_bad_limit(client):

tests/test_ingest.py

@@ -34,21 +34,23 @@ def test_upload_queues_job_success(client, monkeypatch):
     response = client.post(
         "/ingest/upload",
         data={"collection_name": "default"},
-        files={"file": ("sample.txt", b"hello world", "text/plain")},
+        files=[("files", ("sample.txt", b"hello world", "text/plain"))],
     )
 
     assert response.status_code == 200
     body = response.json()
     assert body["status"] == "queued"
     assert body["job_id"] == "job-123"
-    assert "Poll GET /jobs/job-123" in body["message"]
+    assert body["total_files"] == 1
+    assert body["filenames"] == ["sample.txt"]
+    assert "Poll /jobs/job-123" in body["message"]
 
 
 def test_upload_rejects_unsupported_extension(client):
     response = client.post(
         "/ingest/upload",
         data={"collection_name": "default"},
-        files={"file": ("sample.csv", b"a,b\n1,2", "text/csv")},
+        files=[("files", ("sample.csv", b"a,b\n1,2", "text/csv"))],
     )
 
     assert response.status_code == 400
@@ -65,7 +67,7 @@ def test_upload_returns_500_on_job_creation_error(client, monkeypatch):
     response = client.post(
         "/ingest/upload",
         data={"collection_name": "default"},
-        files={"file": ("sample.txt", b"hello", "text/plain")},
+        files=[("files", ("sample.txt", b"hello", "text/plain"))],
     )
 
     assert response.status_code == 500
@@ -75,12 +77,14 @@ def test_upload_returns_500_on_job_creation_error(client, monkeypatch):
 def test_ingest_url_rejects_non_http_scheme(client, monkeypatch):
     monkeypatch.setattr(
         "api.routes.ingest._download_url_to_temp",
-        AsyncMock(side_effect=ingest_route.HTTPException(status_code=400, detail="Only http and https URLs are supported.")),
+        AsyncMock(
+            side_effect=ingest_route.HTTPException(status_code=400, detail="Only http and https URLs are supported.")
+        ),
     )
 
     response = client.post(
         "/ingest/url",
-        json={"url": "https://example.com/file.txt", "collection_name": "default"},
+        json={"urls": ["https://example.com/file.txt"], "collection_name": "default"},
     )
 
     assert response.status_code == 400

tests/test_query.py

@@ -40,20 +40,51 @@ def test_ask_returns_grounded_answer_with_sources(client, monkeypatch):
     monkeypatch.setattr("api.routes.query.create_embedding_function", lambda: object())
     monkeypatch.setattr("api.routes.query.get_vector_store", lambda **_: object())
     monkeypatch.setattr("api.routes.query.retrieve_chunks", lambda *_: chunks)
-    monkeypatch.setattr("api.routes.query.answer_with_grounding", lambda *_: "Audi is expanding EV investment.")
+    monkeypatch.setattr("api.routes.query.answer_with_grounding", lambda *_: ("Audi is expanding EV investment.", 42))
     monkeypatch.setattr("api.routes.query.persist_query_audit", AsyncMock(return_value="evt-1"))
 
     response = client.post(
         "/query/ask",
-        json={"question": "What is Audi doing in EV?", "collection_name": "default"},
+        json={
+            "question": "What is Audi doing in EV markets worldwide?",
+            "collection_name": "default",
+            "top_k": 3,
+            "user_id": "tester",
+        },
     )
 
     assert response.status_code == 200
     body = response.json()
-    assert body["status"] == "success"
     assert body["answer"] == "Audi is expanding EV investment."
+    assert "query_id" in body
+    assert body["question"].startswith("What is Audi")
     assert len(body["sources"]) == 1
-    assert body["sources"][0]["source"] == "strategy.md"
+    assert body["sources"][0]["document_name"] == "strategy.md"
+    assert body["sources"][0]["page_number"] == 1
+    assert body["tokens_used"] == 42
+    assert "response_time_ms" in body
+    assert "model_used" in body
+
+
+def test_ask_respects_top_k_in_retrieve_call(client, monkeypatch):
+    captured: dict[str, object] = {}
+
+    def capture_retrieve(vs, question, k):
+        captured["k"] = k
+        return []
+
+    monkeypatch.setattr("api.routes.query.create_embedding_function", lambda: object())
+    monkeypatch.setattr("api.routes.query.get_vector_store", lambda **_: object())
+    monkeypatch.setattr("api.routes.query.retrieve_chunks", capture_retrieve)
+    monkeypatch.setattr("api.routes.query.answer_with_grounding", lambda *_: ("No match answer", 0))
+    monkeypatch.setattr("api.routes.query.persist_query_audit", AsyncMock())
+
+    response = client.post(
+        "/query/ask",
+        json={"question": "What is known about the topic here?", "collection_name": "default", "top_k": 7},
+    )
+    assert response.status_code == 200
+    assert captured.get("k") == 7
 
 
 def test_ask_returns_422_for_invalid_payload(client):
@@ -61,6 +92,14 @@ def test_ask_returns_422_for_invalid_payload(client):
     assert response.status_code == 422
 
 
+def test_ask_returns_422_for_short_question(client):
+    response = client.post(
+        "/query/ask",
+        json={"question": "hi", "collection_name": "default"},
+    )
+    assert response.status_code == 422
+
+
 def test_ask_returns_500_when_retrieval_fails(client, monkeypatch):
     monkeypatch.setattr("api.routes.query.create_embedding_function", lambda: object())
     monkeypatch.setattr("api.routes.query.get_vector_store", lambda **_: object())
@@ -68,7 +107,7 @@ def test_ask_returns_500_when_retrieval_fails(client, monkeypatch):
 
     response = client.post(
         "/query/ask",
-        json={"question": "What happened?", "collection_name": "default"},
+        json={"question": "What happened in the documents?", "collection_name": "default"},
     )
 
     assert response.status_code == 500
@@ -88,7 +127,8 @@ def test_summarise_returns_500_when_audit_persist_fails(client, monkeypatch):
     monkeypatch.setattr("api.routes.query.create_embedding_function", lambda: object())
     monkeypatch.setattr("api.routes.query.get_vector_store", lambda **_: object())
     monkeypatch.setattr("api.routes.query.retrieve_chunks", lambda *_: chunks)
-    monkeypatch.setattr("api.routes.query.summarise_with_grounding", lambda *_, **__: "Summary output")
+    monkeypatch.setattr("api.routes.query.summarise_with_grounding", lambda *_, **__: ("Summary output", 10))
+    monkeypatch.setattr("api.routes.query.collection_document_count", lambda *_: 5)
     monkeypatch.setattr(
         "api.routes.query.persist_query_audit",
         AsyncMock(side_effect=RuntimeError("audit write failed")),
@@ -96,7 +136,7 @@ def test_summarise_returns_500_when_audit_persist_fails(client, monkeypatch):
 
     response = client.post(
         "/query/summarise",
-        json={"collection_name": "default", "focus": "summarise risks"},
+        json={"collection_name": "default", "focus": "summarise risks", "user_id": "u1"},
     )
 
     assert response.status_code == 500

workers/ingest_worker.py

@@ -5,10 +5,15 @@ from rag.chunker import chunk_documents
 from rag.embedder import create_embedding_function
 from rag.loader import load_documents
 from rag.vector_store import add_documents, get_vector_store
-from storage.job_store import update_ingest_job
+from storage.job_store import (
+    complete_ingest_job,
+    fail_ingest_job,
+    mark_job_processing,
+    update_job_progress,
+)
 
 
-def _ingest_sync(temp_path: str, collection_name: str, chroma_persist_directory: str) -> tuple[list[str], int]:
+def _ingest_one_file_sync(temp_path: str, collection_name: str, chroma_persist_directory: str) -> tuple[list[str], int]:
     documents = load_documents(temp_path)
     chunks = chunk_documents(documents)
     if not chunks:
@@ -25,37 +30,72 @@ def _ingest_sync(temp_path: str, collection_name: str, chroma_persist_directory:
 
 async def run_ingest_job(
     job_id: str,
-    temp_path: str,
+    files: list[tuple[str, str]],
     collection_name: str,
     jobs_db_path: str,
     chroma_persist_directory: str,
 ) -> None:
+    """
+    Process one or more temp files for a single job. ``files`` is (temp_path, display_name).
+    """
+    all_doc_ids: list[str] = []
+    errors: list[str] = []
+    processed = 0
+    failed = 0
+    total = len(files)
+    if total == 0:
+        await fail_ingest_job(jobs_db_path, job_id, message="No files to ingest.")
+        return
+
     try:
-        await update_ingest_job(
-            jobs_db_path,
-            job_id,
-            status="processing",
-            message="Ingestion in progress.",
-        )
-        document_ids, num_chunks = await asyncio.to_thread(
-            _ingest_sync,
-            temp_path,
-            collection_name,
-            chroma_persist_directory,
-        )
-        await update_ingest_job(
+        await mark_job_processing(jobs_db_path, job_id)
+        for temp_path, display_name in files:
+            try:
+                doc_ids, num_chunks = await asyncio.to_thread(
+                    _ingest_one_file_sync,
+                    temp_path,
+                    collection_name,
+                    chroma_persist_directory,
+                )
+                all_doc_ids.extend(doc_ids)
+                processed += 1
+                await update_job_progress(
+                    jobs_db_path,
+                    job_id,
+                    processed_files=processed,
+                    failed_files=failed,
+                    errors=errors,
+                    message=f"Ingested {display_name} ({num_chunks} chunks).",
+                )
+            except Exception as exc:
+                failed += 1
+                errors.append(f"{display_name}: {exc}")
+                await update_job_progress(
+                    jobs_db_path,
+                    job_id,
+                    processed_files=processed,
+                    failed_files=failed,
+                    errors=errors,
+                    message=f"Failed on {display_name}: {exc}",
+                )
+            finally:
+                Path(temp_path).unlink(missing_ok=True)
+
+        if processed == 0:
+            await fail_ingest_job(
+                jobs_db_path,
+                job_id,
+                message="All files failed ingestion.",
+                errors=errors,
+            )
+            return
+
+        chunk_note = f"{len(all_doc_ids)} chunk vector(s) across {processed} file(s)."
+        await complete_ingest_job(
             jobs_db_path,
             job_id,
-            status="completed",
-            message=f"Ingested {num_chunks} chunks.",
-            document_ids=document_ids,
+            document_ids=all_doc_ids,
+            message=f"Ingestion completed. {chunk_note}",
         )
     except Exception as exc:
-        await update_ingest_job(
-            jobs_db_path,
-            job_id,
-            status="failed",
-            message=str(exc),
-        )
-    finally:
-        Path(temp_path).unlink(missing_ok=True)
+        await fail_ingest_job(jobs_db_path, job_id, message=str(exc), errors=errors + [str(exc)])