Add knowledge Q&A API: cited answers, document search, and store stats

app/common/src/main/java/stirling/software/SPDF/config/EndpointConfiguration.java

@@ -440,6 +440,9 @@ public class EndpointConfiguration {
         addEndpointToGroup("DocParse", "smart-split");
         addEndpointToGroup("DocParse", "chunk-document");
         addEndpointToGroup("DocParse", "rag-ingest");
+        addEndpointToGroup("DocParse", "rag-documents");
+        addEndpointToGroup("DocParse", "rag-search");
+        addEndpointToGroup("DocParse", "rag-ask");
         addEndpointToGroup("DocParse", "extract-tables");
         addEndpointToGroup("DocParse", "suggest-schema");
         addEndpointToGroup("DocParse", "fill-template");

app/proprietary/src/main/java/stirling/software/proprietary/controller/api/DocParseController.java

@@ -23,6 +23,8 @@ import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.ModelAttribute;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
@@ -47,7 +49,9 @@ import stirling.software.proprietary.model.api.docparse.ChunkDocumentApiRequest;
 import stirling.software.proprietary.model.api.docparse.ExtractFieldsApiRequest;
 import stirling.software.proprietary.model.api.docparse.ExtractTablesApiRequest;
 import stirling.software.proprietary.model.api.docparse.ParseDocumentApiRequest;
+import stirling.software.proprietary.model.api.docparse.RagAskApiRequest;
 import stirling.software.proprietary.model.api.docparse.RagIngestApiRequest;
+import stirling.software.proprietary.model.api.docparse.RagSearchApiRequest;
 import stirling.software.proprietary.model.api.docparse.SmartSplitApiRequest;
 import stirling.software.proprietary.model.api.docparse.SuggestSchemaApiRequest;
 import stirling.software.proprietary.model.docparse.ChunkDocumentResponse;
@@ -60,6 +64,7 @@ import stirling.software.proprietary.model.docparse.ExtractTablesResponse;
 import stirling.software.proprietary.model.docparse.FillDocxResponse;
 import stirling.software.proprietary.model.docparse.ParseDocumentResponse;
 import stirling.software.proprietary.model.docparse.RagIngestResponse;
+import stirling.software.proprietary.model.docparse.RagStatsView;
 import stirling.software.proprietary.model.docparse.SmartSplitResponse;
 import stirling.software.proprietary.model.docparse.SplitPart;
 import stirling.software.proprietary.model.docparse.SuggestSchemaResponse;
@@ -368,6 +373,52 @@ public class DocParseController {
                 CSV);
     }
 
+    @GetMapping("/rag-stats")
+    @Operation(
+            summary = "RAG store statistics",
+            description =
+                    "The engine's document-store totals (backend, documents, chunks, embedding"
+                            + " model) merged with the DocParse capability fields. Answers with"
+                            + " zeros and engineReachable=false when the engine is down.")
+    public ResponseEntity<RagStatsView> ragStats() {
+        return ResponseEntity.ok(docParseService.ragStats());
+    }
+
+    @GetMapping("/rag-documents")
+    @Operation(
+            summary = "List documents in the RAG store",
+            description =
+                    "Engine passthrough of the caller-visible indexed documents (documentId,"
+                            + " source, chunk count).")
+    public ResponseEntity<String> ragDocuments() throws IOException {
+        return jsonPassthrough(docParseService.ragDocuments());
+    }
+
+    @PostMapping(value = "/rag-search", consumes = MediaType.APPLICATION_JSON_VALUE)
+    @Operation(
+            summary = "Semantic search over the RAG store",
+            description =
+                    "Searches the caller-visible indexed documents and returns the top passages"
+                            + " with scores, page spans, and heading breadcrumbs.")
+    public ResponseEntity<String> ragSearch(@RequestBody RagSearchApiRequest request)
+            throws IOException {
+        return jsonPassthrough(docParseService.ragSearch(request.getQuery(), request.getTopK()));
+    }
+
+    @PostMapping(value = "/rag-ask", consumes = MediaType.APPLICATION_JSON_VALUE)
+    @Operation(
+            summary = "Ask a question over the RAG store",
+            description =
+                    "Answers the question from the caller-visible indexed documents and returns"
+                            + " the answer with its supporting passages.")
+    public ResponseEntity<String> ragAsk(@RequestBody RagAskApiRequest request) throws IOException {
+        return jsonPassthrough(docParseService.ragAsk(request.getQuestion(), request.getTopK()));
+    }
+
+    private static ResponseEntity<String> jsonPassthrough(String engineJson) {
+        return ResponseEntity.ok().contentType(MediaType.APPLICATION_JSON).body(engineJson);
+    }
+
     /** Original + requested corpus files in one ZIP, so destinations receive them together. */
     private byte[] exportZip(
             String fileName, byte[] original, RagIngestResponse result, RagIngestApiRequest request)

app/proprietary/src/main/java/stirling/software/proprietary/model/api/docparse/RagAskApiRequest.java

@@ -0,0 +1,17 @@
+package stirling.software.proprietary.model.api.docparse;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import lombok.Data;
+
+@Data
+public class RagAskApiRequest {
+
+    @Schema(
+            description = "Question to answer from the indexed documents",
+            requiredMode = Schema.RequiredMode.REQUIRED)
+    private String question;
+
+    @Schema(description = "Number of passages to ground the answer on (1-20)", defaultValue = "5")
+    private int topK = 5;
+}

app/proprietary/src/main/java/stirling/software/proprietary/model/api/docparse/RagSearchApiRequest.java

@@ -0,0 +1,17 @@
+package stirling.software.proprietary.model.api.docparse;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import lombok.Data;
+
+@Data
+public class RagSearchApiRequest {
+
+    @Schema(
+            description = "Natural-language search query",
+            requiredMode = Schema.RequiredMode.REQUIRED)
+    private String query;
+
+    @Schema(description = "Number of passages to return (1-50)", defaultValue = "10")
+    private int topK = 10;
+}

app/proprietary/src/main/java/stirling/software/proprietary/model/docparse/DocumentStoreStats.java

@@ -0,0 +1,5 @@
+package stirling.software.proprietary.model.docparse;
+
+/** Engine response for {@code GET /api/v1/documents/stats}: the RAG document store totals. */
+public record DocumentStoreStats(
+        String backend, long documents, long chunks, String embeddingModel) {}

app/proprietary/src/main/java/stirling/software/proprietary/model/docparse/RagAskRequest.java

@@ -0,0 +1,4 @@
+package stirling.software.proprietary.model.docparse;
+
+/** Engine request for {@code POST /api/v1/documents/ask}: grounded Q&A over the RAG store. */
+public record RagAskRequest(String question, int topK) {}

app/proprietary/src/main/java/stirling/software/proprietary/model/docparse/RagSearchRequest.java

@@ -0,0 +1,4 @@
+package stirling.software.proprietary.model.docparse;
+
+/** Engine request for {@code POST /api/v1/documents/search}: semantic search over the RAG store. */
+public record RagSearchRequest(String query, int topK) {}

app/proprietary/src/main/java/stirling/software/proprietary/model/docparse/RagStatsView.java

@@ -0,0 +1,15 @@
+package stirling.software.proprietary.model.docparse;
+
+/**
+ * Merged RAG store view served by {@code GET /api/v1/docparse/rag-stats} (Java side): the engine's
+ * document-store totals plus the cached DocParse capability fields. When the engine is unreachable
+ * the totals are zero and {@code engineReachable} is false.
+ */
+public record RagStatsView(
+        String backend,
+        long documents,
+        long chunks,
+        String embeddingModel,
+        boolean advancedInstalled,
+        String doclingVersion,
+        boolean engineReachable) {}

app/proprietary/src/main/java/stirling/software/proprietary/service/DocParseService.java

@@ -27,6 +27,7 @@ import stirling.software.proprietary.model.docparse.DocparseCapabilities;
 import stirling.software.proprietary.model.docparse.DocparseCapabilitiesView;
 import stirling.software.proprietary.model.docparse.DocparseMode;
 import stirling.software.proprietary.model.docparse.DocparseTier;
+import stirling.software.proprietary.model.docparse.DocumentStoreStats;
 import stirling.software.proprietary.model.docparse.ExtractFieldsRequest;
 import stirling.software.proprietary.model.docparse.ExtractFieldsResponse;
 import stirling.software.proprietary.model.docparse.ExtractTablesRequest;
@@ -35,8 +36,11 @@ import stirling.software.proprietary.model.docparse.FillDocxRequest;
 import stirling.software.proprietary.model.docparse.FillDocxResponse;
 import stirling.software.proprietary.model.docparse.ParseDocumentRequest;
 import stirling.software.proprietary.model.docparse.ParseDocumentResponse;
+import stirling.software.proprietary.model.docparse.RagAskRequest;
 import stirling.software.proprietary.model.docparse.RagIngestRequest;
 import stirling.software.proprietary.model.docparse.RagIngestResponse;
+import stirling.software.proprietary.model.docparse.RagSearchRequest;
+import stirling.software.proprietary.model.docparse.RagStatsView;
 import stirling.software.proprietary.model.docparse.SmartSplitRequest;
 import stirling.software.proprietary.model.docparse.SmartSplitResponse;
 import stirling.software.proprietary.model.docparse.SuggestSchemaRequest;
@@ -63,6 +67,10 @@ public class DocParseService {
     private static final String FILL_DOCX_ENDPOINT = "/api/v1/docparse/fill-docx";
     private static final String SUGGEST_SCHEMA_ENDPOINT = "/api/v1/docparse/suggest-schema";
     private static final String RAG_INGEST_ENDPOINT = "/api/v1/docparse/rag-ingest";
+    private static final String DOCUMENT_STATS_ENDPOINT = "/api/v1/documents/stats";
+    private static final String DOCUMENT_LIST_ENDPOINT = "/api/v1/documents/list";
+    private static final String DOCUMENT_SEARCH_ENDPOINT = "/api/v1/documents/search";
+    private static final String DOCUMENT_ASK_ENDPOINT = "/api/v1/documents/ask";
 
     /** Below this average of extractable chars per page the document is treated as scanned. */
     static final int SCANNED_AVG_CHARS_PER_PAGE = 100;
@@ -353,6 +361,61 @@ public class DocParseService {
         }
     }
 
+    /** Engine RAG store totals merged with the cached capability fields; graceful when down. */
+    public RagStatsView ragStats() {
+        DocparseCapabilities capabilities = capabilityService.capabilities();
+        try {
+            // The engine's documents routes are user-gated; an id-less probe 401s
+            // and would read as "engine offline" in the UI.
+            String json = aiEngineClient.get(DOCUMENT_STATS_ENDPOINT, currentUserId());
+            DocumentStoreStats stats = objectMapper.readValue(json, DocumentStoreStats.class);
+            return new RagStatsView(
+                    stats.backend(),
+                    stats.documents(),
+                    stats.chunks(),
+                    stats.embeddingModel(),
+                    capabilities.advancedInstalled(),
+                    capabilities.doclingVersion(),
+                    true);
+        } catch (Exception e) {
+            log.debug("RAG stats probe failed: {}", e.getMessage());
+            return new RagStatsView(
+                    null,
+                    0,
+                    0,
+                    null,
+                    capabilities.advancedInstalled(),
+                    capabilities.doclingVersion(),
+                    false);
+        }
+    }
+
+    /** Engine document-list passthrough; X-User-Id scopes it to the caller's ACLs. */
+    public String ragDocuments() throws IOException {
+        requireEnabled();
+        return aiEngineClient.get(DOCUMENT_LIST_ENDPOINT, currentUserId());
+    }
+
+    /** Semantic-search passthrough over the caller-visible RAG documents. */
+    public String ragSearch(String query, int topK) throws IOException {
+        requireEnabled();
+        requireNonBlank(query, "query");
+        RagSearchRequest request = new RagSearchRequest(query, Math.clamp(topK, 1, 50));
+        return aiEngineClient.post(
+                DOCUMENT_SEARCH_ENDPOINT,
+                objectMapper.writeValueAsString(request),
+                currentUserId());
+    }
+
+    /** Grounded-answer passthrough; long-running because local models answer slowly. */
+    public String ragAsk(String question, int topK) throws IOException {
+        requireEnabled();
+        requireNonBlank(question, "question");
+        RagAskRequest request = new RagAskRequest(question, Math.clamp(topK, 1, 20));
+        return aiEngineClient.postLongRunning(
+                DOCUMENT_ASK_ENDPOINT, objectMapper.writeValueAsString(request), currentUserId());
+    }
+
     /**
      * The settings mode wins when stricter: a settings {@code basic} always forces basic, a
      * settings {@code advanced} upgrades everything except an explicit basic request.

engine/src/stirling/agents/__init__.py

@@ -2,6 +2,7 @@
 
 from .document_classifier import DocumentClassifierAgent
 from .execution import ExecutionPlanningAgent
+from .knowledge_ask import KnowledgeAskAgent
 from .orchestrator import OrchestratorAgent
 from .pdf_create import PdfCreateAgent
 from .pdf_edit import PdfEditAgent, PdfEditParameterSelector, PdfEditPlanSelection
@@ -12,6 +13,7 @@ from .user_spec import UserSpecAgent
 __all__ = [
     "DocumentClassifierAgent",
     "ExecutionPlanningAgent",
+    "KnowledgeAskAgent",
     "OrchestratorAgent",
     "PdfCreateAgent",
     "PdfEditAgent",

engine/src/stirling/agents/knowledge_ask.py

@@ -0,0 +1,134 @@
+"""Grounded Q&A over the caller's stored documents.
+
+Retrieval runs the same cross-collection search as ``POST /documents/search``;
+one smart-model pass then answers only from the retrieved passages, citing
+document and page inline. No retrieval hit means a plain "not found" answer.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from pydantic import Field
+from pydantic_ai import Agent
+
+from stirling.agents.output_mode import output_retries, structured_output
+from stirling.contracts import AskDocumentsRequest, AskDocumentsResponse, DocumentPassage
+from stirling.documents import CollectionSearchHit
+from stirling.documents.service import PAGE_NUMBER_METADATA_KEY
+from stirling.models import ApiModel, PrincipalId
+from stirling.services import AppRuntime
+
+logger = logging.getLogger(__name__)
+
+# Metadata keys written by docparse rag-ingest (_chunk_metadata) for structure-aware chunks.
+_PAGE_START_KEY = "page_start"
+_PAGE_END_KEY = "page_end"
+_HEADING_PATH_KEY = "heading_path"
+_HEADING_PATH_SEPARATOR = " > "
+
+_SYSTEM_PROMPT = (
+    "You answer questions using ONLY the numbered passages you are given.\n"
+    "\n"
+    "Rules:\n"
+    "- Every statement must come from the passages. Never use outside knowledge, never guess.\n"
+    "- Cite the document and page inline right after each fact, "
+    'e.g. "(invoice.pdf p.2)" or "(report.pdf p.4-6)", using the names and pages '
+    "shown in each passage header.\n"
+    "- If the passages do not answer the question, say plainly that the stored "
+    "documents do not cover it. Do not attempt a partial guess.\n"
+    "- Answer in the same language as the question."
+)
+
+_NO_PASSAGES_ANSWER = "I couldn't find anything relevant to that question in your stored documents."
+
+
+class _AskOutput(ApiModel):
+    """Raw model answer for the single ask pass."""
+
+    answer: str = Field(description="The answer grounded in the passages, with inline citations.")
+
+
+def _meta_int(value: str | None) -> int | None:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except ValueError:
+        return None
+
+
+def passage_from_hit(hit: CollectionSearchHit) -> DocumentPassage:
+    """Map a store search hit onto the wire passage shape.
+
+    Docparse chunks carry page bounds and a heading path; plain page-text
+    chunks only carry ``page_number``, which maps to both bounds.
+    """
+    meta = hit.result.document.metadata
+    page_start = _meta_int(meta.get(_PAGE_START_KEY))
+    page_end = _meta_int(meta.get(_PAGE_END_KEY))
+    if page_start is None and page_end is None:
+        page_start = page_end = _meta_int(meta.get(PAGE_NUMBER_METADATA_KEY))
+    heading = meta.get(_HEADING_PATH_KEY)
+    source = meta.get("source")
+    if source and ":page:" in source:
+        # Page-text chunk sources look like "report.pdf:page:3"; show the file name.
+        source = source.rsplit(":page:", 1)[0]
+    return DocumentPassage(
+        document_id=hit.collection,
+        text=hit.result.document.text,
+        score=hit.result.score,
+        page_start=page_start,
+        page_end=page_end,
+        heading_path=heading.split(_HEADING_PATH_SEPARATOR) if heading else [],
+        source=source or None,
+    )
+
+
+def format_passages(passages: list[DocumentPassage]) -> str:
+    """Render passages for the prompt with the citation handle in each header."""
+    return "\n\n".join(_format_passage(i, passage) for i, passage in enumerate(passages, 1))
+
+
+def _format_passage(index: int, passage: DocumentPassage) -> str:
+    name = passage.source or passage.document_id
+    if passage.page_start is None:
+        pages = ""
+    elif passage.page_end is not None and passage.page_end != passage.page_start:
+        pages = f" p.{passage.page_start}-{passage.page_end}"
+    else:
+        pages = f" p.{passage.page_start}"
+    return f"[Passage {index} | {name}{pages}]\n{passage.text}"
+
+
+class KnowledgeAskAgent:
+    """Answers a question from the caller's stored documents.
+
+    Retrieves the top passages the caller can read (same path as the search
+    endpoint), then runs one smart-model pass over just those passages.
+    """
+
+    def __init__(self, runtime: AppRuntime) -> None:
+        self.runtime = runtime
+        # Ollama/custom block tool-calling under native json-schema output; see agents.output_mode.
+        provider = runtime.settings.chat_provider
+        self._agent: Agent[None, _AskOutput] = Agent(
+            model=runtime.smart_model,
+            output_type=structured_output([_AskOutput], chat_provider=provider),
+            system_prompt=_SYSTEM_PROMPT,
+            model_settings=runtime.smart_model_settings,
+            retries=output_retries(provider),
+        )
+
+    async def ask(self, request: AskDocumentsRequest, principals: list[PrincipalId]) -> AskDocumentsResponse:
+        hits = await self.runtime.documents.search_with_collections(
+            request.question, principals=principals, top_k=request.top_k
+        )
+        passages = [passage_from_hit(hit) for hit in hits]
+        if not passages:
+            logger.info("[knowledge-ask] question=%r -> 0 passages", request.question)
+            return AskDocumentsResponse(answer=_NO_PASSAGES_ANSWER, passages=[])
+        prompt = f"Question: {request.question}\n\nPassages:\n{format_passages(passages)}"
+        logger.debug("[knowledge-ask] prompt:\n%s", prompt)
+        result = await self._agent.run(prompt)
+        return AskDocumentsResponse(answer=result.output.answer, passages=passages)

engine/src/stirling/api/bootstrap.py

@@ -10,6 +10,7 @@ from pydantic_ai.models import Model
 from stirling.agents import (
     DocumentClassifierAgent,
     ExecutionPlanningAgent,
+    KnowledgeAskAgent,
     OrchestratorAgent,
     PdfEditAgent,
     PdfQuestionAgent,
@@ -36,6 +37,7 @@ class AppState:
     math_auditor_agent: MathAuditorAgent
     pdf_comment_agent: PdfCommentAgent
     document_classifier_agent: DocumentClassifierAgent
+    knowledge_ask_agent: KnowledgeAskAgent
     extract_fields_agent: ExtractFieldsAgent
     smart_split_agent: SmartSplitAgent
     suggest_schema_agent: SuggestSchemaAgent
@@ -67,6 +69,7 @@ def build_app_state(
         math_auditor_agent=MathAuditorAgent(runtime),
         pdf_comment_agent=PdfCommentAgent(runtime),
         document_classifier_agent=DocumentClassifierAgent(runtime),
+        knowledge_ask_agent=KnowledgeAskAgent(runtime),
         extract_fields_agent=ExtractFieldsAgent(runtime),
         smart_split_agent=SmartSplitAgent(runtime),
         suggest_schema_agent=SuggestSchemaAgent(runtime),

engine/src/stirling/api/dependencies.py

@@ -7,6 +7,7 @@ from fastapi import Depends, HTTPException, Request, status
 from stirling.agents import (
     DocumentClassifierAgent,
     ExecutionPlanningAgent,
+    KnowledgeAskAgent,
     OrchestratorAgent,
     PdfEditAgent,
     PdfQuestionAgent,
@@ -61,18 +62,22 @@ def get_document_classifier_agent(request: Request) -> DocumentClassifierAgent:
     return request.app.state.document_classifier_agent
 
 
+def get_knowledge_ask_agent(request: Request) -> KnowledgeAskAgent:
+    return request.app.state.knowledge_ask_agent
+
+
 def get_extract_fields_agent(request: Request) -> ExtractFieldsAgent:
     return request.app.state.extract_fields_agent
 
 
-def get_suggest_schema_agent(request: Request) -> SuggestSchemaAgent:
-    return request.app.state.suggest_schema_agent
-
-
 def get_smart_split_agent(request: Request) -> SmartSplitAgent:
     return request.app.state.smart_split_agent
 
 
+def get_suggest_schema_agent(request: Request) -> SuggestSchemaAgent:
+    return request.app.state.suggest_schema_agent
+
+
 def require_user_id() -> UserId:
     """FastAPI dependency for routes that touch per-user storage.
 

engine/src/stirling/api/routes/documents.py

@@ -5,13 +5,24 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends
 
-from stirling.api.dependencies import get_document_service, require_user_id
+from stirling.agents.knowledge_ask import KnowledgeAskAgent, passage_from_hit
+from stirling.api.dependencies import get_document_service, get_knowledge_ask_agent, require_user_id
+from stirling.config import load_settings
 from stirling.contracts import (
     DeleteDocumentResponse,
     IngestDocumentRequest,
     IngestDocumentResponse,
 )
-from stirling.contracts.documents import PurgeOwnerResponse
+from stirling.contracts.documents import (
+    AskDocumentsRequest,
+    AskDocumentsResponse,
+    DocumentStatsResponse,
+    DocumentSummary,
+    ListDocumentsResponse,
+    PurgeOwnerResponse,
+    SearchDocumentsRequest,
+    SearchDocumentsResponse,
+)
 from stirling.documents import DocumentService
 from stirling.models import FileId, OwnerId, PrincipalId, UserId
 
@@ -45,6 +56,70 @@ async def ingest_document(
     return IngestDocumentResponse(document_id=request.document_id, chunks_indexed=chunks_indexed)
 
 
+@router.get("/stats", response_model=DocumentStatsResponse)
+async def document_stats(
+    documents: Annotated[DocumentService, Depends(get_document_service)],
+    _user_id: Annotated[UserId, Depends(require_user_id)],
+) -> DocumentStatsResponse:
+    """Deployment-wide store counts for the admin dashboard.
+
+    Not tenant-filtered: counts cover every owner's content, so this reports
+    what the whole store holds, not what the caller can read.
+    """
+    settings = load_settings()
+    counts = await documents.stats()
+    return DocumentStatsResponse(
+        backend=settings.documents_backend.value,
+        documents=counts.documents,
+        chunks=counts.chunks,
+        embedding_model=settings.rag_embedding_model,
+    )
+
+
+@router.get("/list", response_model=ListDocumentsResponse)
+async def list_documents(
+    documents: Annotated[DocumentService, Depends(get_document_service)],
+    user_id: Annotated[UserId, Depends(require_user_id)],
+) -> ListDocumentsResponse:
+    """Per-document rollup of what the caller can read: distinct document ids
+    with their stored source label and chunk count. Never shows another
+    principal's documents.
+    """
+    summaries = await documents.list_documents([PrincipalId(user_id)])
+    return ListDocumentsResponse(
+        documents=[
+            DocumentSummary(document_id=FileId(s.collection), source=s.source, chunks=s.chunks) for s in summaries
+        ]
+    )
+
+
+@router.post("/search", response_model=SearchDocumentsResponse)
+async def search_documents(
+    request: SearchDocumentsRequest,
+    documents: Annotated[DocumentService, Depends(get_document_service)],
+    user_id: Annotated[UserId, Depends(require_user_id)],
+) -> SearchDocumentsResponse:
+    """Semantic search across every document the caller can read.
+
+    Same retrieval path as the RAG toolset: embed the query, search the
+    caller's readable collections, merge by score.
+    """
+    hits = await documents.search_with_collections(
+        request.query, principals=[PrincipalId(user_id)], top_k=request.top_k
+    )
+    return SearchDocumentsResponse(passages=[passage_from_hit(hit) for hit in hits])
+
+
+@router.post("/ask", response_model=AskDocumentsResponse)
+async def ask_documents(
+    request: AskDocumentsRequest,
+    agent: Annotated[KnowledgeAskAgent, Depends(get_knowledge_ask_agent)],
+    user_id: Annotated[UserId, Depends(require_user_id)],
+) -> AskDocumentsResponse:
+    """Answer a question from the caller's stored documents with inline citations."""
+    return await agent.ask(request, principals=[PrincipalId(user_id)])
+
+
 @router.delete("/by-id/{document_id}", response_model=DeleteDocumentResponse)
 async def delete_document(
     document_id: FileId,

engine/src/stirling/contracts/__init__.py

@@ -42,6 +42,36 @@ from .contradiction import (
     ContradictionReport,
     ContradictionSeverity,
 )
+from .docparse import (
+    BlockType,
+    ChunkDocumentRequest,
+    ChunkDocumentResponse,
+    DocBlock,
+    DocChunk,
+    DocparseCapabilities,
+    DocparseMode,
+    DocparseTier,
+    DocTable,
+    ExtractedField,
+    ExtractFieldsRequest,
+    ExtractFieldsResponse,
+    ExtractTablesRequest,
+    ExtractTablesResponse,
+    FieldCitation,
+    FillDocxRequest,
+    FillDocxResponse,
+    ParseDocumentRequest,
+    ParseDocumentResponse,
+    RagIngestRequest,
+    RagIngestResponse,
+    SmartSplitRequest,
+    SmartSplitResponse,
+    SplitPart,
+    SuggestedField,
+    SuggestedFieldType,
+    SuggestSchemaRequest,
+    SuggestSchemaResponse,
+)
 from .document_classifier import (
     ClassifyDocumentRequest,
     ClassifyDocumentResponse,
@@ -49,13 +79,21 @@ from .document_classifier import (
     LabelOption,
 )
 from .documents import (
+    AskDocumentsRequest,
+    AskDocumentsResponse,
     DeleteDocumentResponse,
+    DocumentPassage,
+    DocumentStatsResponse,
+    DocumentSummary,
     IngestDocumentRequest,
     IngestDocumentResponse,
+    ListDocumentsResponse,
     Page,
     PageRange,
     PageText,
     PurgeOwnerResponse,
+    SearchDocumentsRequest,
+    SearchDocumentsResponse,
 )
 from .execution import (
     AgentExecutionRequest,
@@ -140,10 +178,15 @@ __all__ = [
     "AiFile",
     "AiToolAgentStep",
     "ArtifactKind",
+    "AskDocumentsRequest",
+    "AskDocumentsResponse",
+    "BlockType",
     "CannotContinueExecutionAction",
+    "ChunkDocumentRequest",
+    "ChunkDocumentResponse",
+    "Claim",
     "ClassifyDocumentRequest",
     "ClassifyDocumentResponse",
-    "Claim",
     "CommentSpec",
     "CompletedExecutionAction",
     "ConfigApplyResponse",
@@ -156,30 +199,45 @@ __all__ = [
     "ContradictionSeverity",
     "ConversationMessage",
     "DeleteDocumentResponse",
-    "PurgeOwnerResponse",
     "Discrepancy",
-    "DocumentClassificationResponse",
-    "LabelOption",
-    "DocumentMeta",
-    "DocumentSections",
     "DiscrepancyKind",
+    "DocBlock",
+    "DocChunk",
+    "DocTable",
+    "DocparseCapabilities",
+    "DocparseMode",
+    "DocparseTier",
+    "DocumentClassificationResponse",
+    "DocumentMeta",
+    "DocumentPassage",
+    "DocumentSections",
+    "DocumentStatsResponse",
+    "DocumentSummary",
     "EditCannotDoResponse",
     "EditClarificationRequest",
     "EditPlanResponse",
     "Evidence",
     "ExecutionContext",
     "ExecutionStepResult",
+    "ExtractFieldsRequest",
+    "ExtractFieldsResponse",
+    "ExtractTablesRequest",
+    "ExtractTablesResponse",
+    "ExtractedField",
     "ExtractedFileText",
     "ExtractedTextArtifact",
+    "FieldCitation",
+    "FillDocxRequest",
+    "FillDocxResponse",
     "Folio",
     "FolioManifest",
     "FolioType",
-    "format_conversation_history",
-    "format_file_names",
     "GenerateFileResponse",
     "HealthResponse",
     "IngestDocumentRequest",
     "IngestDocumentResponse",
+    "LabelOption",
+    "ListDocumentsResponse",
     "MathAuditorToolReportArtifact",
     "NeedContentFileRequest",
     "NeedContentResponse",
@@ -190,6 +248,8 @@ __all__ = [
     "Page",
     "PageRange",
     "PageText",
+    "ParseDocumentRequest",
+    "ParseDocumentResponse",
     "PdfCommentInstruction",
     "PdfCommentReport",
     "PdfCommentRequest",
@@ -212,9 +272,21 @@ __all__ = [
     "PdfReviewOrchestrateResponse",
     "PdfTextSelection",
     "ProgressEvent",
+    "PurgeOwnerResponse",
+    "RagIngestRequest",
+    "RagIngestResponse",
     "Requisition",
+    "SearchDocumentsRequest",
+    "SearchDocumentsResponse",
     "Severity",
+    "SmartSplitRequest",
+    "SmartSplitResponse",
+    "SplitPart",
     "StepKind",
+    "SuggestSchemaRequest",
+    "SuggestSchemaResponse",
+    "SuggestedField",
+    "SuggestedFieldType",
     "SupportedCapability",
     "TextChunk",
     "ToolCallExecutionAction",
@@ -228,4 +300,6 @@ __all__ = [
     "WholeDocSliceDone",
     "WorkflowArtifact",
     "WorkflowOutcome",
+    "format_conversation_history",
+    "format_file_names",
 ]

engine/src/stirling/contracts/documents.py

@@ -75,3 +75,65 @@ class PurgeOwnerResponse(ApiModel):
 
     owner_id: OwnerId
     deleted: int = Field(ge=0)
+
+
+class DocumentStatsResponse(ApiModel):
+    """Returned by ``GET /api/v1/documents/stats``. Deployment-wide counts
+    (every owner's content) powering the admin dashboard."""
+
+    backend: str
+    documents: int = Field(ge=0)
+    chunks: int = Field(ge=0)
+    embedding_model: str
+
+
+class DocumentSummary(ApiModel):
+    """One stored document the caller can read: its id, source label, chunk count."""
+
+    document_id: FileId
+    source: str
+    chunks: int = Field(ge=0)
+
+
+class ListDocumentsResponse(ApiModel):
+    """Returned by ``GET /api/v1/documents/list``. Caller-scoped rollup."""
+
+    documents: list[DocumentSummary]
+
+
+class SearchDocumentsRequest(ApiModel):
+    """Semantic search over every document the caller can read."""
+
+    query: str = Field(min_length=1)
+    top_k: int = Field(default=8, ge=1, le=50)
+
+
+class DocumentPassage(ApiModel):
+    """A retrieved chunk on the wire. Page bounds and heading path come from
+    chunk metadata when present (docparse chunks carry them); nulls otherwise."""
+
+    document_id: FileId
+    text: str
+    score: float
+    page_start: int | None = None
+    page_end: int | None = None
+    heading_path: list[str] = Field(default_factory=list)
+    source: str | None = None
+
+
+class SearchDocumentsResponse(ApiModel):
+    passages: list[DocumentPassage]
+
+
+class AskDocumentsRequest(ApiModel):
+    """Question answered only from the caller's stored documents."""
+
+    question: str = Field(min_length=1)
+    top_k: int = Field(default=8, ge=1, le=20)
+
+
+class AskDocumentsResponse(ApiModel):
+    """Grounded answer with inline citations plus the passages it drew from."""
+
+    answer: str
+    passages: list[DocumentPassage]

engine/src/stirling/documents/__init__.py

@@ -3,11 +3,20 @@ from __future__ import annotations
 from stirling.documents.embedder import EmbeddingService
 from stirling.documents.pgvector_store import PgVectorStore
 from stirling.documents.rag_capability import RagCapability
-from stirling.documents.service import DocumentService
+from stirling.documents.service import CollectionSearchHit, DocumentService
 from stirling.documents.sqlite_vec_store import SqliteVecStore
-from stirling.documents.store import Document, DocumentStore, SearchResult, StoredPage
+from stirling.documents.store import (
+    CollectionSummary,
+    Document,
+    DocumentStore,
+    SearchResult,
+    StoredPage,
+    StoreStats,
+)
 
 __all__ = [
+    "CollectionSearchHit",
+    "CollectionSummary",
     "Document",
     "DocumentService",
     "DocumentStore",
@@ -16,5 +25,6 @@ __all__ = [
     "RagCapability",
     "SearchResult",
     "SqliteVecStore",
+    "StoreStats",
     "StoredPage",
 ]

engine/src/stirling/documents/pgvector_store.py

@@ -10,7 +10,14 @@ from pgvector.psycopg import register_vector_async
 from psycopg_pool import AsyncConnectionPool
 
 from stirling.contracts.documents import Page, PageRange
-from stirling.documents.store import Document, DocumentStore, SearchResult, StoredPage
+from stirling.documents.store import (
+    CollectionSummary,
+    Document,
+    DocumentStore,
+    SearchResult,
+    StoredPage,
+    StoreStats,
+)
 from stirling.models import OwnerId, PrincipalId
 
 _READ_PERMISSION = "read"
@@ -411,5 +418,44 @@ class PgVectorStore(DocumentStore):
                 rows = await cur.fetchall()
         return [r[0] for r in rows]
 
+    async def list_collection_summaries(self, principals: list[PrincipalId]) -> list[CollectionSummary]:
+        if not principals:
+            return []
+        await self._ensure_ready()
+        async with self._pool.connection() as conn:
+            async with conn.cursor() as cur:
+                # MIN(owner_id) mirrors _readable_owner_for's ORDER BY owner_id LIMIT 1.
+                await cur.execute(
+                    """
+                    SELECT r.collection, m.source, COUNT(d.id)
+                    FROM (
+                        SELECT collection, MIN(owner_id) AS owner_id
+                        FROM document_acl
+                        WHERE permission = %s AND principal_id = ANY(%s)
+                        GROUP BY collection
+                    ) r
+                    JOIN documents_meta m ON m.collection = r.collection AND m.owner_id = r.owner_id
+                    LEFT JOIN rag_documents d ON d.collection = r.collection AND d.owner_id = r.owner_id
+                    GROUP BY r.collection, m.source
+                    ORDER BY r.collection
+                    """,
+                    (_READ_PERMISSION, list(principals)),
+                )
+                rows = await cur.fetchall()
+        return [CollectionSummary(collection=r[0], source=r[1], chunks=int(r[2])) for r in rows]
+
+    async def stats(self) -> StoreStats:
+        await self._ensure_ready()
+        async with self._pool.connection() as conn:
+            async with conn.cursor() as cur:
+                await cur.execute("SELECT COUNT(DISTINCT collection) FROM documents_meta")
+                doc_row = await cur.fetchone()
+                await cur.execute("SELECT COUNT(*) FROM rag_documents")
+                chunk_row = await cur.fetchone()
+        return StoreStats(
+            documents=int(doc_row[0]) if doc_row else 0,
+            chunks=int(chunk_row[0]) if chunk_row else 0,
+        )
+
     async def close(self) -> None:
         await self._pool.close()

engine/src/stirling/documents/service.py

@@ -1,15 +1,32 @@
 from __future__ import annotations
 
 import logging
+from dataclasses import dataclass
 from datetime import datetime
 
 from stirling.contracts.documents import Page, PageRange, PageText
 from stirling.documents.embedder import EmbeddingService
-from stirling.documents.store import Document, DocumentStore, SearchResult, StoredPage
+from stirling.documents.store import (
+    CollectionSummary,
+    Document,
+    DocumentStore,
+    SearchResult,
+    StoredPage,
+    StoreStats,
+)
 from stirling.models import FileId, OwnerId, PrincipalId
 
 logger = logging.getLogger(__name__)
 
+
+@dataclass(frozen=True)
+class CollectionSearchHit:
+    """A search result tagged with the collection it came from."""
+
+    collection: FileId
+    result: SearchResult
+
+
 PAGE_NUMBER_METADATA_KEY = "page_number"
 CONTENT_TYPE_METADATA_KEY = "content_type"
 PAGE_TEXT_CONTENT_TYPE = "page_text"
@@ -187,6 +204,32 @@ class DocumentService:
         all_results.sort(key=lambda r: r.score, reverse=True)
         return all_results[:k]
 
+    async def search_with_collections(
+        self,
+        query: str,
+        principals: list[PrincipalId],
+        top_k: int | None = None,
+    ) -> list[CollectionSearchHit]:
+        """Cross-collection search like :meth:`search`, but every result keeps
+        the collection it came from. Restricted to what ``principals`` can read.
+        """
+        k = top_k if top_k is not None else self._default_top_k
+        query_embedding = await self._embedder.embed_query(query)
+        hits: list[CollectionSearchHit] = []
+        for col_name in await self._store.list_collections(principals):
+            try:
+                results = await self._store.search(col_name, query_embedding, k, principals)
+            except Exception:  # noqa: BLE001 - any backend error on one collection should not stop the others
+                logger.warning(
+                    "Skipping collection %s during cross-collection search",
+                    col_name,
+                    exc_info=True,
+                )
+                continue
+            hits.extend(CollectionSearchHit(collection=FileId(col_name), result=r) for r in results)
+        hits.sort(key=lambda hit: hit.result.score, reverse=True)
+        return hits[:k]
+
     async def read_pages(
         self,
         collection: FileId,
@@ -223,6 +266,10 @@ class DocumentService:
         """List collections readable by at least one of ``principals``."""
         return [FileId(name) for name in await self._store.list_collections(principals)]
 
+    async def list_documents(self, principals: list[PrincipalId]) -> list[CollectionSummary]:
+        """Per-document rollup (source, chunk count) readable by ``principals``."""
+        return await self._store.list_collection_summaries(principals)
+
     async def grant_read(
         self,
         collection: FileId,
@@ -241,6 +288,10 @@ class DocumentService:
         """Revoke a principal's access on an existing doc."""
         await self._store.revoke(collection, owner_id, principal)
 
+    async def stats(self) -> StoreStats:
+        """Deployment-wide document/chunk counts from the backing store."""
+        return await self._store.stats()
+
     async def close(self) -> None:
         """Release the underlying store's resources."""
         await self._store.close()

engine/src/stirling/documents/sqlite_vec_store.py

@@ -11,7 +11,14 @@ from pathlib import Path
 import sqlite_vec
 
 from stirling.contracts.documents import Page, PageRange
-from stirling.documents.store import Document, DocumentStore, SearchResult, StoredPage
+from stirling.documents.store import (
+    CollectionSummary,
+    Document,
+    DocumentStore,
+    SearchResult,
+    StoredPage,
+    StoreStats,
+)
 from stirling.models import OwnerId, PrincipalId
 
 _READ_PERMISSION = "read"
@@ -538,6 +545,42 @@ class SqliteVecStore(DocumentStore):
         ).fetchall()
         return [r[0] for r in rows]
 
+    async def list_collection_summaries(self, principals: list[PrincipalId]) -> list[CollectionSummary]:
+        async with self._lock:
+            return await asyncio.to_thread(self._sync_list_collection_summaries, principals)
+
+    def _sync_list_collection_summaries(self, principals: list[PrincipalId]) -> list[CollectionSummary]:
+        if not principals:
+            return []
+        placeholders = ",".join("?" * len(principals))
+        # MIN(owner_id) mirrors _readable_owner_for's ORDER BY owner_id LIMIT 1.
+        rows = self._conn.execute(
+            f"""
+            SELECT r.collection, m.source, COUNT(d.id)
+            FROM (
+                SELECT collection, MIN(owner_id) AS owner_id
+                FROM document_acl
+                WHERE permission = ? AND principal_id IN ({placeholders})
+                GROUP BY collection
+            ) r
+            JOIN documents_meta m ON m.collection = r.collection AND m.owner_id = r.owner_id
+            LEFT JOIN documents d ON d.collection = r.collection AND d.owner_id = r.owner_id
+            GROUP BY r.collection, m.source
+            ORDER BY r.collection
+            """,
+            (_READ_PERMISSION, *principals),
+        ).fetchall()
+        return [CollectionSummary(collection=r[0], source=r[1], chunks=int(r[2])) for r in rows]
+
+    async def stats(self) -> StoreStats:
+        async with self._lock:
+            return await asyncio.to_thread(self._sync_stats)
+
+    def _sync_stats(self) -> StoreStats:
+        documents = self._conn.execute("SELECT COUNT(DISTINCT collection) FROM documents_meta").fetchone()[0]
+        chunks = self._conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
+        return StoreStats(documents=int(documents), chunks=int(chunks))
+
     async def close(self) -> None:
         async with self._lock:
             await asyncio.to_thread(self._sync_close)

engine/src/stirling/documents/store.py

@@ -34,6 +34,23 @@ class StoredPage:
     char_count: int
 
 
+@dataclass
+class StoreStats:
+    """Deployment-wide counts: distinct document ids and total vector-chunk rows."""
+
+    documents: int
+    chunks: int
+
+
+@dataclass
+class CollectionSummary:
+    """Rollup row for one readable collection: stored source label + chunk count."""
+
+    collection: str
+    source: str
+    chunks: int
+
+
 class DocumentStore(ABC):
     """Abstract interface for document storage backends.
 
@@ -148,6 +165,20 @@ class DocumentStore(ABC):
     async def list_collections(self, principals: list[PrincipalId]) -> list[str]:
         """Return collection names readable by at least one of ``principals``."""
 
+    @abstractmethod
+    async def list_collection_summaries(self, principals: list[PrincipalId]) -> list[CollectionSummary]:
+        """Per-collection rollup (source + chunk count) readable by ``principals``.
+
+        Counts cover the same owner's copy a read would resolve to, so the
+        rollup never leaks another tenant's content.
+        """
+
+    # ── deployment-wide stats (not tenant-scoped) ──────────────────────────
+
+    @abstractmethod
+    async def stats(self) -> StoreStats:
+        """Count every owner's content: distinct document ids + total chunk rows."""
+
     # ── lifecycle ──────────────────────────────────────────────────────────
 
     @abstractmethod

engine/tests/test_documents.py

@@ -7,7 +7,7 @@ from stirling.documents.chunker import chunk_text
 from stirling.documents.rag_capability import RagCapability
 from stirling.documents.service import DocumentService
 from stirling.documents.sqlite_vec_store import SqliteVecStore
-from stirling.documents.store import Document, SearchResult
+from stirling.documents.store import CollectionSummary, Document, SearchResult
 from stirling.models import FileId, OwnerId, PrincipalId
 
 # Personal-doc tests reuse the same opaque string in all three roles — keeps the
@@ -178,6 +178,27 @@ class TestSqliteVecStore:
         assert await store.list_collections(OWNER_PRINCIPALS) == []
         assert await store.list_collections(OTHER_OWNER_PRINCIPALS) == ["c.pdf"]
 
+    @pytest.mark.anyio
+    async def test_stats_count_distinct_document_ids_and_chunk_rows(self) -> None:
+        """Stats span every owner; a document id shared by two owners counts once."""
+        store = SqliteVecStore.ephemeral()
+        empty = await store.stats()
+        assert (empty.documents, empty.chunks) == (0, 0)
+
+        for owner, principals, name, texts in (
+            (OWNER, OWNER_PRINCIPALS, "doc-a", ["one", "two"]),
+            (OWNER, OWNER_PRINCIPALS, "doc-b", ["three"]),
+            (OTHER_OWNER, OTHER_OWNER_PRINCIPALS, "doc-b", ["four"]),
+        ):
+            await store.ensure_collection(name, f"{name}.pdf", owner, None)
+            await store.grant_read(name, owner, principals)
+            docs = [Document(id=str(i), text=t, metadata={}) for i, t in enumerate(texts)]
+            await store.add_documents(name, docs, [[1.0, 0.0]] * len(docs), owner)
+
+        stats = await store.stats()
+        assert stats.documents == 2
+        assert stats.chunks == 4
+
     @pytest.mark.anyio
     async def test_reap_expired_drops_collections_past_expires_at(self) -> None:
         """TTL backstop: rows with ``expires_at`` in the past go away on reap."""
@@ -239,6 +260,47 @@ class TestSqliteVecStore:
         # Owner still can.
         assert await store.has_collection("doc", OWNER_PRINCIPALS) is True
 
+    @pytest.mark.anyio
+    async def test_list_collection_summaries_rolls_up_readable_collections(self) -> None:
+        """Rollup: one row per readable collection with its source and chunk count."""
+        store = SqliteVecStore.ephemeral()
+        await store.ensure_collection("doc-a", "a.pdf", OWNER, None)
+        await store.grant_read("doc-a", OWNER, OWNER_PRINCIPALS)
+        docs = [Document(id="1", text="one", metadata={}), Document(id="2", text="two", metadata={})]
+        await store.add_documents("doc-a", docs, [[1.0, 0.0], [0.0, 1.0]], OWNER)
+        # Collection with no vector chunks yet: still listed, zero count.
+        await store.ensure_collection("doc-b", "b.pdf", OWNER, None)
+        await store.grant_read("doc-b", OWNER, OWNER_PRINCIPALS)
+
+        summaries = await store.list_collection_summaries(OWNER_PRINCIPALS)
+        assert summaries == [
+            CollectionSummary(collection="doc-a", source="a.pdf", chunks=2),
+            CollectionSummary(collection="doc-b", source="b.pdf", chunks=0),
+        ]
+
+    @pytest.mark.anyio
+    async def test_list_collection_summaries_scoped_to_principals(self) -> None:
+        """One principal's rollup never lists, or counts, another owner's copy."""
+        store = SqliteVecStore.ephemeral()
+        await store.ensure_collection("shared-id", "alice.pdf", OWNER, None)
+        await store.grant_read("shared-id", OWNER, OWNER_PRINCIPALS)
+        await store.add_documents("shared-id", [Document(id="1", text="alice", metadata={})], [[1.0, 0.0]], OWNER)
+        await store.ensure_collection("shared-id", "bob.pdf", OTHER_OWNER, None)
+        await store.grant_read("shared-id", OTHER_OWNER, OTHER_OWNER_PRINCIPALS)
+        bob_docs = [Document(id="1", text="bob", metadata={}), Document(id="2", text="bob2", metadata={})]
+        await store.add_documents("shared-id", bob_docs, [[1.0, 0.0], [0.0, 1.0]], OTHER_OWNER)
+        await store.ensure_collection("bob-only", "bob-only.pdf", OTHER_OWNER, None)
+        await store.grant_read("bob-only", OTHER_OWNER, OTHER_OWNER_PRINCIPALS)
+
+        assert await store.list_collection_summaries(OWNER_PRINCIPALS) == [
+            CollectionSummary(collection="shared-id", source="alice.pdf", chunks=1)
+        ]
+        assert await store.list_collection_summaries(OTHER_OWNER_PRINCIPALS) == [
+            CollectionSummary(collection="bob-only", source="bob-only.pdf", chunks=0),
+            CollectionSummary(collection="shared-id", source="bob.pdf", chunks=2),
+        ]
+        assert await store.list_collection_summaries([]) == []
+
 
 # DocumentService (with stub embedder)
 
@@ -398,6 +460,31 @@ class TestDocumentService:
         multi_results = await documents.search("deploy", principals=[hr_group, eng_group])
         assert len(multi_results) > 0
 
+    @pytest.mark.anyio
+    async def test_search_with_collections_tags_results_and_respects_acl(self, documents: DocumentService) -> None:
+        """Collection-tagged search only reaches collections the caller can read."""
+        await documents.ingest(
+            FileId("col-a"),
+            _pages("Alpha content."),
+            source="a.pdf",
+            owner_id=OWNER,
+            read_principals=OWNER_PRINCIPALS,
+            expires_at=None,
+        )
+        await documents.ingest(
+            FileId("col-b"),
+            _pages("Beta content."),
+            source="b.pdf",
+            owner_id=OTHER_OWNER,
+            read_principals=OTHER_OWNER_PRINCIPALS,
+            expires_at=None,
+        )
+
+        hits = await documents.search_with_collections("content", principals=OWNER_PRINCIPALS)
+        assert hits
+        assert {hit.collection for hit in hits} == {"col-a"}
+        assert all(hit.result.document.text for hit in hits)
+
     @pytest.mark.anyio
     async def test_delete_collection(self, documents: DocumentService) -> None:
         await documents.ingest(

engine/tests/test_documents_routes.py

@@ -6,9 +6,10 @@ import pytest
 from fastapi.testclient import TestClient
 
 from stirling.api import app
-from stirling.api.dependencies import get_document_service
+from stirling.api.dependencies import get_document_service, get_knowledge_ask_agent
+from stirling.contracts import AskDocumentsRequest, AskDocumentsResponse, DocumentPassage
 from stirling.documents import Document, DocumentService, SqliteVecStore
-from stirling.models import FileId, PrincipalId, UserId
+from stirling.models import FileId, OwnerId, PrincipalId, UserId
 
 USER = UserId("test-user")
 USER_PRINCIPALS = [PrincipalId("test-user")]
@@ -348,6 +349,274 @@ def test_purge_by_owner_rejects_missing_user_header(client: TestClient) -> None:
     assert response.status_code == 401
 
 
+# ── GET /documents/list ─────────────────────────────────────────────────
+
+
+def _ingest(client: TestClient, document_id: str, source: str, texts: list[str], owner: str) -> None:
+    client.post(
+        "/api/v1/documents",
+        json={
+            "documentId": document_id,
+            "source": source,
+            "pageText": [{"pageNumber": i, "text": t} for i, t in enumerate(texts, 1)],
+            "ownerId": owner,
+            "readPrincipals": [owner],
+            "expiresAt": None,
+        },
+        headers={"X-User-Id": owner},
+    )
+
+
+def test_list_documents_returns_caller_rollup(client: TestClient) -> None:
+    _ingest(client, "list-a", "a.pdf", ["Page one text.", "Page two text."], USER)
+    _ingest(client, "list-b", "b.pdf", ["Only page."], USER)
+
+    response = client.get("/api/v1/documents/list", headers=HEADERS)
+    assert response.status_code == 200
+    documents = response.json()["documents"]
+    assert [d["documentId"] for d in documents] == ["list-a", "list-b"]
+    by_id = {d["documentId"]: d for d in documents}
+    assert by_id["list-a"]["source"] == "a.pdf"
+    assert by_id["list-a"]["chunks"] >= 2
+    assert by_id["list-b"]["source"] == "b.pdf"
+    assert by_id["list-b"]["chunks"] >= 1
+
+
+def test_list_documents_empty_for_new_user(client: TestClient) -> None:
+    response = client.get("/api/v1/documents/list", headers=HEADERS)
+    assert response.status_code == 200
+    assert response.json() == {"documents": []}
+
+
+def test_list_documents_hides_other_users_documents(client: TestClient) -> None:
+    """User A must never see user B's documents in the rollup."""
+    _ingest(client, "alice-doc", "alice.pdf", ["alice content"], "alice")
+    _ingest(client, "bob-doc", "bob.pdf", ["bob content"], "bob")
+
+    alice_docs = client.get("/api/v1/documents/list", headers={"X-User-Id": "alice"}).json()["documents"]
+    bob_docs = client.get("/api/v1/documents/list", headers={"X-User-Id": "bob"}).json()["documents"]
+    assert [d["documentId"] for d in alice_docs] == ["alice-doc"]
+    assert [d["documentId"] for d in bob_docs] == ["bob-doc"]
+
+
+def test_list_documents_rejects_missing_user_header(client: TestClient) -> None:
+    assert client.get("/api/v1/documents/list").status_code == 401
+
+
+# ── POST /documents/search ──────────────────────────────────────────────
+
+
+def test_search_documents_maps_page_text_chunks(client: TestClient) -> None:
+    """Plain ingested chunks only carry page_number: both bounds map to it and
+    the ":page:N" suffix is stripped off the source."""
+    client.post(
+        "/api/v1/documents",
+        json={
+            "documentId": "report",
+            "source": "report.pdf",
+            "pageText": [{"pageNumber": 3, "text": "The launch is planned for October."}],
+            "ownerId": USER,
+            "readPrincipals": [USER],
+            "expiresAt": None,
+        },
+        headers=HEADERS,
+    )
+
+    response = client.post("/api/v1/documents/search", json={"query": "launch", "topK": 5}, headers=HEADERS)
+    assert response.status_code == 200
+    passages = response.json()["passages"]
+    assert len(passages) >= 1
+    passage = passages[0]
+    assert passage["documentId"] == "report"
+    assert passage["pageStart"] == 3
+    assert passage["pageEnd"] == 3
+    assert passage["headingPath"] == []
+    assert passage["source"] == "report.pdf"
+    assert "launch" in passage["text"]
+    assert isinstance(passage["score"], float)
+
+
+@pytest.mark.anyio
+async def test_search_documents_maps_docparse_chunk_metadata(client: TestClient, service: DocumentService) -> None:
+    """Docparse chunks carry page bounds + heading path; they map straight onto the wire."""
+    await service.ingest_prepared(
+        collection=FileId("dp-doc"),
+        chunks=[
+            (
+                "Revenue grew 12% in Q2.",
+                {
+                    "content_type": "docparse_chunk",
+                    "page_start": "2",
+                    "page_end": "3",
+                    "heading_path": "Report > Finance",
+                },
+            )
+        ],
+        source="q2.pdf",
+        owner_id=OwnerId(USER),
+        read_principals=USER_PRINCIPALS,
+        expires_at=None,
+    )
+
+    response = client.post("/api/v1/documents/search", json={"query": "revenue"}, headers=HEADERS)
+    assert response.status_code == 200
+    passage = response.json()["passages"][0]
+    assert passage["documentId"] == "dp-doc"
+    assert passage["pageStart"] == 2
+    assert passage["pageEnd"] == 3
+    assert passage["headingPath"] == ["Report", "Finance"]
+    assert passage["source"] == "q2.pdf"
+
+
+def test_search_documents_cannot_see_other_users_documents(client: TestClient) -> None:
+    """User B searching for user A's content must get nothing back."""
+    _ingest(client, "alice-doc", "alice.pdf", ["The secret launch code is October."], "alice")
+
+    bob = client.post("/api/v1/documents/search", json={"query": "secret launch"}, headers={"X-User-Id": "bob"})
+    assert bob.status_code == 200
+    assert bob.json()["passages"] == []
+
+    alice = client.post("/api/v1/documents/search", json={"query": "secret launch"}, headers={"X-User-Id": "alice"})
+    assert alice.json()["passages"] != []
+
+
+def test_search_documents_rejects_empty_query(client: TestClient) -> None:
+    response = client.post("/api/v1/documents/search", json={"query": ""}, headers=HEADERS)
+    assert response.status_code == 422
+
+
+def test_search_documents_rejects_top_k_above_cap(client: TestClient) -> None:
+    response = client.post("/api/v1/documents/search", json={"query": "x", "topK": 51}, headers=HEADERS)
+    assert response.status_code == 422
+
+
+def test_search_documents_rejects_missing_user_header(client: TestClient) -> None:
+    assert client.post("/api/v1/documents/search", json={"query": "x"}).status_code == 401
+
+
+# ── POST /documents/ask ─────────────────────────────────────────────────
+
+
+class StubKnowledgeAskAgent:
+    """Stands in for KnowledgeAskAgent so route tests don't call a model."""
+
+    def __init__(self, response: AskDocumentsResponse) -> None:
+        self._response = response
+        self.calls: list[tuple[AskDocumentsRequest, list[PrincipalId]]] = []
+
+    async def ask(self, request: AskDocumentsRequest, principals: list[PrincipalId]) -> AskDocumentsResponse:
+        self.calls.append((request, principals))
+        return self._response
+
+
+@pytest.fixture
+def ask_agent() -> StubKnowledgeAskAgent:
+    return StubKnowledgeAskAgent(
+        AskDocumentsResponse(
+            answer="Revenue grew 12% (q2.pdf p.2).",
+            passages=[
+                DocumentPassage(
+                    document_id=FileId("dp-doc"),
+                    text="Revenue grew 12% in Q2.",
+                    score=0.91,
+                    page_start=2,
+                    page_end=3,
+                    heading_path=["Report", "Finance"],
+                    source="q2.pdf",
+                )
+            ],
+        )
+    )
+
+
+@pytest.fixture
+def ask_client(client: TestClient, ask_agent: StubKnowledgeAskAgent) -> Iterator[TestClient]:
+    app.dependency_overrides[get_knowledge_ask_agent] = lambda: ask_agent
+    try:
+        yield client
+    finally:
+        app.dependency_overrides.pop(get_knowledge_ask_agent, None)
+
+
+def test_ask_documents_returns_answer_and_passages(ask_client: TestClient) -> None:
+    response = ask_client.post("/api/v1/documents/ask", json={"question": "How did revenue do?"}, headers=HEADERS)
+    assert response.status_code == 200
+    body = response.json()
+    assert body["answer"] == "Revenue grew 12% (q2.pdf p.2)."
+    assert body["passages"] == [
+        {
+            "documentId": "dp-doc",
+            "text": "Revenue grew 12% in Q2.",
+            "score": 0.91,
+            "pageStart": 2,
+            "pageEnd": 3,
+            "headingPath": ["Report", "Finance"],
+            "source": "q2.pdf",
+        }
+    ]
+
+
+def test_ask_documents_scopes_to_calling_user(ask_client: TestClient, ask_agent: StubKnowledgeAskAgent) -> None:
+    """The route hands the agent exactly the caller's principal set."""
+    ask_client.post("/api/v1/documents/ask", json={"question": "anything"}, headers=HEADERS)
+    request, principals = ask_agent.calls[0]
+    assert principals == [PrincipalId(USER)]
+    assert request.top_k == 8
+
+
+def test_ask_documents_rejects_empty_question(ask_client: TestClient) -> None:
+    response = ask_client.post("/api/v1/documents/ask", json={"question": ""}, headers=HEADERS)
+    assert response.status_code == 422
+
+
+def test_ask_documents_rejects_top_k_above_cap(ask_client: TestClient) -> None:
+    response = ask_client.post("/api/v1/documents/ask", json={"question": "x", "topK": 21}, headers=HEADERS)
+    assert response.status_code == 422
+
+
+def test_ask_documents_rejects_missing_user_header(ask_client: TestClient) -> None:
+    assert ask_client.post("/api/v1/documents/ask", json={"question": "x"}).status_code == 401
+
+
+# ── GET /documents/stats ────────────────────────────────────────────────
+
+
+def test_stats_on_empty_store_reports_zero(client: TestClient) -> None:
+    response = client.get("/api/v1/documents/stats", headers=HEADERS)
+    assert response.status_code == 200
+    body = response.json()
+    assert body["documents"] == 0
+    assert body["chunks"] == 0
+    assert body["backend"] in ("sqlite", "pgvector")
+    assert body["embeddingModel"]
+
+
+def test_stats_counts_seeded_documents_across_owners(client: TestClient) -> None:
+    """Stats are deployment-wide: both owners' content is counted."""
+    for owner, doc in (("alice", "doc-a"), ("bob", "doc-b")):
+        client.post(
+            "/api/v1/documents",
+            json={
+                "documentId": doc,
+                "source": f"{doc}.pdf",
+                "pageText": [{"pageNumber": 1, "text": "Some content for the stats endpoint."}],
+                "ownerId": owner,
+                "readPrincipals": [owner],
+                "expiresAt": None,
+            },
+            headers={"X-User-Id": owner},
+        )
+    response = client.get("/api/v1/documents/stats", headers=HEADERS)
+    assert response.status_code == 200
+    body = response.json()
+    assert body["documents"] == 2
+    assert body["chunks"] >= 2
+
+
+def test_stats_rejects_missing_user_header(client: TestClient) -> None:
+    assert client.get("/api/v1/documents/stats").status_code == 401
+
+
 def test_delete_document_only_affects_calling_user(client: TestClient) -> None:
     """Two users with the same document id: one user's delete must not remove the other's."""
     alice_body = {

engine/tests/test_knowledge_ask.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import pytest
+
+from stirling.agents.knowledge_ask import KnowledgeAskAgent, format_passages, passage_from_hit
+from stirling.config import AppSettings
+from stirling.contracts import AskDocumentsRequest, DocumentPassage
+from stirling.documents import CollectionSearchHit, Document, DocumentService, SearchResult, SqliteVecStore
+from stirling.models import FileId, PrincipalId
+from stirling.services import build_runtime
+
+PRINCIPALS = [PrincipalId("test-user")]
+
+
+def _hit(metadata: dict[str, str], text: str = "chunk text", score: float = 0.8) -> CollectionSearchHit:
+    return CollectionSearchHit(
+        collection=FileId("doc-1"),
+        result=SearchResult(document=Document(id="c1", text=text, metadata=metadata), score=score),
+    )
+
+
+# ── passage_from_hit ────────────────────────────────────────────────────
+
+
+def test_passage_from_hit_maps_docparse_metadata() -> None:
+    passage = passage_from_hit(
+        _hit({"source": "q2.pdf", "page_start": "2", "page_end": "3", "heading_path": "Report > Finance"})
+    )
+    assert passage.document_id == "doc-1"
+    assert passage.page_start == 2
+    assert passage.page_end == 3
+    assert passage.heading_path == ["Report", "Finance"]
+    assert passage.source == "q2.pdf"
+
+
+def test_passage_from_hit_falls_back_to_page_number() -> None:
+    """Plain page-text chunks: page_number fills both bounds, source drops the page suffix."""
+    passage = passage_from_hit(_hit({"source": "report.pdf:page:4", "page_number": "4"}))
+    assert passage.page_start == 4
+    assert passage.page_end == 4
+    assert passage.heading_path == []
+    assert passage.source == "report.pdf"
+
+
+def test_passage_from_hit_tolerates_missing_and_bad_metadata() -> None:
+    passage = passage_from_hit(_hit({"page_start": "not-a-number"}))
+    assert passage.page_start is None
+    assert passage.page_end is None
+    assert passage.heading_path == []
+    assert passage.source is None
+
+
+# ── format_passages ─────────────────────────────────────────────────────
+
+
+def test_format_passages_includes_citation_handles() -> None:
+    passages = [
+        DocumentPassage(document_id=FileId("d1"), text="Alpha.", score=0.9, page_start=2, page_end=3, source="a.pdf"),
+        DocumentPassage(document_id=FileId("d2"), text="Beta.", score=0.5, page_start=7, page_end=7, source="b.pdf"),
+        DocumentPassage(document_id=FileId("d3"), text="Gamma.", score=0.4),
+    ]
+    rendered = format_passages(passages)
+    assert "[Passage 1 | a.pdf p.2-3]\nAlpha." in rendered
+    assert "[Passage 2 | b.pdf p.7]\nBeta." in rendered
+    # No source or pages: fall back to the document id alone.
+    assert "[Passage 3 | d3]\nGamma." in rendered
+
+
+# ── KnowledgeAskAgent ───────────────────────────────────────────────────
+
+
+class _StubEmbedder:
+    """Deterministic embeddings so the agent test needs no provider."""
+
+    async def embed_query(self, text: str) -> list[float]:
+        return [1.0, 0.0]
+
+    async def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        return [[1.0, 0.0] for _ in texts]
+
+
+@pytest.mark.anyio
+async def test_ask_answers_plainly_when_nothing_retrieved(app_settings: AppSettings) -> None:
+    """Empty retrieval short-circuits: no model call, honest not-found answer."""
+    documents = DocumentService(embedder=_StubEmbedder(), store=SqliteVecStore.ephemeral(), default_top_k=3)  # type: ignore[arg-type]
+    runtime = build_runtime(app_settings, documents=documents)
+    agent = KnowledgeAskAgent(runtime)
+
+    response = await agent.ask(AskDocumentsRequest(question="What is the launch date?"), principals=PRINCIPALS)
+    assert response.passages == []
+    assert "couldn't find" in response.answer