diff --git a/.env.example b/.env.example index df52d7c..fd9111a 100644 --- a/.env.example +++ b/.env.example @@ -6,8 +6,39 @@ DOCULENS_INITIALIZE_DATABASE=true DOCULENS_SEED_DEMO_USERS=false DOCULENS_SEED_DEMO_WORKSPACE=false DOCULENS_SHOWCASE_READ_ONLY=false +DOCULENS_REQUIRE_AUTH=false +DOCULENS_SERVE_FRONTEND=false DOCULENS_AUTH_SECRET=replace-with-a-long-random-value +DOCULENS_AUTH_COOKIE_SECURE=false DOCULENS_API_KEY= +SUPABASE_URL= +SUPABASE_PUBLISHABLE_KEY= +DOCULENS_SUPABASE_ALLOWED_EMAILS=[] +DOCULENS_SUPABASE_ALLOWED_DOMAINS=[] +DOCULENS_ALLOW_PUBLIC_SUPABASE_SIGNIN=false +DOCULENS_SUPABASE_GOOGLE_ENABLED=false +DOCULENS_TASK_MODE=celery +DOCULENS_EXTRACTION_BACKEND=docling + +# Adaptive investigation agent +DOCULENS_AGENT_ENABLED=true +DOCULENS_AGENT_PROVIDER=openai +# Optional override; otherwise the selected provider's default model is used. +DOCULENS_AGENT_MODEL= +DOCULENS_AGENT_CHECKPOINT_BACKEND=memory +DOCULENS_AGENT_MAX_STEPS=10 +DOCULENS_AGENT_MAX_EVIDENCE=24 +DOCULENS_AGENT_SEARCH_LIMIT=6 + +# Document storage (use s3 with Railway Buckets, R2, S3, or compatible storage) +DOCULENS_STORAGE_BACKEND=local +DOCULENS_STORAGE_LOCAL_PATH=/workspace/app/data/ingestion +DOCULENS_S3_ENDPOINT_URL= +DOCULENS_S3_BUCKET_NAME= +DOCULENS_S3_ACCESS_KEY_ID= +DOCULENS_S3_SECRET_ACCESS_KEY= +DOCULENS_S3_REGION=auto +DOCULENS_S3_FORCE_PATH_STYLE=false # AI providers OPENAI_API_KEY= @@ -20,6 +51,7 @@ DOCULENS_EMBEDDING_CACHE_SIZE=1024 # Infrastructure PROJECT_NAME=doculens +DATABASE_URL= DATABASE_HOST=doculens_database DATABASE_PORT=5432 DATABASE_NAME=doculens diff --git a/.github/workflows/backend-ci.yml b/.github/workflows/backend-ci.yml index 9081b67..06d6cae 100644 --- a/.github/workflows/backend-ci.yml +++ b/.github/workflows/backend-ci.yml @@ -22,7 +22,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v5 with: - python-version: "3.11" + python-version: "3.12" cache: pip - name: Install dependencies diff --git a/Makefile b/Makefile index 46b5c42..1b6e704 100644 --- a/Makefile +++ b/Makefile @@ -1,18 +1,21 @@ SHELL := /bin/bash .DEFAULT_GOAL := help -.PHONY: help install dev up down showcase-up showcase-down showcase-logs test lint format typecheck check sample +.PHONY: help install dev preview up down showcase-up showcase-down showcase-logs test lint format typecheck check sample help: ## Show available commands @awk 'BEGIN {FS = ":.*## "; printf "DocuLens developer commands\n\n"} /^[a-zA-Z_-]+:.*## / {printf " %-12s %s\n", $$1, $$2}' $(MAKEFILE_LIST) install: ## Install backend and frontend development dependencies - python3 -m pip install -e '.[dev]' + python3 -m pip install -e '.[dev,ocr]' npm --prefix frontend ci dev: ## Run API with hot reload uvicorn app.main:app --reload --port 8080 +preview: ## Start the local product preview without the heavy OCR worker + docker compose --env-file .env -f docker/docker-compose.yml up --build -d database redis api frontend + up: ## Start the complete Docker development stack docker compose --env-file .env -f docker/docker-compose.yml up --build -d diff --git a/README.md b/README.md index 9697c9a..acb0cdb 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,30 @@ # DocuLens AI -**Turn unstructured business documents into searchable, cited, operational knowledge.** +**An evidence-grounded AI agent for investigating business documents and producing auditable decisions.** -DocuLens is an open-source document intelligence system for teams that need more than a chat-with-PDF demo. It accepts documents asynchronously, extracts layout-aware content, creates citation-ready embeddings, and exposes classification, summarization, semantic search, and grounded question answering through a versioned API and an operator console. +DocuLens is an open-source document intelligence system for teams that need more +than chat-with-PDF. Give the agent an outcome—such as reviewing vendor renewals +or comparing security obligations—and it chooses its own bounded research path, +inspects source passages, repairs unsupported claims, and produces a cited +decision brief with a complete evidence ledger. [![Backend CI](https://github.com/codewithmoin/doculens-ai/actions/workflows/backend-ci.yml/badge.svg)](https://github.com/codewithmoin/doculens-ai/actions/workflows/backend-ci.yml) -[![Python 3.11+](https://img.shields.io/badge/python-3.11%2B-3776AB)](https://www.python.org/) +[![Python 3.12+](https://img.shields.io/badge/python-3.12%2B-3776AB)](https://www.python.org/) [![FastAPI](https://img.shields.io/badge/FastAPI-API-009688)](https://fastapi.tiangolo.com/) [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE) +[![Live Demo](https://img.shields.io/badge/live_demo-open-2563EB)](https://doculens-ai.pages.dev/) + +**[Explore the live showcase →](https://doculens-ai.pages.dev/)** No sign-in required. The hosted workspace is read-only and uses clearly labelled synthetic data. ## Why it is technically interesting +- **Adaptive investigation agent:** a typed LangGraph state machine lets the model + select, repeat, and stop document tools dynamically while enforcing step, + evidence, and cost boundaries. +- **Evidence ledger and citation gate:** report findings may cite only passages + retrieved during the run; invalid references route the graph back into research. +- **Durable agent state:** PostgreSQL checkpoints preserve every graph step for + fault recovery, investigation history, and future human approval workflows. - **Layout-aware ingestion:** Docling preserves headings, tables, provenance, and page numbers instead of flattening a document into one string. - **Asynchronous AI workflows:** FastAPI accepts work quickly; Celery executes extraction and model calls with bounded task time, late acknowledgement, and worker-loss recovery. - **Citation-first RAG:** every vector carries document, chunk, page, title, and token metadata; QA prompts use stable references and require grounded answers. @@ -45,22 +59,36 @@ The redesigned experience separates the public product story from a focused auth ```mermaid flowchart LR - UI["React operator console"] -->|REST /api/v1| API["FastAPI gateway"] + UI["React operator console"] -->|REST + SSE /api/v1| API["FastAPI gateway"] Client["API clients"] --> API - API -->|persist event| DB[("TimescaleDB + PostgreSQL")] + API --> Agent["LangGraph investigation agent"] + Agent --> Decide{"Choose next action"} + Decide --> Tools["Inventory · search · inspect"] + Tools --> Evidence["Evidence ledger"] + Evidence --> Decide + Decide --> Report["Cited decision brief"] + Report --> Verify{"Citation gate"} + Verify -->|repair| Decide + Verify -->|valid| UI + API -->|persist event| DB[("PostgreSQL + pgvector")] API -->|enqueue id| Redis[("Redis broker")] Redis --> Worker["Celery worker"] Worker --> Extract["Docling extraction"] Extract --> Chunk["Layout-aware chunking"] Chunk --> Embed["Batched embeddings"] - Embed --> Vector[("pgvector / DiskANN")] + Embed --> Vector[("pgvector / HNSW")] Worker --> LLM["Structured LLM calls"] LLM --> DB Vector --> Retrieve["Filtered retrieval"] Retrieve --> LLM + Agent --> Checkpoints[("PostgreSQL checkpoints")] ``` -The event record is the durable boundary between HTTP and AI work. A worker validates the stored event, selects a typed pipeline from the registry, runs its nodes, and stores a serializable task context. This design keeps slow or retryable model work away from request threads while retaining an auditable input/output trail. +The document pipeline and investigation agent intentionally use different +execution models. Document ingestion remains an asynchronous, retryable pipeline. +Investigations are adaptive graphs: the model chooses the next read-only tool, +while application-owned state, budgets, provenance, and validation determine +what it is allowed to do and when the run can finish. ## Capabilities @@ -70,8 +98,12 @@ The event record is the durable boundary between HTTP and AI work. A worker vali - document summaries with source chunk provenance - metadata-filtered semantic and keyword search - retrieval-augmented QA with stable citations and confidence +- nonlinear, cross-document investigations with model-selected tools +- streamed execution activity without exposing private chain-of-thought +- per-run evidence ledgers and deterministic citation validation +- PostgreSQL-backed LangGraph checkpoints and saved investigation history - archive, restore, and soft-delete lifecycle operations -- JWT personas plus optional API-key protection +- Supabase email-link and Google sign-in, secure session cookies, JWT personas, and optional API keys - work queues, notifications, dashboards, and request history - OpenAI, Anthropic, OpenRouter, and local OpenAI-compatible model adapters @@ -80,9 +112,10 @@ The event record is the durable boundary between HTTP and AI work. A worker vali | Layer | Technology | Responsibility | | --- | --- | --- | | API | FastAPI, Pydantic, SQLAlchemy | contracts, validation, authentication, persistence | +| Agent | LangGraph, Instructor, typed tool contracts | adaptive routing, checkpoints, evidence validation | | Jobs | Celery, Redis | resilient long-running document and AI processing | | Retrieval | Docling, tiktoken, OpenAI embeddings | extraction, bounded chunks, batched embeddings | -| Data | PostgreSQL, Timescale Vector, pgvector | events, metadata, keyword and vector search | +| Data | PostgreSQL, pgvector, HNSW | events, metadata, keyword and vector search | | Models | Instructor, OpenAI, Anthropic | provider-neutral structured outputs | | Web | React 19, TypeScript, Vite, TanStack Query | operator workflows and evidence review | | Quality | pytest, Ruff, Pyright, pre-commit, GitHub Actions | repeatable engineering checks | @@ -93,7 +126,11 @@ The event record is the durable boundary between HTTP and AI work. A worker vali - Docker with Compose v2 - an OpenAI API key (required for embeddings; model providers are configurable) -- Python 3.11+ and Node 20+ only when running services outside Docker +- Python 3.12 and Node 20+ only when running services outside Docker + +`make install` includes the optional Docling OCR stack for full local +development. The one-service Railway image intentionally installs only the +lightweight runtime; the Celery worker image installs the `ocr` extra. ```bash git clone https://github.com/codewithmoin/doculens-ai.git @@ -123,6 +160,14 @@ make showcase-up See the [showcase deployment runbook](docs/deploy-showcase.md) for DNS, TLS, verification, backups, and rollback. This mode demonstrates the finished AI workflow without accepting public uploads or spending money on visitor model calls. +### Deploy the interactive app on Railway + +The repository also includes a one-service Railway profile with Supabase Auth, +durable S3-compatible uploads, portable pgvector retrieval, and an upgrade path +to a separate Celery worker. See the +[Railway deployment runbook](docs/deploy-railway.md) for the Free portfolio +topology and the full all-Railway topology. + ## API examples All stable endpoints are under `/api/v1`. Legacy `/events` routes remain available for existing clients. @@ -148,6 +193,21 @@ curl --fail-with-body http://localhost:8080/api/v1/events/ \ The API returns `202 Accepted` with an event id. Poll the event resource until its task context contains the answer and `chunk_references` used to construct it. Ready-made payloads live in [`requests/events`](requests/events). +Start a streaming investigation: + +```bash +curl --no-buffer --fail-with-body \ + http://localhost:8080/api/v1/investigations/stream \ + -H 'Content-Type: application/json' \ + -H 'Accept: text/event-stream' \ + -H "X-API-Key: $DOCULENS_API_KEY" \ + -d '{"goal":"Review vendor agreements for renewal deadlines and contractual risk."}' +``` + +The stream contains user-safe activity events, newly discovered evidence, +citation verification, and the final decision brief. Model reasoning is never +returned as chain-of-thought. + ## Configuration Configuration is validated once at startup. See [`.env.example`](.env.example) for the full local template. @@ -163,6 +223,9 @@ Configuration is validated once at startup. See [`.env.example`](.env.example) f | `DOCULENS_PROVIDER_TIMEOUT_SECONDS` | 30 | AI provider network timeout | | `DOCULENS_QA_TOP_K` | 5 | default QA retrieval breadth | | `DOCULENS_SHOWCASE_READ_ONLY` | false | blocks workspace mutations and enables the public product-tour UX | +| `DOCULENS_AGENT_CHECKPOINT_BACKEND` | `memory` | use `postgres` for durable production investigations | +| `DOCULENS_AGENT_MAX_STEPS` | 10 | hard bound on adaptive agent iterations | +| `DOCULENS_AGENT_MAX_EVIDENCE` | 24 | maximum passages retained in one evidence ledger | ## Development @@ -178,10 +241,15 @@ Useful commands are discoverable with `make help`. CI runs the same Ruff, Pyrigh Create labelled `RetrievalExample` cases with known relevant chunk ids and run `evaluate(examples, k=5)`. Track Recall@5 and MRR before changing chunk size, embedding model, filters, or ranking. The utility is intentionally offline and deterministic: it belongs in CI; live provider quality and latency belong in a separate scheduled benchmark. +### Agent evaluation + +Create representative investigation goals with expected source documents, required report terms, citation thresholds, and step budgets. `evaluate_investigation` scores the observable graph state for document recall, report coverage, citation coverage, completion, and budget adherence—without using another model as an opaque judge. This gives agent-routing or prompt changes a deterministic regression gate while leaving semantic answer-quality judging to a separately versioned benchmark. + ## Project structure ```text app/ +├── agents/ LangGraph state, model boundary, tools, and citation gates ├── api/ HTTP contracts, auth, dependencies, versioned routers ├── config/ validated runtime and infrastructure settings ├── core/ pipeline primitives and observability @@ -204,6 +272,12 @@ requests/ executable example payloads - **Async jobs, synchronous pipeline nodes:** document work is asynchronous at the system boundary; node code stays easy to reason about because most provider SDKs and extraction libraries are synchronous. Worker concurrency supplies parallelism. - **Process-local embedding cache:** avoids repeated provider calls without introducing another consistency-sensitive cache. It resets on deploy and is not intended as durable storage. - **Provider-neutral structured outputs:** improves validation and portability, but provider behavior still differs and must be evaluated per model. +- **Bounded autonomy:** the model selects the investigation path, but tool + permissions, workspace scope, step budgets, and citation validation remain + deterministic application policy. +- **One agent, explicit tools:** a single orchestrator is easier to evaluate and + operate than a fashionable multi-agent swarm. Parallel document workers can be + introduced only when a labelled benchmark demonstrates value. - **Compatibility versioning:** `/api/v1` is canonical while legacy routes remain during migration. Removing aliases is a future breaking release. - **Product hierarchy over dashboard density:** the public landing page explains the problem and architecture; `/app` is reserved for focused document work. The UI uses route-level code splitting, a paper-and-ink system with cobalt focus and amber evidence, persistent dark mode, and reduced-motion fallbacks instead of a large animation runtime. @@ -215,11 +289,15 @@ More detail is recorded in [`docs/engineering-notes.md`](docs/engineering-notes. - The current retrieval path is dense-first; hybrid ranking exists but needs a labelled corpus before tuning. - The included deployment manifest is deliberately scoped to a single-node, read-only portfolio showcase—not a multi-tenant customer environment. - Authentication is suitable for a single workspace; multi-tenant authorization is not implemented. +- Free-tier investigations execute through an active SSE request. Checkpoints + preserve completed work, but unattended continuation requires the optional worker. - Retrieval metrics are utilities, not a bundled benchmark dataset—the project does not claim quality without domain-labelled examples. ## Roadmap -- persist event lifecycle states and retry diagnostics as first-class columns +- add resumable human approval before report export +- evaluate parallel per-document workers for large contract collections +- add agent task-completion, citation-coverage, latency, and cost benchmarks - add a small, redistributable labelled retrieval benchmark - evaluate reciprocal-rank fusion for dense + keyword retrieval - add OpenTelemetry traces and provider latency/cost dashboards diff --git a/app/agents/__init__.py b/app/agents/__init__.py new file mode 100644 index 0000000..78a98be --- /dev/null +++ b/app/agents/__init__.py @@ -0,0 +1,5 @@ +"""Evidence-grounded investigation agent.""" + +from app.agents.graph import InvestigationRuntime, build_investigation_graph + +__all__ = ["InvestigationRuntime", "build_investigation_graph"] diff --git a/app/agents/graph.py b/app/agents/graph.py new file mode 100644 index 0000000..ca3fbcc --- /dev/null +++ b/app/agents/graph.py @@ -0,0 +1,311 @@ +"""Adaptive LangGraph orchestration for document investigations.""" + +from __future__ import annotations + +from typing import Any, Iterable, Literal + +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.base import BaseCheckpointSaver +from langgraph.graph import END, START, StateGraph + +from app.agents.model_client import InstructorInvestigationModel, InvestigationModel +from app.agents.models import ( + AgentDecision, + EvidenceItem, + InvestigationActivity, + InvestigationReport, + InvestigationState, + VerificationResult, +) +from app.agents.tools import DocumentInvestigationTools, InvestigationTools + + +def _append_activity( + state: InvestigationState, + *, + kind: Literal[ + "planning", + "tool", + "evidence", + "verification", + "clarification", + "report", + "system", + ], + title: str, + detail: str | None = None, + status: Literal["running", "completed", "warning", "failed"] = "completed", + tool: str | None = None, +) -> list[dict[str, Any]]: + activity = InvestigationActivity( + kind=kind, + title=title, + detail=detail, + status=status, + tool=tool, + ) + return [*state.get("activities", []), activity.model_dump(mode="json")] + + +def _merge_by_key( + existing: list[dict[str, Any]], + incoming: Iterable[dict[str, Any]], + *, + key: str, + limit: int, +) -> list[dict[str, Any]]: + merged = {str(item.get(key)): item for item in existing} + for item in incoming: + merged[str(item.get(key))] = item + return list(merged.values())[:limit] + + +def verify_report( + report: InvestigationReport, + evidence: list[dict[str, Any]], +) -> VerificationResult: + """Reject report findings that are not connected to retrieved evidence.""" + known_refs = { + str(item.get("reference")) + for item in evidence + if item.get("reference") + } + unsupported: list[str] = [] + unknown: list[str] = [] + cited_findings = 0 + for finding in report.findings: + valid_refs = [ref for ref in finding.citation_refs if ref in known_refs] + unknown.extend(ref for ref in finding.citation_refs if ref not in known_refs) + if valid_refs: + cited_findings += 1 + else: + unsupported.append(finding.claim) + finding_count = len(report.findings) + coverage = cited_findings / finding_count if finding_count else (1.0 if evidence else 0.0) + return VerificationResult( + valid=bool(evidence) and not unsupported and not unknown, + citation_coverage=coverage, + unsupported_claims=unsupported, + unknown_citations=sorted(set(unknown)), + ) + + +def build_investigation_graph( + *, + model: InvestigationModel | None = None, + tools: InvestigationTools | None = None, + checkpointer: BaseCheckpointSaver[Any] | None = None, +): + """Compile the bounded adaptive graph with injectable reasoning and tools.""" + reasoning_model = model or InstructorInvestigationModel() + document_tools = tools or DocumentInvestigationTools() + + def decide_next(state: InvestigationState) -> dict[str, Any]: + step_count = state.get("step_count", 0) + if step_count >= state["max_steps"]: + decision = AgentDecision( + action="finish", + public_summary="The investigation budget is exhausted; preparing the best supported report.", + ) + next_step_count = step_count + else: + decision = reasoning_model.decide(state) + next_step_count = step_count + 1 + return { + "status": "running", + "step_count": next_step_count, + "decision": decision.model_dump(mode="json"), + "activities": _append_activity( + state, + kind="planning", + title=decision.public_summary, + detail=f"Step {min(step_count + 1, state['max_steps'])} of {state['max_steps']}", + ), + } + + def route_decision( + state: InvestigationState, + ) -> Literal["execute_tool", "synthesize", "request_clarification"]: + action = AgentDecision.model_validate(state["decision"]).action + if action == "finish": + return "synthesize" + if action == "request_clarification": + return "request_clarification" + return "execute_tool" + + def execute_tool(state: InvestigationState) -> dict[str, Any]: + decision = AgentDecision.model_validate(state["decision"]) + try: + observation = document_tools.execute( + decision, + state.get("document_scope", []), + ) + except Exception as exc: + return { + "error": str(exc), + "activities": _append_activity( + state, + kind="tool", + title=f"{decision.action.replace('_', ' ').title()} could not complete", + detail=str(exc), + status="warning", + tool=decision.action, + ), + } + + documents = _merge_by_key( + state.get("documents", []), + (item.model_dump(mode="json") for item in observation.documents), + key="document_id", + limit=100, + ) + evidence = _merge_by_key( + state.get("evidence", []), + (item.model_dump(mode="json") for item in observation.evidence), + key="evidence_id", + limit=state["max_evidence"], + ) + activities = _append_activity( + state, + kind="tool", + title=observation.summary, + detail=decision.public_summary, + tool=observation.tool, + ) + if observation.evidence: + activities = [ + *activities, + InvestigationActivity( + kind="evidence", + title=f"Added {len(observation.evidence)} source passages", + detail=f"{len(evidence)} unique passages are now in the evidence ledger.", + ).model_dump(mode="json"), + ] + return { + "documents": documents, + "evidence": evidence, + "activities": activities, + "error": "", + } + + def synthesize(state: InvestigationState) -> dict[str, Any]: + if state.get("evidence"): + report = reasoning_model.write_report(state) + else: + report = InvestigationReport( + title="Investigation could not establish evidence", + executive_summary=( + "No source passages were available to support a document-grounded conclusion." + ), + findings=[], + recommended_actions=["Confirm that documents have been indexed, then retry."], + limitations=["The evidence ledger is empty."], + ) + return { + "report": report.model_dump(mode="json"), + "activities": _append_activity( + state, + kind="report", + title="Drafted an evidence-grounded report", + detail=f"{len(report.findings)} findings prepared for citation validation.", + ), + } + + def validate_citations(state: InvestigationState) -> dict[str, Any]: + report = InvestigationReport.model_validate(state["report"]) + verification = verify_report(report, state.get("evidence", [])) + status: Literal["completed", "warning"] = ( + "completed" if verification.valid else "warning" + ) + return { + "verification": verification.model_dump(mode="json"), + "activities": _append_activity( + state, + kind="verification", + title=( + "All report claims are linked to evidence" + if verification.valid + else "Citation validation found unsupported claims" + ), + detail=f"Citation coverage: {verification.citation_coverage:.0%}", + status=status, + ), + } + + def route_verification(state: InvestigationState) -> Literal["finalize", "decide_next"]: + verification = VerificationResult.model_validate(state["verification"]) + if verification.valid or state["step_count"] >= state["max_steps"]: + return "finalize" + return "decide_next" + + def finalize(state: InvestigationState) -> dict[str, Any]: + verification = VerificationResult.model_validate(state["verification"]) + return { + "status": "completed" if verification.valid else "partial", + "activities": _append_activity( + state, + kind="system", + title=( + "Investigation completed" + if verification.valid + else "Investigation completed with limitations" + ), + detail=f"{len(state.get('evidence', []))} evidence passages reviewed.", + status="completed" if verification.valid else "warning", + ), + } + + def request_clarification(state: InvestigationState) -> dict[str, Any]: + decision = AgentDecision.model_validate(state["decision"]) + question = decision.clarification_question or "What should this investigation focus on?" + return { + "status": "needs_input", + "clarification_question": question, + "activities": _append_activity( + state, + kind="clarification", + title="The agent needs clarification", + detail=question, + status="warning", + ), + } + + builder = StateGraph(InvestigationState) + builder.add_node("decide_next", decide_next) + builder.add_node("execute_tool", execute_tool) + builder.add_node("synthesize", synthesize) + builder.add_node("validate_citations", validate_citations) + builder.add_node("finalize", finalize) + builder.add_node("request_clarification", request_clarification) + builder.add_edge(START, "decide_next") + builder.add_conditional_edges("decide_next", route_decision) + builder.add_edge("execute_tool", "decide_next") + builder.add_edge("synthesize", "validate_citations") + builder.add_conditional_edges("validate_citations", route_verification) + builder.add_edge("finalize", END) + builder.add_edge("request_clarification", END) + return builder.compile(checkpointer=checkpointer) + + +class InvestigationRuntime: + """Process-level compiled graph used by FastAPI request handlers.""" + + def __init__( + self, + checkpointer: BaseCheckpointSaver[Any] | None = None, + *, + model: InvestigationModel | None = None, + tools: InvestigationTools | None = None, + ) -> None: + self.graph = build_investigation_graph( + model=model, + tools=tools, + checkpointer=checkpointer, + ) + + def stream(self, state: InvestigationState): + config: RunnableConfig = { + "configurable": {"thread_id": state["investigation_id"]}, + "recursion_limit": state["max_steps"] * 4 + 8, + } + yield from self.graph.stream(state, config=config, stream_mode="values") diff --git a/app/agents/model_client.py b/app/agents/model_client.py new file mode 100644 index 0000000..d94e245 --- /dev/null +++ b/app/agents/model_client.py @@ -0,0 +1,102 @@ +"""Model boundary for adaptive decisions and cited report synthesis.""" + +from __future__ import annotations + +import json +from typing import Protocol + +from app.agents.models import AgentDecision, InvestigationReport, InvestigationState +from app.config.settings import get_settings +from app.services.llm_factory import LLMFactory + +DECISION_SYSTEM_PROMPT = """\ +You are the bounded investigation controller for DocuLens. +Choose exactly one next action from: +- list_documents: discover the workspace inventory. +- search_evidence: semantic search for passages relevant to the goal or an evidence gap. +- inspect_document: read ordered chunks from one known document. +- finish: stop gathering and produce the report. +- request_clarification: use only when the goal cannot be investigated safely without user input. + +You may revisit tools with improved queries. Prefer evidence from multiple relevant documents +for comparative goals. Never invent document ids. Stop when the available evidence can support +a useful answer. public_summary is shown to the user and must describe the action without exposing +private reasoning. Do not claim that a tool has run before it runs. +""" + +REPORT_SYSTEM_PROMPT = """\ +You are the report writer for an evidence-grounded document investigation. +Use only the supplied evidence. Every finding must cite one or more exact evidence references. +Do not cite a document generally when the evidence contains a page or chunk reference. +If the evidence is incomplete or contradictory, state that in limitations. Prefer a short, +decision-ready report over generic prose. +""" + + +class InvestigationModel(Protocol): + """Replaceable reasoning boundary used by graph tests and production providers.""" + + def decide(self, state: InvestigationState) -> AgentDecision: + ... + + def write_report(self, state: InvestigationState) -> InvestigationReport: + ... + + +class InstructorInvestigationModel: + """Use the repository's provider abstraction for structured model outputs.""" + + def __init__(self) -> None: + settings = get_settings() + self.settings = settings + self.client = LLMFactory(settings.agent_provider) + + def decide(self, state: InvestigationState) -> AgentDecision: + evidence = state.get("evidence", []) + compact_evidence = [ + { + "reference": item.get("reference"), + "document_id": item.get("document_id"), + "text": str(item.get("text", ""))[:500], + } + for item in evidence[-12:] + ] + payload = { + "goal": state["goal"], + "document_scope": state.get("document_scope", []), + "known_documents": state.get("documents", [])[-30:], + "evidence": compact_evidence, + "previous_verification": state.get("verification"), + "last_tool_error": state.get("error"), + "step": state.get("step_count", 0), + "max_steps": state.get("max_steps"), + } + result, _ = self.client.create_completion( + response_model=AgentDecision, + messages=[ + {"role": "system", "content": DECISION_SYSTEM_PROMPT}, + {"role": "user", "content": json.dumps(payload, default=str)}, + ], + model=self.settings.agent_model or self.client.settings.default_model, + temperature=0.1, + max_tokens=500, + ) + return AgentDecision.model_validate(result) + + def write_report(self, state: InvestigationState) -> InvestigationReport: + payload = { + "goal": state["goal"], + "evidence": state.get("evidence", []), + "known_documents": state.get("documents", []), + } + result, _ = self.client.create_completion( + response_model=InvestigationReport, + messages=[ + {"role": "system", "content": REPORT_SYSTEM_PROMPT}, + {"role": "user", "content": json.dumps(payload, default=str)}, + ], + model=self.settings.agent_model or self.client.settings.default_model, + temperature=0.1, + max_tokens=1400, + ) + return InvestigationReport.model_validate(result) diff --git a/app/agents/models.py b/app/agents/models.py new file mode 100644 index 0000000..1a99a15 --- /dev/null +++ b/app/agents/models.py @@ -0,0 +1,155 @@ +"""Typed contracts shared by the investigation graph, tools, API, and tests.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Literal, Optional, TypedDict +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +ToolName = Literal[ + "list_documents", + "search_evidence", + "inspect_document", + "finish", + "request_clarification", +] +InvestigationStatus = Literal[ + "queued", + "running", + "needs_input", + "completed", + "partial", + "failed", +] + + +def utc_now_iso() -> str: + """Return a checkpoint-safe UTC timestamp.""" + return datetime.now(timezone.utc).isoformat() + + +class StrictModel(BaseModel): + """Forbid model-produced fields that the application does not understand.""" + + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + +class AgentDecision(StrictModel): + """One adaptive action selected by the model from the bounded tool set.""" + + action: ToolName + public_summary: str = Field( + min_length=3, + max_length=180, + description="A concise user-visible description of the next action; never hidden reasoning.", + ) + query: Optional[str] = Field(default=None, max_length=800) + document_id: Optional[str] = Field(default=None, max_length=255) + filters: dict[str, Any] = Field(default_factory=dict) + requested_limit: Optional[int] = Field(default=None, ge=1, le=20) + clarification_question: Optional[str] = Field(default=None, max_length=500) + + @model_validator(mode="after") + def validate_action_arguments(self) -> "AgentDecision": + if self.action == "search_evidence" and not self.query: + raise ValueError("search_evidence requires a query") + if self.action == "inspect_document" and not self.document_id: + raise ValueError("inspect_document requires a document_id") + if self.action == "request_clarification" and not self.clarification_question: + raise ValueError("request_clarification requires clarification_question") + return self + + +class DocumentCandidate(StrictModel): + document_id: str + filename: str + doc_type: Optional[str] = None + status: Optional[str] = None + summary: Optional[str] = None + + +class EvidenceItem(StrictModel): + """A passage with stable provenance that can support report claims.""" + + evidence_id: str + reference: str + document_id: str + filename: str + text: str + chunk_id: Optional[str] = None + chunk_index: Optional[int] = None + page_number: Optional[int] = None + relevance_score: Optional[float] = None + source_query: Optional[str] = None + + +class ToolObservation(StrictModel): + tool: Literal["list_documents", "search_evidence", "inspect_document"] + summary: str + documents: list[DocumentCandidate] = Field(default_factory=list) + evidence: list[EvidenceItem] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class InvestigationActivity(StrictModel): + """An observable execution event safe to render in the product UI.""" + + activity_id: str = Field(default_factory=lambda: str(uuid4())) + kind: Literal[ + "planning", + "tool", + "evidence", + "verification", + "clarification", + "report", + "system", + ] + title: str + detail: Optional[str] = None + status: Literal["running", "completed", "warning", "failed"] = "completed" + tool: Optional[str] = None + created_at: str = Field(default_factory=utc_now_iso) + + +class ReportFinding(StrictModel): + claim: str + significance: Literal["low", "medium", "high", "critical"] = "medium" + explanation: str + citation_refs: list[str] = Field(min_length=1) + + +class InvestigationReport(StrictModel): + title: str + executive_summary: str + findings: list[ReportFinding] = Field(default_factory=list) + recommended_actions: list[str] = Field(default_factory=list) + limitations: list[str] = Field(default_factory=list) + + +class VerificationResult(StrictModel): + valid: bool + citation_coverage: float = Field(ge=0.0, le=1.0) + unsupported_claims: list[str] = Field(default_factory=list) + unknown_citations: list[str] = Field(default_factory=list) + + +class InvestigationState(TypedDict): + """Serializable LangGraph state; no SDK or database objects belong here.""" + + investigation_id: str + goal: str + status: InvestigationStatus + step_count: int + max_steps: int + max_evidence: int + document_scope: list[str] + documents: list[dict[str, Any]] + evidence: list[dict[str, Any]] + activities: list[dict[str, Any]] + decision: dict[str, Any] + report: dict[str, Any] + verification: dict[str, Any] + clarification_question: str + error: str diff --git a/app/agents/tools.py b/app/agents/tools.py new file mode 100644 index 0000000..84273e9 --- /dev/null +++ b/app/agents/tools.py @@ -0,0 +1,196 @@ +"""Read-only document tools available to the investigation controller.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Protocol + +from sqlalchemy import text + +from app.agents.models import ( + AgentDecision, + DocumentCandidate, + EvidenceItem, + ToolObservation, +) +from app.config.settings import get_settings +from app.database.session import SessionLocal +from app.doc_utils.search import semantic_search_docling +from app.services.vector_store import VectorStore + + +class InvestigationTools(Protocol): + def execute(self, decision: AgentDecision, document_scope: list[str]) -> ToolObservation: + ... + + +def _as_int(value: Any) -> int | None: + if isinstance(value, int): + return value + if isinstance(value, str) and value.isdigit(): + return int(value) + return None + + +def _evidence_from_record(record: dict[str, Any], query: str | None) -> EvidenceItem: + metadata = record.get("metadata") or {} + document_id = str(metadata.get("document_id") or "unknown-document") + filename = str( + metadata.get("original_filename") + or metadata.get("filename") + or document_id + ) + chunk_index = _as_int(metadata.get("chunk_index")) + pages = metadata.get("page_numbers") + page_number = ( + _as_int(pages[0]) + if isinstance(pages, list) and pages + else _as_int(metadata.get("page_number")) + ) + reference_parts = [filename] + if page_number is not None: + reference_parts.append(f"page {page_number}") + if chunk_index is not None: + reference_parts.append(f"chunk {chunk_index}") + reference = str(metadata.get("reference") or " · ".join(reference_parts)) + distance = record.get("distance") + relevance = None + if isinstance(distance, (float, int)): + relevance = max(0.0, min(1.0, 1.0 - float(distance))) + return EvidenceItem( + evidence_id=str(record.get("id") or reference), + reference=reference, + document_id=document_id, + filename=filename, + chunk_id=str(record.get("id")) if record.get("id") else None, + chunk_index=chunk_index, + page_number=page_number, + text=str(record.get("contents") or ""), + relevance_score=relevance, + source_query=query, + ) + + +class DocumentInvestigationTools: + """Concrete, workspace-scoped tools backed by PostgreSQL and pgvector.""" + + def execute(self, decision: AgentDecision, document_scope: list[str]) -> ToolObservation: + if decision.action == "list_documents": + return self.list_documents(document_scope) + if decision.action == "search_evidence": + return self.search_evidence(decision, document_scope) + if decision.action == "inspect_document": + return self.inspect_document(decision, document_scope) + raise ValueError(f"{decision.action} is not an executable document tool") + + def list_documents(self, document_scope: list[str]) -> ToolObservation: + with SessionLocal() as session: + rows = session.execute( + text( + """ + SELECT id, data, task_context + FROM events + WHERE data->>'event_type' = 'document_upload' + ORDER BY created_at DESC + LIMIT 100 + """ + ) + ).mappings() + documents: list[DocumentCandidate] = [] + for row in rows: + task_context = row.get("task_context") or {} + metadata = task_context.get("metadata") or {} + document = metadata.get("document") or {} + upload = row.get("data") or {} + upload_metadata = upload.get("metadata") or {} + document_id = str(document.get("id") or row.get("id")) + if document_scope and document_id not in document_scope: + continue + raw_filename = ( + upload_metadata.get("uploaded_filename") + or document.get("original_filename") + or document.get("filename") + or upload.get("filename") + or document_id + ) + filename = Path(str(raw_filename)).name + documents.append( + DocumentCandidate( + document_id=document_id, + filename=filename, + doc_type=document.get("doc_type") or upload.get("doc_type"), + status=document.get("status") or upload_metadata.get("status"), + ) + ) + return ToolObservation( + tool="list_documents", + summary=f"Found {len(documents)} documents in scope.", + documents=documents, + metadata={"document_count": len(documents)}, + ) + + def search_evidence( + self, + decision: AgentDecision, + document_scope: list[str], + ) -> ToolObservation: + settings = get_settings() + limit = decision.requested_limit or settings.agent_search_limit + filters = dict(decision.filters) + metadata_filter: dict[str, Any] | list[dict[str, Any]] | None + if decision.document_id: + filters["document_id"] = decision.document_id + metadata_filter = filters + elif len(document_scope) == 1: + filters["document_id"] = document_scope[0] + metadata_filter = filters + elif document_scope: + metadata_filter = [ + {**filters, "document_id": document_id} + for document_id in document_scope + ] + else: + metadata_filter = filters or None + + records = semantic_search_docling( + query=decision.query or "", + limit=limit, + metadata_filter=metadata_filter, + ) + if document_scope: + records = [ + record + for record in records + if str((record.get("metadata") or {}).get("document_id")) in document_scope + ] + evidence = [_evidence_from_record(record, decision.query) for record in records] + return ToolObservation( + tool="search_evidence", + summary=f"Retrieved {len(evidence)} passages for “{decision.query}”.", + evidence=evidence, + metadata={ + "query": decision.query, + "filters": metadata_filter, + "result_count": len(evidence), + }, + ) + + def inspect_document( + self, + decision: AgentDecision, + document_scope: list[str], + ) -> ToolObservation: + document_id = decision.document_id or "" + if document_scope and document_id not in document_scope: + raise PermissionError("The requested document is outside the investigation scope.") + records = VectorStore().fetch_document_chunks( + document_id=document_id, + limit=decision.requested_limit or get_settings().agent_search_limit, + ) + evidence = [_evidence_from_record(record, decision.query) for record in records] + return ToolObservation( + tool="inspect_document", + summary=f"Inspected {len(evidence)} ordered passages from {document_id}.", + evidence=evidence, + metadata={"document_id": document_id, "result_count": len(evidence)}, + ) diff --git a/app/alembic/env.py b/app/alembic/env.py index 85aa0aa..fea4230 100644 --- a/app/alembic/env.py +++ b/app/alembic/env.py @@ -10,6 +10,7 @@ # This import is required for autogenerate support from app.database.event import * # noqa: F401,F403 +from app.database.investigation import * # noqa: F401,F403 from app.database.user import * # noqa: F401,F403 """ diff --git a/app/alembic/versions/20260730_0005_add_federated_identity.py b/app/alembic/versions/20260730_0005_add_federated_identity.py new file mode 100644 index 0000000..571558c --- /dev/null +++ b/app/alembic/versions/20260730_0005_add_federated_identity.py @@ -0,0 +1,38 @@ +"""add federated identity fields + +Revision ID: 20260730_0005 +Revises: 20250215_0004 +Create Date: 2026-07-30 00:05:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260730_0005" +down_revision: Union[str, None] = "20250215_0004" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "users", + sa.Column("auth_provider", sa.String(length=32), nullable=False, server_default="password"), + ) + op.add_column("users", sa.Column("provider_subject", sa.String(length=255), nullable=True)) + op.add_column("users", sa.Column("avatar_url", sa.String(length=1024), nullable=True)) + op.create_unique_constraint( + "uq_users_auth_identity", + "users", + ["auth_provider", "provider_subject"], + ) + + +def downgrade() -> None: + op.drop_constraint("uq_users_auth_identity", "users", type_="unique") + op.drop_column("users", "avatar_url") + op.drop_column("users", "provider_subject") + op.drop_column("users", "auth_provider") diff --git a/app/alembic/versions/20260730_0006_create_investigations.py b/app/alembic/versions/20260730_0006_create_investigations.py new file mode 100644 index 0000000..ceb7419 --- /dev/null +++ b/app/alembic/versions/20260730_0006_create_investigations.py @@ -0,0 +1,40 @@ +"""create investigations table + +Revision ID: 20260730_0006 +Revises: 20260730_0005 +Create Date: 2026-07-30 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = "20260730_0006" +down_revision = "20260730_0005" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "investigations", + sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("user_id", postgresql.UUID(as_uuid=True), nullable=True), + sa.Column("goal", sa.Text(), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("state", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("created_at", sa.DateTime(), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_investigations_created_at", "investigations", ["created_at"]) + op.create_index("ix_investigations_status", "investigations", ["status"]) + op.create_index("ix_investigations_user_id", "investigations", ["user_id"]) + + +def downgrade() -> None: + op.drop_index("ix_investigations_user_id", table_name="investigations") + op.drop_index("ix_investigations_status", table_name="investigations") + op.drop_index("ix_investigations_created_at", table_name="investigations") + op.drop_table("investigations") diff --git a/app/api/auth_router.py b/app/api/auth_router.py index 0faaead..44d52d1 100644 --- a/app/api/auth_router.py +++ b/app/api/auth_router.py @@ -1,4 +1,4 @@ -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Response, status from pydantic import BaseModel, EmailStr from sqlalchemy.orm import Session @@ -11,6 +11,7 @@ create_access_token, get_current_user, ) +from app.config.settings import get_settings class LoginRequest(BaseModel): @@ -25,6 +26,8 @@ class UserProfile(BaseModel): persona: str role: str access_level: str + avatar_url: str | None = None + auth_provider: str @classmethod def from_orm(cls, user: User) -> "UserProfile": @@ -35,6 +38,8 @@ def from_orm(cls, user: User) -> "UserProfile": persona=user.persona, role=user.role, access_level=user.access_level, + avatar_url=user.avatar_url, + auth_provider=user.auth_provider, ) @@ -46,11 +51,39 @@ class TokenResponse(BaseModel): roles: dict[str, dict[str, str]] +class SupabaseLoginRequest(BaseModel): + access_token: str + + router = APIRouter(prefix="/auth", tags=["auth"]) +def _build_auth_response(*, user: User, response: Response) -> TokenResponse: + settings = get_settings() + token = create_access_token(user=user) + response.set_cookie( + key=settings.auth_cookie_name, + value=token, + max_age=settings.auth_token_exp_minutes * 60, + httponly=True, + secure=settings.auth_cookie_secure, + samesite="lax", + path="/", + ) + return TokenResponse( + access_token=token, + user=UserProfile.from_orm(user), + personas=PERSONA_OPTIONS, + roles=ROLE_DEFINITIONS, + ) + + @router.post("/login", response_model=TokenResponse) -def login(request: LoginRequest, session: Session = Depends(db_session)) -> TokenResponse: +def login( + request: LoginRequest, + response: Response, + session: Session = Depends(db_session), +) -> TokenResponse: service = AuthService(session) user = service.authenticate_user(email=request.email, password=request.password) if not user: @@ -58,13 +91,30 @@ def login(request: LoginRequest, session: Session = Depends(db_session)) -> Toke status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials.", ) - token = create_access_token(user=user) - return TokenResponse( - access_token=token, - user=UserProfile.from_orm(user), - personas=PERSONA_OPTIONS, - roles=ROLE_DEFINITIONS, + return _build_auth_response(user=user, response=response) + + +@router.post("/supabase", response_model=TokenResponse) +def supabase_login( + request: SupabaseLoginRequest, + response: Response, + session: Session = Depends(db_session), +) -> TokenResponse: + user = AuthService(session).authenticate_supabase(access_token=request.access_token) + return _build_auth_response(user=user, response=response) + + +@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) +def logout(response: Response) -> Response: + settings = get_settings() + response.delete_cookie( + key=settings.auth_cookie_name, + httponly=True, + secure=settings.auth_cookie_secure, + samesite="lax", + path="/", ) + return response @router.get("/me", response_model=UserProfile) diff --git a/app/api/endpoint.py b/app/api/endpoint.py index 60f7d94..56aa250 100644 --- a/app/api/endpoint.py +++ b/app/api/endpoint.py @@ -7,7 +7,18 @@ from typing import Any, Dict, Iterable, List, Literal, Optional, Sequence, Tuple from uuid import UUID, uuid4 -from fastapi import APIRouter, Body, Depends, File, Form, HTTPException, Query, Response, UploadFile +from fastapi import ( + APIRouter, + BackgroundTasks, + Body, + Depends, + File, + Form, + HTTPException, + Query, + Response, + UploadFile, +) from sqlalchemy import select, text from sqlalchemy.orm import Session from pydantic import BaseModel, Field @@ -30,6 +41,7 @@ delete_document as delete_document_service, restore_document as restore_document_service, ) +from app.services.document_storage import get_document_storage from app.services.label_service import LabelConflictError, LabelService from app.database.models import DocumentClassificationHistory, DocumentLabel from app.services.auth_service import PERSONA_OPTIONS, ROLE_DEFINITIONS @@ -71,15 +83,22 @@ def _strip_ingestion_prefix(filename: str) -> str: return candidate -def _store_event_and_dispatch(session: Session, payload: Dict[str, Any]) -> Tuple[Event, str]: +def _store_event_and_dispatch( + session: Session, + payload: Dict[str, Any], + background_tasks: BackgroundTasks, +) -> Tuple[Event, str]: """Persist an event and enqueue the Celery worker.""" repository = GenericRepository(session=session, model=Event) event = Event(data=payload) repository.create(obj=event) - task_result = celery_app.send_task( - "process_incoming_event", - args=[str(event.id)], - ) + if get_settings().task_mode == "local": + from app.tasks.tasks import process_incoming_event + + background_tasks.add_task(process_incoming_event.run, str(event.id)) + return event, f"local-{event.id}" + + task_result = celery_app.send_task("process_incoming_event", args=[str(event.id)]) task_id = getattr(task_result, "id", str(task_result)) return event, task_id @@ -292,11 +311,18 @@ def get_runtime_config() -> Dict[str, Any]: "search_result_limit": settings.search_result_limit, "search_preview_limit": settings.search_preview_limit, "chunk_preview_limit": settings.chunk_preview_limit, - "auth_required": bool(settings.api_key), + "auth_required": bool(settings.api_key or settings.require_auth), + "session_auth_required": settings.require_auth, + "supabase_url": settings.supabase_url, + "supabase_publishable_key": settings.supabase_publishable_key, + "supabase_google_enabled": settings.supabase_google_enabled, "showcase_read_only": settings.showcase_read_only, "api_key_header": settings.api_key_header, "persona_options": PERSONA_OPTIONS, "role_definitions": ROLE_DEFINITIONS, + "agent_enabled": settings.agent_enabled, + "agent_max_steps": settings.agent_max_steps, + "agent_max_evidence": settings.agent_max_evidence, } @@ -1042,6 +1068,7 @@ def override_classification( @router.post("/documents/upload", status_code=HTTPStatus.ACCEPTED) async def upload_document( + background_tasks: BackgroundTasks, file: UploadFile = File(...), doc_type: Optional[str] = Form(default=None), metadata: Optional[str] = Form(default=None), @@ -1051,70 +1078,50 @@ async def upload_document( if not file.filename: raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Uploaded file must have a filename.") - ingestion_dir = _ensure_ingestion_dir() original_name = Path(file.filename).name stored_filename = f"{uuid4().hex}_{original_name}" - stored_path = ingestion_dir / stored_filename - - bytes_written = 0 - max_upload_bytes = get_settings().max_upload_bytes - try: - with stored_path.open("wb") as buffer: - while True: - chunk = await file.read(1 << 20) - if not chunk: - break - bytes_written += len(chunk) - if bytes_written > max_upload_bytes: - raise HTTPException( - status_code=HTTPStatus.REQUEST_ENTITY_TOO_LARGE, - detail=f"Upload exceeds the {max_upload_bytes}-byte limit.", - ) - buffer.write(chunk) - except Exception: - stored_path.unlink(missing_ok=True) - raise - finally: - await file.close() metadata_dict: Dict[str, Any] = {} if metadata: try: decoded = json.loads(metadata) except json.JSONDecodeError as exc: - stored_path.unlink(missing_ok=True) raise HTTPException( status_code=HTTPStatus.BAD_REQUEST, detail="metadata must be valid JSON.", ) from exc if not isinstance(decoded, dict): - stored_path.unlink(missing_ok=True) raise HTTPException( status_code=HTTPStatus.BAD_REQUEST, detail="metadata must be a JSON object.", ) metadata_dict = decoded + stored_document = await get_document_storage().store_upload( + file, + object_name=stored_filename, + max_bytes=get_settings().max_upload_bytes, + ) metadata_dict.setdefault("uploaded_filename", original_name) - metadata_dict.setdefault("ingest_path", str(stored_path)) + metadata_dict.setdefault("storage_reference", stored_document.reference) event_payload: Dict[str, Any] = { "event_type": "document_upload", - "filename": str(stored_path), + "filename": stored_document.reference, "file_url": "", "doc_type": doc_type or None, "metadata": metadata_dict, } - event, task_id = _store_event_and_dispatch(session, event_payload) + event, task_id = _store_event_and_dispatch(session, event_payload, background_tasks) return { "message": "Document upload accepted", "event_id": str(event.id), "task_id": task_id, "original_filename": original_name, - "stored_path": str(stored_path), - "size_bytes": bytes_written, + "stored_path": stored_document.reference, + "size_bytes": stored_document.bytes_written, } @@ -1236,6 +1243,7 @@ def get_event( @router.post("/", dependencies=[], status_code=HTTPStatus.ACCEPTED) def handle_event( data: EventSchema, + background_tasks: BackgroundTasks, session: Session = Depends(db_session), ) -> Dict[str, Any]: """Handles incoming event submissions. @@ -1256,7 +1264,7 @@ def handle_event( Use the task ID in the response to check processing status. """ event_payload = data.model_dump(mode="json") - event, task_id = _store_event_and_dispatch(session, event_payload) + event, task_id = _store_event_and_dispatch(session, event_payload, background_tasks) return { "message": "process_incoming_event started", diff --git a/app/api/investigation_router.py b/app/api/investigation_router.py new file mode 100644 index 0000000..173d0cb --- /dev/null +++ b/app/api/investigation_router.py @@ -0,0 +1,200 @@ +"""Streaming API for bounded evidence-grounded investigations.""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Iterator +from typing import Any, Optional, cast +from uuid import UUID, uuid4 + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, ConfigDict, Field +from sqlalchemy.orm import Session + +from app.agents.graph import InvestigationRuntime +from app.agents.models import InvestigationState +from app.api.dependencies import db_session +from app.config.settings import get_settings +from app.database.session import SessionLocal +from app.database.user import User +from app.services.auth_service import require_current_user_if_enabled +from app.services.investigation_service import ( + create_investigation, + get_investigation, + list_investigations, + persist_investigation_state, + serialize_investigation, +) + +logger = logging.getLogger(__name__) +router = APIRouter() + + +class InvestigationRequest(BaseModel): + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + goal: str = Field(min_length=10, max_length=2000) + document_ids: list[str] = Field(default_factory=list, max_length=100) + + +def _user_id(user: Optional[User]) -> UUID | None: + return cast(UUID, user.id) if user is not None else None + + +def _sse(event: str, payload: Any) -> str: + return f"event: {event}\ndata: {json.dumps(payload, default=str)}\n\n" + + +def _save_snapshot(investigation_id: UUID, state: InvestigationState) -> None: + with SessionLocal() as session: + persist_investigation_state(session, investigation_id, state) + session.commit() + + +@router.post("/stream", summary="Start and stream an adaptive document investigation") +def stream_investigation( + payload: InvestigationRequest, + request: Request, + session: Session = Depends(db_session), + user: Optional[User] = Depends(require_current_user_if_enabled), +) -> StreamingResponse: + settings = get_settings() + if not settings.agent_enabled: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The investigation agent is disabled.", + ) + runtime = getattr(request.app.state, "investigation_runtime", None) + if not isinstance(runtime, InvestigationRuntime): + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="The investigation runtime is unavailable.", + ) + + investigation_id = uuid4() + initial_state: InvestigationState = { + "investigation_id": str(investigation_id), + "goal": payload.goal, + "status": "queued", + "step_count": 0, + "max_steps": settings.agent_max_steps, + "max_evidence": settings.agent_max_evidence, + "document_scope": list(dict.fromkeys(payload.document_ids)), + "documents": [], + "evidence": [], + "activities": [], + "decision": {}, + "report": {}, + "verification": {}, + "clarification_question": "", + "error": "", + } + create_investigation( + session, + investigation_id=investigation_id, + goal=payload.goal, + user_id=_user_id(user), + initial_state=initial_state, + ) + # The stream runs after the response starts, so make the record visible first. + session.commit() + + def generate() -> Iterator[str]: + seen_activities: set[str] = set() + seen_evidence: set[str] = set() + final_state = initial_state + yield _sse( + "investigation", + { + "id": str(investigation_id), + "goal": payload.goal, + "status": "running", + "max_steps": settings.agent_max_steps, + }, + ) + try: + running_state = InvestigationState(**{**initial_state, "status": "running"}) + for snapshot in runtime.stream(running_state): + final_state = InvestigationState(**snapshot) + _save_snapshot(investigation_id, final_state) + for activity in final_state.get("activities", []): + activity_id = str(activity.get("activity_id")) + if activity_id not in seen_activities: + seen_activities.add(activity_id) + yield _sse("activity", activity) + for evidence in final_state.get("evidence", []): + evidence_id = str(evidence.get("evidence_id")) + if evidence_id not in seen_evidence: + seen_evidence.add(evidence_id) + yield _sse("evidence", evidence) + + if final_state.get("report"): + yield _sse( + "report", + { + "report": final_state["report"], + "verification": final_state.get("verification", {}), + }, + ) + yield _sse( + "done", + { + "id": str(investigation_id), + "status": final_state.get("status", "completed"), + "step_count": final_state.get("step_count", 0), + "evidence_count": len(final_state.get("evidence", [])), + "clarification_question": final_state.get("clarification_question"), + }, + ) + except Exception as exc: + logger.exception("Investigation %s failed", investigation_id) + failed_state: InvestigationState = { + **final_state, + "status": "failed", + "error": str(exc), + } + _save_snapshot(investigation_id, failed_state) + yield _sse( + "error", + { + "id": str(investigation_id), + "message": "The investigation stopped before completion.", + }, + ) + + return StreamingResponse( + generate(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "Connection": "keep-alive", + }, + ) + + +@router.get("", summary="List recent investigations") +def get_investigations( + limit: int = Query(default=20, ge=1, le=100), + session: Session = Depends(db_session), + user: Optional[User] = Depends(require_current_user_if_enabled), +) -> list[dict[str, Any]]: + records = list_investigations(session, user_id=_user_id(user), limit=limit) + return [serialize_investigation(record, include_state=False) for record in records] + + +@router.get("/{investigation_id}", summary="Get an investigation and its evidence ledger") +def get_investigation_detail( + investigation_id: UUID, + session: Session = Depends(db_session), + user: Optional[User] = Depends(require_current_user_if_enabled), +) -> dict[str, Any]: + record = get_investigation(session, investigation_id, user_id=_user_id(user)) + if record is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Investigation not found.", + ) + return serialize_investigation(record) diff --git a/app/api/router.py b/app/api/router.py index ac0cc16..074ec42 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -3,6 +3,8 @@ from app.api.dependencies import require_api_key, require_showcase_writable from app.api.endpoint import router as endpoint_router, public_router as public_endpoint_router from app.api.auth_router import router as auth_router +from app.api.investigation_router import router as investigation_router +from app.services.auth_service import require_current_user_if_enabled """ API Router Module @@ -14,9 +16,18 @@ router = APIRouter() secured_router = APIRouter( - dependencies=[Depends(require_api_key), Depends(require_showcase_writable)] + dependencies=[ + Depends(require_api_key), + Depends(require_showcase_writable), + Depends(require_current_user_if_enabled), + ] ) secured_router.include_router(endpoint_router, prefix="/events", tags=["events"]) +secured_router.include_router( + investigation_router, + prefix="/investigations", + tags=["investigations"], +) router.include_router(auth_router) router.include_router(public_endpoint_router, prefix="/events", tags=["events"]) diff --git a/app/config/celery_config.py b/app/config/celery_config.py index 40e3330..88be556 100644 --- a/app/config/celery_config.py +++ b/app/config/celery_config.py @@ -59,5 +59,6 @@ def get_celery_config(): celery_app = Celery("tasks") celery_app.config_from_object(get_celery_config()) -# Automatically discover and register tasks -celery_app.autodiscover_tasks(["app.tasks"], force=True) +# Workers import task registrations during boot. The web process avoids eagerly +# loading the document ML stack until local task mode actually processes a file. +celery_app.conf.imports = ("app.tasks.tasks",) diff --git a/app/config/database_config.py b/app/config/database_config.py index 1bae633..ebf600a 100644 --- a/app/config/database_config.py +++ b/app/config/database_config.py @@ -1,11 +1,7 @@ import os -from datetime import timedelta -from dotenv import load_dotenv from pydantic_settings import BaseSettings -load_dotenv() - """ Configuration for the VectorStore. """ @@ -16,12 +12,12 @@ class VectorStoreConfig(BaseSettings): table_name: str = "embeddings" embedding_dimensions: int = 1536 - time_partition_interval: timedelta = timedelta(days=7) class DatabaseConfig(BaseSettings): """Settings for the database.""" + url: str | None = os.getenv("DATABASE_URL") host: str = os.getenv("DATABASE_HOST", "doculens_database") port: str = os.getenv("DATABASE_PORT", "5432") name: str = os.getenv("DATABASE_NAME", "doculens") @@ -32,13 +28,24 @@ class DatabaseConfig(BaseSettings): @property def service_url(self) -> str: """Generate the service URL based on the environment.""" + if self.url: + return self._normalise_url(self.url) if self.local: return f"postgres://{self.pg_user}:{self.password}@localhost:{self.port}/{self.name}" return f"postgres://{self.pg_user}:{self.password}@{self.host}:{self.port}/{self.name}" def service_url_for(self, *, local: bool = False) -> str: """Build a URL without mutating the cached settings singleton.""" + if self.url and not local: + return self._normalise_url(self.url) host = "localhost" if local else self.host return f"postgres://{self.pg_user}:{self.password}@{host}:{self.port}/{self.name}" + @staticmethod + def _normalise_url(url: str) -> str: + """Use SQLAlchemy's explicit PostgreSQL scheme for provider URLs.""" + if url.startswith("postgres://"): + return "postgresql://" + url.removeprefix("postgres://") + return url + vector_store: VectorStoreConfig = VectorStoreConfig() diff --git a/app/config/settings.py b/app/config/settings.py index f51fcc6..c76a885 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -8,7 +8,7 @@ from functools import lru_cache from typing import Literal, Optional -from pydantic import Field, field_validator +from pydantic import AliasChoices, Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict from app.config.database_config import DatabaseConfig @@ -32,6 +32,11 @@ class Settings(BaseSettings): seed_demo_users: bool = Field(default=False, alias="DOCULENS_SEED_DEMO_USERS") seed_demo_workspace: bool = Field(default=False, alias="DOCULENS_SEED_DEMO_WORKSPACE") showcase_read_only: bool = Field(default=False, alias="DOCULENS_SHOWCASE_READ_ONLY") + require_auth: bool = Field(default=False, alias="DOCULENS_REQUIRE_AUTH") + serve_frontend: bool = Field(default=False, alias="DOCULENS_SERVE_FRONTEND") + frontend_dist_path: str = Field( + default="/workspace/frontend_dist", alias="DOCULENS_FRONTEND_DIST_PATH" + ) llm: LLMConfig = Field(default_factory=LLMConfig) database: DatabaseConfig = Field(default_factory=DatabaseConfig) @@ -50,6 +55,60 @@ class Settings(BaseSettings): auth_secret_key: str = Field(default="doculens-dev-secret", alias="DOCULENS_AUTH_SECRET") auth_algorithm: str = Field(default="HS256", alias="DOCULENS_AUTH_ALGORITHM") auth_token_exp_minutes: int = Field(default=120, ge=5, alias="DOCULENS_AUTH_TOKEN_EXP_MINUTES") + auth_cookie_name: str = Field(default="doculens_session", alias="DOCULENS_AUTH_COOKIE_NAME") + auth_cookie_secure: bool = Field(default=False, alias="DOCULENS_AUTH_COOKIE_SECURE") + supabase_url: Optional[str] = Field(default=None, alias="SUPABASE_URL") + supabase_publishable_key: Optional[str] = Field( + default=None, + validation_alias=AliasChoices("SUPABASE_PUBLISHABLE_KEY", "SUPABASE_ANON_KEY"), + ) + supabase_allowed_emails: list[str] = Field( + default_factory=list, alias="DOCULENS_SUPABASE_ALLOWED_EMAILS" + ) + supabase_allowed_domains: list[str] = Field( + default_factory=list, alias="DOCULENS_SUPABASE_ALLOWED_DOMAINS" + ) + allow_public_supabase_signin: bool = Field( + default=False, alias="DOCULENS_ALLOW_PUBLIC_SUPABASE_SIGNIN" + ) + supabase_google_enabled: bool = Field( + default=False, alias="DOCULENS_SUPABASE_GOOGLE_ENABLED" + ) + storage_backend: Literal["local", "s3"] = Field( + default="local", alias="DOCULENS_STORAGE_BACKEND" + ) + task_mode: Literal["celery", "local"] = Field( + default="celery", alias="DOCULENS_TASK_MODE" + ) + extraction_backend: Literal["docling", "lightweight"] = Field( + default="docling", alias="DOCULENS_EXTRACTION_BACKEND" + ) + storage_local_path: str = Field( + default="/workspace/app/data/ingestion", alias="DOCULENS_STORAGE_LOCAL_PATH" + ) + s3_endpoint_url: Optional[str] = Field(default=None, alias="DOCULENS_S3_ENDPOINT_URL") + s3_bucket_name: Optional[str] = Field(default=None, alias="DOCULENS_S3_BUCKET_NAME") + s3_access_key_id: Optional[str] = Field(default=None, alias="DOCULENS_S3_ACCESS_KEY_ID") + s3_secret_access_key: Optional[str] = Field( + default=None, alias="DOCULENS_S3_SECRET_ACCESS_KEY" + ) + s3_region: str = Field(default="auto", alias="DOCULENS_S3_REGION") + s3_force_path_style: bool = Field(default=False, alias="DOCULENS_S3_FORCE_PATH_STYLE") + agent_enabled: bool = Field(default=True, alias="DOCULENS_AGENT_ENABLED") + agent_provider: Literal["openai", "anthropic", "openrouter", "llama"] = Field( + default="openai", alias="DOCULENS_AGENT_PROVIDER" + ) + agent_model: Optional[str] = Field(default=None, alias="DOCULENS_AGENT_MODEL") + agent_checkpoint_backend: Literal["memory", "postgres"] = Field( + default="memory", alias="DOCULENS_AGENT_CHECKPOINT_BACKEND" + ) + agent_max_steps: int = Field(default=10, ge=2, le=30, alias="DOCULENS_AGENT_MAX_STEPS") + agent_max_evidence: int = Field( + default=24, ge=4, le=100, alias="DOCULENS_AGENT_MAX_EVIDENCE" + ) + agent_search_limit: int = Field( + default=6, ge=1, le=20, alias="DOCULENS_AGENT_SEARCH_LIMIT" + ) @field_validator("log_level") @classmethod @@ -73,6 +132,50 @@ def assert_production_safe(self) -> None: raise ValueError( "DOCULENS_SEED_DEMO_WORKSPACE requires DOCULENS_SHOWCASE_READ_ONLY=true in production" ) + if self.environment == "production" and not self.showcase_read_only and not self.require_auth: + raise ValueError("DOCULENS_REQUIRE_AUTH must be true in production") + if self.environment == "production" and self.require_auth and not self.auth_cookie_secure: + raise ValueError("DOCULENS_AUTH_COOKIE_SECURE must be true in production") + if ( + self.environment == "production" + and self.require_auth + and (not self.supabase_url or not self.supabase_publishable_key) + ): + raise ValueError( + "SUPABASE_URL and SUPABASE_PUBLISHABLE_KEY are required when authentication " + "is enabled in production" + ) + if ( + self.environment == "production" + and self.require_auth + and not self.allow_public_supabase_signin + and not self.supabase_allowed_emails + and not self.supabase_allowed_domains + ): + raise ValueError( + "Restrict sign-in with DOCULENS_SUPABASE_ALLOWED_EMAILS or " + "DOCULENS_SUPABASE_ALLOWED_DOMAINS, or explicitly set " + "DOCULENS_ALLOW_PUBLIC_SUPABASE_SIGNIN=true" + ) + if self.storage_backend == "s3": + required_storage_values = { + "DOCULENS_S3_ENDPOINT_URL": self.s3_endpoint_url, + "DOCULENS_S3_BUCKET_NAME": self.s3_bucket_name, + "DOCULENS_S3_ACCESS_KEY_ID": self.s3_access_key_id, + "DOCULENS_S3_SECRET_ACCESS_KEY": self.s3_secret_access_key, + } + missing = [name for name, value in required_storage_values.items() if not value] + if missing: + raise ValueError(f"Missing S3 storage settings: {', '.join(missing)}") + if ( + self.environment == "production" + and self.agent_enabled + and not self.showcase_read_only + and self.agent_checkpoint_backend != "postgres" + ): + raise ValueError( + "DOCULENS_AGENT_CHECKPOINT_BACKEND must be postgres for a writable production deployment" + ) @lru_cache diff --git a/app/core/observability.py b/app/core/observability.py index 2848109..4ae2527 100644 --- a/app/core/observability.py +++ b/app/core/observability.py @@ -22,6 +22,20 @@ async def request_context_middleware(request: Request, call_next) -> Response: started = time.perf_counter() response = await call_next(request) response.headers["X-Request-ID"] = request_id + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + response.headers["Cross-Origin-Opener-Policy"] = "same-origin-allow-popups" + response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" + response.headers["Content-Security-Policy"] = ( + "default-src 'self'; " + "script-src 'self'; " + "style-src 'self' 'unsafe-inline'; " + "img-src 'self' data: https:; " + "connect-src 'self' https://*.supabase.co; " + "font-src 'self' data:; object-src 'none'; base-uri 'self'; frame-ancestors 'none'" + ) + if request.url.scheme == "https": + response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" logging.getLogger("doculens.http").info( "request_complete method=%s path=%s status=%s duration_ms=%.2f request_id=%s", request.method, diff --git a/app/database/database_utils.py b/app/database/database_utils.py index f6ad2a5..baf5757 100644 --- a/app/database/database_utils.py +++ b/app/database/database_utils.py @@ -15,6 +15,12 @@ class DatabaseUtils: @staticmethod def get_connection_string(): + database_url = os.getenv("DATABASE_URL") + if database_url: + if database_url.startswith("postgres://"): + return "postgresql://" + database_url.removeprefix("postgres://") + return database_url + db_host = os.getenv("DATABASE_HOST", "localhost") db_port = os.getenv("DATABASE_PORT", "5432") db_name = os.getenv("DATABASE_NAME", "postgres") diff --git a/app/database/investigation.py b/app/database/investigation.py new file mode 100644 index 0000000..5e2d6e6 --- /dev/null +++ b/app/database/investigation.py @@ -0,0 +1,44 @@ +"""Durable investigation summaries separate from LangGraph's step checkpoints.""" + +from __future__ import annotations + +import uuid + +from datetime import datetime +from typing import Any + +from sqlalchemy import DateTime, ForeignKey, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.database.session import Base + + +class Investigation(Base): + """Queryable product record for an agent run and its final evidence ledger.""" + + __tablename__ = "investigations" + + id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + user_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("users.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + goal: Mapped[str] = mapped_column(Text, nullable=False) + status: Mapped[str] = mapped_column( + String(32), nullable=False, default="queued", index=True + ) + state: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + created_at: Mapped[datetime] = mapped_column( + DateTime, nullable=False, server_default=func.now(), index=True + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + nullable=False, + server_default=func.now(), + onupdate=func.now(), + ) diff --git a/app/database/user.py b/app/database/user.py index 4245403..b546bd9 100644 --- a/app/database/user.py +++ b/app/database/user.py @@ -1,7 +1,7 @@ import uuid from datetime import datetime -from sqlalchemy import Boolean, Column, DateTime, String +from sqlalchemy import Boolean, Column, DateTime, String, UniqueConstraint from sqlalchemy.dialects.postgresql import UUID from app.database.session import Base @@ -11,6 +11,9 @@ class User(Base): """Account record used for authentication and role-based access controls.""" __tablename__ = "users" + __table_args__ = ( + UniqueConstraint("auth_provider", "provider_subject", name="uq_users_auth_identity"), + ) id = Column( UUID(as_uuid=True), @@ -21,6 +24,9 @@ class User(Base): email = Column(String(255), unique=True, nullable=False, index=True) full_name = Column(String(255), nullable=False) hashed_password = Column(String(255), nullable=False) + auth_provider = Column(String(32), nullable=False, default="password") + provider_subject = Column(String(255), nullable=True) + avatar_url = Column(String(1024), nullable=True) persona = Column(String(50), nullable=False) role = Column(String(50), nullable=False) access_level = Column(String(50), nullable=False, default="standard") diff --git a/app/doc_utils/chunking.py b/app/doc_utils/chunking.py index 7eb412e..5e64faa 100644 --- a/app/doc_utils/chunking.py +++ b/app/doc_utils/chunking.py @@ -1,15 +1,80 @@ +from dataclasses import dataclass +from pathlib import Path from typing import Any, List -from docling_core.transforms.chunker.hybrid_chunker import HybridChunker - from app.config.settings import get_settings -from app.doc_utils.utils.tokenizer import OpenAITokenizerWrapper +from app.doc_utils.extraction import ExtractedTextDocument +from app.doc_utils.utils.tokenizer import ( + OpenAITokenizerWrapper, + create_docling_tokenizer, +) # Deliberately much smaller than the embedding model limit. Retrieval quality is # generally better with focused chunks than with page-sized 8k token passages. DEFAULT_MAX_TOKENS: int = 800 +@dataclass(frozen=True) +class _Origin: + filename: str + + +@dataclass(frozen=True) +class _Provenance: + page_no: int + + +@dataclass(frozen=True) +class _DocumentItem: + prov: List[_Provenance] + + +@dataclass(frozen=True) +class _ChunkMetadata: + origin: _Origin + doc_items: List[_DocumentItem] + headings: List[str] + + +@dataclass(frozen=True) +class LightweightChunk: + text: str + meta: _ChunkMetadata + + +def _chunk_lightweight_document( + document: ExtractedTextDocument, + *, + max_tokens: int, +) -> List[LightweightChunk]: + # Keep the lightweight deployment profile independent from tokenizer model + # downloads. English prose averages roughly 0.75 words per model token; the + # embedding layer still applies the provider's exact token limit later. + max_words = max(1, int(max_tokens * 0.75)) + overlap_words = min(60, max_words // 5) + stride = max(1, max_words - overlap_words) + chunks: List[LightweightChunk] = [] + for page in document.pages: + words = page.text.split() + for offset in range(0, len(words), stride): + window = words[offset : offset + max_words] + if not window: + continue + chunks.append( + LightweightChunk( + text=" ".join(window), + meta=_ChunkMetadata( + origin=_Origin(filename=document.filename), + doc_items=[_DocumentItem(prov=[_Provenance(page_no=page.page_number)])], + headings=[Path(document.filename).stem], + ), + ) + ) + if offset + max_words >= len(words): + break + return chunks + + def chunk_document( docling_document: Any, max_tokens: int | None = None, @@ -32,7 +97,14 @@ def chunk_document( raise ValueError("Cannot chunk a null document. Ensure extraction succeeded.") effective_max_tokens = max_tokens or get_settings().chunk_max_tokens - tokenizer = OpenAITokenizerWrapper() + if isinstance(docling_document, ExtractedTextDocument): + return _chunk_lightweight_document( + docling_document, + max_tokens=effective_max_tokens, + ) + from docling_core.transforms.chunker.hybrid_chunker import HybridChunker + + tokenizer = create_docling_tokenizer(OpenAITokenizerWrapper()) chunker = HybridChunker( tokenizer=tokenizer, merge_peers=merge_peers, diff --git a/app/doc_utils/embedding.py b/app/doc_utils/embedding.py index f2cb837..9c16653 100644 --- a/app/doc_utils/embedding.py +++ b/app/doc_utils/embedding.py @@ -1,11 +1,10 @@ """Chunk preparation and batched embedding persistence.""" -from datetime import datetime, timedelta, timezone from functools import lru_cache from typing import Any, Dict, List, Optional +from uuid import uuid1 import pandas as pd -from timescale_vector.client import uuid_from_time from app.doc_utils.utils.tokenizer import OpenAITokenizerWrapper from app.services.vector_store import VectorStore @@ -88,8 +87,7 @@ def embed_and_upsert_chunks( tokenizer = _get_tokenizer() texts = [text for _, text in non_empty] embeddings = store.embed_texts(texts, model=embedding_model) - base_time = datetime.now(timezone.utc) - ids = [str(uuid_from_time(base_time + timedelta(microseconds=index))) for index in range(len(texts))] + ids = [str(uuid1()) for _ in texts] metadata = [ _chunk_metadata(chunk, document_id, index, tokenizer.count_tokens(text), document_metadata) for index, ((chunk, text)) in enumerate(non_empty) diff --git a/app/doc_utils/extraction.py b/app/doc_utils/extraction.py index 581e4f9..0bc1711 100644 --- a/app/doc_utils/extraction.py +++ b/app/doc_utils/extraction.py @@ -1,12 +1,25 @@ import os +from dataclasses import dataclass from pathlib import Path - -from docling.document_converter import DocumentConverter +from typing import List EASYOCR_HOME = Path(os.getenv("EASYOCR_HOME", Path.home() / ".EasyOCR")) EASYOCR_HOME.mkdir(parents=True, exist_ok=True) os.environ.setdefault("EASYOCR_HOME", str(EASYOCR_HOME)) + +@dataclass(frozen=True) +class ExtractedTextPage: + page_number: int + text: str + + +@dataclass(frozen=True) +class ExtractedTextDocument: + filename: str + pages: List[ExtractedTextPage] + + def extract_docling_document(source: str): """ Extract a Docling document from a file path or URL. @@ -15,6 +28,44 @@ def extract_docling_document(source: str): Returns: Docling document object or None if extraction fails. """ + # Docling is intentionally imported on demand. Its ML stack is large, and + # loading it during API startup wastes memory for search-only requests. + from docling.document_converter import DocumentConverter + converter = DocumentConverter() result = converter.convert(source) return result.document if result else None + + +def extract_lightweight_document(source: str) -> ExtractedTextDocument: + """Extract text PDFs and plain-text files without loading an OCR model.""" + path = Path(source) + if path.suffix.lower() == ".pdf": + from pypdf import PdfReader + + reader = PdfReader(str(path)) + pages = [ + ExtractedTextPage(page_number=index + 1, text=(page.extract_text() or "").strip()) + for index, page in enumerate(reader.pages) + ] + else: + pages = [ + ExtractedTextPage( + page_number=1, + text=path.read_text(encoding="utf-8", errors="replace").strip(), + ) + ] + if not any(page.text for page in pages): + raise ValueError( + "No embedded text was found. Use the Docling extraction profile for scanned documents." + ) + return ExtractedTextDocument(filename=path.name, pages=pages) + + +def extract_document(source: str): + """Select the configured extraction strategy.""" + from app.config.settings import get_settings + + if get_settings().extraction_backend == "lightweight": + return extract_lightweight_document(source) + return extract_docling_document(source) diff --git a/app/doc_utils/search.py b/app/doc_utils/search.py index 683016a..d015913 100644 --- a/app/doc_utils/search.py +++ b/app/doc_utils/search.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from app.services.vector_store import VectorStore @@ -6,7 +6,7 @@ def semantic_search_docling( query: str, limit: int = 5, - metadata_filter: Optional[Dict[str, Any]] = None, + metadata_filter: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None, ) -> List[Dict[str, Any]]: """Perform semantic search over the Docling chunks stored in the vector store. @@ -19,7 +19,7 @@ def semantic_search_docling( List of result dictionaries (safe for JSON serialization). """ vector_store = VectorStore() - results = vector_store.semantic_search( + results: Any = vector_store.semantic_search( query=query, limit=limit, metadata_filter=metadata_filter, diff --git a/app/doc_utils/utils/tokenizer.py b/app/doc_utils/utils/tokenizer.py index 22b979c..ab62f4e 100644 --- a/app/doc_utils/utils/tokenizer.py +++ b/app/doc_utils/utils/tokenizer.py @@ -1,18 +1,10 @@ from typing import Any, List -from docling_core.transforms.chunker.tokenizer.base import BaseTokenizer -from pydantic import ConfigDict from tiktoken import get_encoding -class OpenAITokenizerWrapper(BaseTokenizer): - """Wrapper to make OpenAI's tiktoken tokenizer compatible with Docling's HybridChunker.""" - - model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow") - - model_name: str - max_length: int - tokenizer: Any +class OpenAITokenizerWrapper: + """Small tiktoken adapter shared by lightweight and Docling profiles.""" def __init__(self, model_name: str = "cl100k_base", max_length: int = 8191): """ @@ -20,9 +12,9 @@ def __init__(self, model_name: str = "cl100k_base", max_length: int = 8191): model_name: The name of the tiktoken encoding to use (e.g. 'cl100k_base') max_length: Maximum number of tokens the model can handle """ - object.__setattr__(self, "model_name", model_name) - object.__setattr__(self, "max_length", max_length) - object.__setattr__(self, "tokenizer", get_encoding(model_name)) + self.model_name = model_name + self.max_length = max_length + self.tokenizer = get_encoding(model_name) # --- Required abstract methods for BaseTokenizer --- def get_tokenizer(self): @@ -49,3 +41,27 @@ def detokenize(self, tokens: List[str]) -> str: # --- Optional helper methods --- def vocab_size(self) -> int: return self.tokenizer.max_token_value + + +def create_docling_tokenizer( + tokenizer: OpenAITokenizerWrapper | None = None, +) -> Any: + """Adapt tiktoken to Docling without importing its ML stack in the web profile.""" + from docling_core.transforms.chunker.tokenizer.base import BaseTokenizer + from pydantic import ConfigDict + + class _DoclingTokenizerAdapter(BaseTokenizer): + model_config = ConfigDict(arbitrary_types_allowed=True) + + delegate: Any + + def get_tokenizer(self) -> Any: + return self.delegate.get_tokenizer() + + def get_max_tokens(self) -> int: + return self.delegate.get_max_tokens() + + def count_tokens(self, text: str) -> int: + return self.delegate.count_tokens(text) + + return _DoclingTokenizerAdapter(delegate=tokenizer or OpenAITokenizerWrapper()) diff --git a/app/evaluation/investigation.py b/app/evaluation/investigation.py new file mode 100644 index 0000000..609fce3 --- /dev/null +++ b/app/evaluation/investigation.py @@ -0,0 +1,103 @@ +"""Deterministic metrics for regression-testing agent investigations.""" + +from __future__ import annotations + +from dataclasses import dataclass +from statistics import mean + +from app.agents.models import InvestigationReport, InvestigationState, VerificationResult + + +@dataclass(frozen=True) +class InvestigationExpectation: + """Labelled success criteria for one representative investigation goal.""" + + expected_document_ids: frozenset[str] = frozenset() + required_report_terms: frozenset[str] = frozenset() + minimum_citation_coverage: float = 1.0 + maximum_steps: int = 10 + + +@dataclass(frozen=True) +class InvestigationMetrics: + completed: bool + document_recall: float + required_term_coverage: float + citation_coverage: float + meets_citation_threshold: bool + within_step_budget: bool + + @property + def passed(self) -> bool: + return ( + self.completed + and self.document_recall == 1.0 + and self.required_term_coverage == 1.0 + and self.meets_citation_threshold + and self.within_step_budget + ) + + +def evaluate_investigation( + state: InvestigationState, + expectation: InvestigationExpectation, +) -> InvestigationMetrics: + """Score observable state without another model call.""" + evidence_documents = { + str(item.get("document_id")) + for item in state.get("evidence", []) + if item.get("document_id") + } + expected_documents = expectation.expected_document_ids + document_recall = ( + len(expected_documents & evidence_documents) / len(expected_documents) + if expected_documents + else 1.0 + ) + + report = InvestigationReport.model_validate(state["report"]) + report_text = " ".join( + [ + report.title, + report.executive_summary, + *(finding.claim for finding in report.findings), + *(finding.explanation for finding in report.findings), + ] + ).lower() + required_terms = {term.lower() for term in expectation.required_report_terms} + required_term_coverage = ( + sum(term in report_text for term in required_terms) / len(required_terms) + if required_terms + else 1.0 + ) + verification = VerificationResult.model_validate(state["verification"]) + + return InvestigationMetrics( + completed=state["status"] == "completed", + document_recall=document_recall, + required_term_coverage=required_term_coverage, + citation_coverage=verification.citation_coverage, + meets_citation_threshold=( + verification.citation_coverage >= expectation.minimum_citation_coverage + ), + within_step_budget=state["step_count"] <= expectation.maximum_steps, + ) + + +def aggregate_investigation_metrics( + metrics: list[InvestigationMetrics], +) -> dict[str, float]: + """Return CI-friendly aggregate scores for an evaluation suite.""" + if not metrics: + return { + "pass_rate": 0.0, + "document_recall": 0.0, + "required_term_coverage": 0.0, + "citation_coverage": 0.0, + } + return { + "pass_rate": mean(float(item.passed) for item in metrics), + "document_recall": mean(item.document_recall for item in metrics), + "required_term_coverage": mean(item.required_term_coverage for item in metrics), + "citation_coverage": mean(item.citation_coverage for item in metrics), + } diff --git a/app/frontend.py b/app/frontend.py new file mode 100644 index 0000000..5d51880 --- /dev/null +++ b/app/frontend.py @@ -0,0 +1,18 @@ +"""Static frontend serving with single-page application route fallback.""" + +from fastapi import status +from fastapi.staticfiles import StaticFiles +from starlette.exceptions import HTTPException +from starlette.types import Scope + + +class SpaStaticFiles(StaticFiles): + """Serve the React entrypoint for client-side application routes.""" + + async def get_response(self, path: str, scope: Scope): + try: + return await super().get_response(path, scope) + except HTTPException as exc: + if exc.status_code != status.HTTP_404_NOT_FOUND: + raise + return await super().get_response("index.html", scope) diff --git a/app/main.py b/app/main.py index f7096b6..cdc640e 100644 --- a/app/main.py +++ b/app/main.py @@ -2,16 +2,25 @@ import logging from contextlib import asynccontextmanager, contextmanager -from typing import Iterator - -from fastapi import FastAPI +from pathlib import Path +from typing import Any, Iterator, cast + +from fastapi import FastAPI, HTTPException, status +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.checkpoint.postgres import PostgresSaver +from psycopg.rows import dict_row +from psycopg_pool import ConnectionPool from fastapi.middleware.cors import CORSMiddleware +from sqlalchemy import text from sqlalchemy.orm import Session from app.api.router import router as api_router +from app.agents.graph import InvestigationRuntime from app.config.settings import Settings, get_settings from app.core.observability import configure_logging, request_context_middleware +from app.frontend import SpaStaticFiles from app.database import event as _event_model # noqa: F401 +from app.database import investigation as _investigation_model # noqa: F401 from app.database import user as _user_model # noqa: F401 from app.database.session import Base, SessionLocal, engine from app.services.auth_service import AuthService @@ -49,6 +58,7 @@ def initialize_dependencies(settings: Settings) -> None: AuthService(session).ensure_seed_users(DEMO_USERS) store = VectorStore() store.create_tables() + store.create_index() store.create_keyword_search_index() if settings.seed_demo_workspace: with session_scope() as session: @@ -67,12 +77,37 @@ def create_app(settings: Settings | None = None) -> FastAPI: configure_logging(runtime.log_level) @asynccontextmanager - async def lifespan(_: FastAPI): + async def lifespan(application: FastAPI): try: initialize_dependencies(runtime) except Exception: logger.exception("Dependency initialization failed; readiness will remain unhealthy") - yield + checkpoint_pool = None + if runtime.agent_enabled: + if runtime.agent_checkpoint_backend == "postgres": + checkpoint_pool = ConnectionPool( + conninfo=runtime.database.service_url, + min_size=1, + max_size=5, + open=False, + kwargs={ + "autocommit": True, + "prepare_threshold": 0, + "row_factory": dict_row, + }, + ) + checkpoint_pool.open() + checkpoint_pool.wait() + checkpointer = PostgresSaver(cast(Any, checkpoint_pool)) + checkpointer.setup() + else: + checkpointer = InMemorySaver() + application.state.investigation_runtime = InvestigationRuntime(checkpointer) + try: + yield + finally: + if checkpoint_pool is not None: + checkpoint_pool.close() application = FastAPI( title=runtime.app_name, @@ -95,6 +130,27 @@ async def lifespan(_: FastAPI): def liveness() -> dict[str, str]: return {"status": "ok"} + @application.get("/health/ready", tags=["health"], summary="Dependency readiness") + def readiness() -> dict[str, str]: + try: + with engine.connect() as connection: + connection.execute(text("SELECT 1")) + except Exception as exc: + logger.warning("Readiness database check failed: %s", exc) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Database is unavailable.", + ) from exc + return {"status": "ready"} + + if runtime.serve_frontend: + frontend_dist = Path(runtime.frontend_dist_path) + if not (frontend_dist / "index.html").exists(): + raise RuntimeError( + f"DOCULENS_SERVE_FRONTEND is enabled but {frontend_dist}/index.html is missing" + ) + application.mount("/", SpaStaticFiles(directory=frontend_dist, html=True), name="frontend") + return application diff --git a/app/pipelines/doculens_pipeline.py b/app/pipelines/doculens_pipeline.py index 380b75d..c747b1f 100644 --- a/app/pipelines/doculens_pipeline.py +++ b/app/pipelines/doculens_pipeline.py @@ -23,7 +23,7 @@ from app.core.schema import NodeConfig, PipelineSchema from app.doc_utils.chunking import chunk_document from app.doc_utils.embedding import embed_and_upsert_chunks -from app.doc_utils.extraction import extract_docling_document +from app.doc_utils.extraction import extract_document from app.doc_utils.search import semantic_search_docling from app.services.llm_factory import LLMFactory from app.services.classification_service import ClassificationResult, ClassificationScore @@ -31,6 +31,7 @@ from app.database.session import SessionLocal from app.services.prompt_loader import PromptManager from app.services.vector_store import VectorStore +from app.services.document_storage import get_document_storage logger = logging.getLogger(__name__) @@ -160,6 +161,7 @@ def process(self, task_context): ingestion_dir = _ensure_ingestion_dir() source_name = Path(event.filename or f"{document_id}.bin").name local_path = ingestion_dir / f"{document_id}_{source_name}" + temporary_local_path = False if event.file_url: logger.info("Downloading document for ingestion: %s", event.file_url) @@ -167,12 +169,10 @@ def process(self, task_context): response.raise_for_status() local_path.write_bytes(response.content) else: - source_path = Path(event.filename) - if not source_path.exists(): - raise FileNotFoundError( - f"DocumentUploadEvent filename '{event.filename}' not found and no file_url provided." - ) - local_path = source_path.resolve() + local_path, temporary_local_path = get_document_storage().materialize( + event.filename, + destination_dir=ingestion_dir, + ) upload_metadata = event.metadata or {} original_filename = upload_metadata.get("uploaded_filename") or upload_metadata.get("original_filename") @@ -180,7 +180,7 @@ def process(self, task_context): # Fall back to stripping any UUID prefix we added during upload. original_filename = source_name.partition("_")[2] or source_name - docling_doc = extract_docling_document(str(local_path)) + docling_doc = extract_document(str(local_path)) if docling_doc is None: raise ValueError("Docling conversion failed to produce a document.") @@ -188,6 +188,7 @@ def process(self, task_context): task_context.state["docling_doc"] = docling_doc task_context.state["local_path"] = str(local_path) + task_context.state["temporary_local_path"] = temporary_local_path task_context.metadata["document"] = { "id": document_id, "original_filename": original_filename, @@ -271,6 +272,9 @@ def process(self, task_context): # Cleanup heavy artifacts task_context.state.pop("chunks", None) task_context.state.pop("docling_doc", None) + local_path = task_context.state.pop("local_path", None) + if task_context.state.pop("temporary_local_path", False) and local_path: + Path(local_path).unlink(missing_ok=True) return task_context diff --git a/app/services/auth_service.py b/app/services/auth_service.py index f22e58a..ef07206 100644 --- a/app/services/auth_service.py +++ b/app/services/auth_service.py @@ -1,9 +1,11 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone +import secrets from typing import Dict, Iterable, Optional -from fastapi import Depends, HTTPException, status +import httpx +from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from passlib.context import CryptContext @@ -88,6 +90,105 @@ def authenticate_user(self, *, email: str, password: str) -> Optional[User]: return None return user + def get_user_by_provider_subject(self, *, provider: str, subject: str) -> Optional[User]: + return ( + self.session.query(User) + .filter( + User.auth_provider == provider, + User.provider_subject == subject, + ) + .first() + ) + + def authenticate_supabase(self, *, access_token: str) -> User: + """Verify a Supabase access token and provision or link its local account.""" + settings = get_settings() + if not settings.supabase_url or not settings.supabase_publishable_key: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Supabase authentication is not configured.", + ) + + try: + with httpx.Client(timeout=settings.provider_timeout_seconds) as client: + response = client.get( + f"{settings.supabase_url.rstrip('/')}/auth/v1/user", + headers={ + "apikey": settings.supabase_publishable_key, + "Authorization": f"Bearer {access_token}", + }, + ) + response.raise_for_status() + claims = response.json() + except (httpx.HTTPError, ValueError) as exc: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Supabase could not verify this sign-in.", + ) from exc + + return self._upsert_supabase_user(claims) + + def _upsert_supabase_user(self, claims: Dict[str, object]) -> User: + subject = str(claims.get("id") or claims.get("sub") or "").strip() + email = str(claims.get("email") or "").strip().lower() + if not subject or not email: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Supabase account identity is incomplete.", + ) + + self._enforce_supabase_allowlist(email=email) + + user = self.get_user_by_provider_subject(provider="supabase", subject=subject) + if user is None: + user = self.get_user_by_email(email=email) + + raw_metadata = claims.get("user_metadata") + metadata = raw_metadata if isinstance(raw_metadata, dict) else {} + full_name = str(metadata.get("full_name") or email.split("@", maxsplit=1)[0]).strip() + avatar_url = str(metadata.get("avatar_url") or "").strip() or None + + if user is None: + user = User( + email=email, + full_name=full_name, + hashed_password=self.hash_password(secrets.token_urlsafe(32)), + auth_provider="supabase", + provider_subject=subject, + avatar_url=avatar_url, + persona="analyst", + role="analyst", + access_level=ROLE_DEFINITIONS["analyst"]["access_level"], + ) + else: + user.email = email + user.full_name = full_name + user.auth_provider = "supabase" + user.provider_subject = subject + user.avatar_url = avatar_url + + self.session.add(user) + self.session.commit() + self.session.refresh(user) + return user + + @staticmethod + def _enforce_supabase_allowlist(*, email: str) -> None: + settings = get_settings() + if settings.allow_public_supabase_signin: + return + + allowed_emails = {item.strip().lower() for item in settings.supabase_allowed_emails} + allowed_domains = {item.strip().lower() for item in settings.supabase_allowed_domains} + domain = email.rsplit("@", maxsplit=1)[-1] if "@" in email else "" + if email in allowed_emails or domain in allowed_domains: + return + + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="This email address is not allowed to access the workspace.", + ) + def create_user( self, *, @@ -186,13 +287,29 @@ def decode_access_token(token: str, session: Session) -> User: def get_current_user( + request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme), session: Session = Depends(db_session), ) -> User: - if credentials is None: + settings = get_settings() + token = credentials.credentials if credentials is not None else request.cookies.get( + settings.auth_cookie_name + ) + if not token: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authorization credentials.", headers={"WWW-Authenticate": "Bearer"}, ) - return decode_access_token(credentials.credentials, session) + return decode_access_token(token, session) + + +def require_current_user_if_enabled( + request: Request, + credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer_scheme), + session: Session = Depends(db_session), +) -> Optional[User]: + """Require an application session only when the deployment enables auth.""" + if not get_settings().require_auth: + return None + return get_current_user(request=request, credentials=credentials, session=session) diff --git a/app/services/document_storage.py b/app/services/document_storage.py new file mode 100644 index 0000000..7fe5cc8 --- /dev/null +++ b/app/services/document_storage.py @@ -0,0 +1,141 @@ +"""Durable document storage shared by the API and background processors.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import Protocol + +import boto3 +from botocore.config import Config +from fastapi import HTTPException, UploadFile, status + +from app.config.settings import Settings, get_settings + + +@dataclass(frozen=True) +class StoredDocument: + """Reference returned after a document has been persisted.""" + + reference: str + bytes_written: int + + +class DocumentStorage(Protocol): + async def store_upload( + self, + upload: UploadFile, + *, + object_name: str, + max_bytes: int, + ) -> StoredDocument: ... + + def materialize(self, reference: str, *, destination_dir: Path) -> tuple[Path, bool]: ... + + +async def _write_bounded_upload(upload: UploadFile, destination: Path, max_bytes: int) -> int: + bytes_written = 0 + try: + with destination.open("wb") as buffer: + while True: + chunk = await upload.read(1 << 20) + if not chunk: + break + bytes_written += len(chunk) + if bytes_written > max_bytes: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail=f"Upload exceeds the {max_bytes}-byte limit.", + ) + buffer.write(chunk) + except Exception: + destination.unlink(missing_ok=True) + raise + finally: + await upload.close() + return bytes_written + + +class LocalDocumentStorage: + """Filesystem storage for local development and single-process deployments.""" + + def __init__(self, root: Path): + self.root = root + + async def store_upload( + self, + upload: UploadFile, + *, + object_name: str, + max_bytes: int, + ) -> StoredDocument: + self.root.mkdir(parents=True, exist_ok=True) + destination = self.root / Path(object_name).name + bytes_written = await _write_bounded_upload(upload, destination, max_bytes) + return StoredDocument(reference=str(destination.resolve()), bytes_written=bytes_written) + + def materialize(self, reference: str, *, destination_dir: Path) -> tuple[Path, bool]: + path = Path(reference) + if not path.exists(): + raise FileNotFoundError(f"Stored document '{reference}' does not exist.") + return path.resolve(), False + + +class S3DocumentStorage: + """Private S3-compatible storage for Railway Buckets, R2, or S3.""" + + def __init__(self, settings: Settings): + self.bucket_name = settings.s3_bucket_name or "" + self.client = boto3.client( + "s3", + endpoint_url=settings.s3_endpoint_url, + aws_access_key_id=settings.s3_access_key_id, + aws_secret_access_key=settings.s3_secret_access_key, + region_name=settings.s3_region, + config=Config( + s3={"addressing_style": "path" if settings.s3_force_path_style else "virtual"} + ), + ) + + async def store_upload( + self, + upload: UploadFile, + *, + object_name: str, + max_bytes: int, + ) -> StoredDocument: + key = f"documents/{Path(object_name).name}" + with TemporaryDirectory(prefix="doculens-upload-") as temporary_dir: + temporary_path = Path(temporary_dir) / Path(object_name).name + bytes_written = await _write_bounded_upload(upload, temporary_path, max_bytes) + await asyncio.to_thread( + self.client.upload_file, + str(temporary_path), + self.bucket_name, + key, + ) + return StoredDocument( + reference=f"s3://{self.bucket_name}/{key}", + bytes_written=bytes_written, + ) + + def materialize(self, reference: str, *, destination_dir: Path) -> tuple[Path, bool]: + prefix = f"s3://{self.bucket_name}/" + if not reference.startswith(prefix): + raise ValueError("Document storage reference does not belong to the configured bucket.") + key = reference.removeprefix(prefix) + destination_dir.mkdir(parents=True, exist_ok=True) + destination = destination_dir / Path(key).name + self.client.download_file(self.bucket_name, key, str(destination)) + return destination.resolve(), True + + +@lru_cache +def get_document_storage() -> DocumentStorage: + settings = get_settings() + if settings.storage_backend == "s3": + return S3DocumentStorage(settings) + return LocalDocumentStorage(Path(settings.storage_local_path)) diff --git a/app/services/investigation_service.py b/app/services/investigation_service.py new file mode 100644 index 0000000..eba6a01 --- /dev/null +++ b/app/services/investigation_service.py @@ -0,0 +1,96 @@ +"""Persistence helpers for user-facing investigation records.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.agents.models import InvestigationState +from app.database.investigation import Investigation + + +def create_investigation( + session: Session, + *, + investigation_id: UUID, + goal: str, + user_id: UUID | None, + initial_state: InvestigationState, +) -> Investigation: + record = Investigation( + id=investigation_id, + user_id=user_id, + goal=goal, + status=initial_state["status"], + state=dict(initial_state), + ) + session.add(record) + session.flush() + return record + + +def persist_investigation_state( + session: Session, + investigation_id: UUID, + state: InvestigationState, +) -> None: + record = session.get(Investigation, investigation_id) + if record is None: + raise LookupError(f"Investigation {investigation_id} does not exist") + record.status = state.get("status", record.status) + record.state = dict(state) + session.add(record) + + +def get_investigation( + session: Session, + investigation_id: UUID, + *, + user_id: UUID | None, +) -> Investigation | None: + statement = select(Investigation).where(Investigation.id == investigation_id) + if user_id is not None: + statement = statement.where(Investigation.user_id == user_id) + return session.execute(statement).scalar_one_or_none() + + +def list_investigations( + session: Session, + *, + user_id: UUID | None, + limit: int, +) -> list[Investigation]: + statement = select(Investigation) + if user_id is not None: + statement = statement.where(Investigation.user_id == user_id) + statement = statement.order_by(Investigation.created_at.desc()).limit(limit) + return list(session.execute(statement).scalars()) + + +def serialize_investigation( + record: Investigation, + *, + include_state: bool = True, +) -> dict[str, Any]: + state = record.state or {} + payload: dict[str, Any] = { + "id": str(record.id), + "goal": record.goal, + "status": record.status, + "created_at": _as_iso(record.created_at), + "updated_at": _as_iso(record.updated_at), + "evidence_count": len(state.get("evidence") or []), + "step_count": state.get("step_count", 0), + "report_title": (state.get("report") or {}).get("title"), + } + if include_state: + payload["state"] = state + return payload + + +def _as_iso(value: datetime | None) -> str | None: + return value.isoformat() if value is not None else None diff --git a/app/services/vector_store.py b/app/services/vector_store.py index c32e10c..e2b24e9 100644 --- a/app/services/vector_store.py +++ b/app/services/vector_store.py @@ -1,4 +1,6 @@ import logging +import json +import re from collections import OrderedDict from datetime import datetime from pathlib import Path @@ -7,20 +9,19 @@ import pandas as pd import psycopg2 -from psycopg2.extras import RealDictCursor +from psycopg2.extras import RealDictCursor, execute_values from app.config.settings import get_settings from openai import OpenAI -from timescale_vector import client """ Vector Store Management Module This module provides functionality for managing vector embeddings and similarity search -operations using TimescaleDB and OpenAI embeddings. It supports semantic search, +operations using PostgreSQL, pgvector, and OpenAI embeddings. It supports semantic search, keyword search, and hybrid search capabilities with metadata filtering. -The implementation uses the timescale-vector client for efficient vector operations -and supports both exact and approximate nearest neighbor search through StreamingDiskANN. +The implementation uses portable SQL and an HNSW index so it runs on managed +PostgreSQL providers without proprietary extensions. """ @@ -32,7 +33,7 @@ class VectorStore: def __init__(self, local: bool = False): """ - Initialize the VectorStore with settings, OpenAI client, and Timescale Vector client. + Initialize the vector store with settings and an OpenAI client. Args: local (bool): If True, overrides .env to use localhost DB for running outside Docker. @@ -47,12 +48,9 @@ def __init__(self, local: bool = False): self.vector_settings = self.settings.database.vector_store database_url = self.settings.database.service_url_for(local=local) self.database_url = database_url - self.vec_client = client.Sync( - database_url, - self.vector_settings.table_name, - self.vector_settings.embedding_dimensions, - time_partition_interval=self.vector_settings.time_partition_interval, - ) + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", self.vector_settings.table_name): + raise ValueError("Vector table name must be a valid PostgreSQL identifier.") + self.table_name = self.vector_settings.table_name def create_keyword_search_index(self): """Create a GIN index for keyword search if it doesn't exist.""" @@ -115,16 +113,40 @@ def embed_texts(self, texts: List[str], model: Optional[str] = None) -> List[Lis return [vector for vector in results if vector is not None] def create_tables(self) -> None: - """Create the necessary tablesin the database""" - self.vec_client.create_tables() + """Create the portable pgvector table and metadata index.""" + query = f""" + CREATE EXTENSION IF NOT EXISTS vector; + CREATE TABLE IF NOT EXISTS {self.table_name} ( + id UUID PRIMARY KEY, + metadata JSONB NOT NULL DEFAULT '{{}}'::jsonb, + contents TEXT NOT NULL, + embedding VECTOR({self.vector_settings.embedding_dimensions}) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + ALTER TABLE {self.table_name} + ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(); + CREATE INDEX IF NOT EXISTS idx_{self.table_name}_metadata + ON {self.table_name} USING GIN(metadata jsonb_path_ops); + """ + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + cursor.execute(query) def create_index(self) -> None: - """Create the StreamingDiskANN index to spseed up similarity search""" - self.vec_client.create_embedding_index(client.DiskAnnIndex()) + """Create an HNSW cosine index supported by standard pgvector.""" + query = f""" + CREATE INDEX IF NOT EXISTS idx_{self.table_name}_embedding_hnsw + ON {self.table_name} USING hnsw (embedding vector_cosine_ops) + """ + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + cursor.execute(query) def drop_index(self) -> None: - """Drop the StreamingDiskANN index in the database""" - self.vec_client.drop_embedding_index() + """Drop the HNSW embedding index.""" + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + cursor.execute(f"DROP INDEX IF EXISTS idx_{self.table_name}_embedding_hnsw") def upsert(self, df: pd.DataFrame) -> None: """ @@ -134,8 +156,31 @@ def upsert(self, df: pd.DataFrame) -> None: df: A pandas DataFrame containing the data to insert or update. Expected columns: id, metadata, contents, embedding """ - records = df.to_records(index=False) - self.vec_client.upsert(list(records)) + records = [ + ( + str(row["id"]), + json.dumps(row["metadata"] or {}), + str(row["contents"]), + self._vector_literal(row["embedding"]), + ) + for _, row in df.iterrows() + ] + query = f""" + INSERT INTO {self.table_name} (id, metadata, contents, embedding) + VALUES %s + ON CONFLICT (id) DO UPDATE SET + metadata = EXCLUDED.metadata, + contents = EXCLUDED.contents, + embedding = EXCLUDED.embedding + """ + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + execute_values( + cursor, + query, + records, + template="(%s::uuid, %s::jsonb, %s, %s::vector)", + ) logging.info( f"Inserted {len(df)} records into {self.vector_settings.table_name}" ) @@ -144,26 +189,19 @@ def semantic_search( self, query: str, limit: int = 5, - metadata_filter: Union[dict, List[dict]] = None, - predicates: Optional[client.Predicates] = None, + metadata_filter: Optional[Union[dict, List[dict]]] = None, + predicates: Optional[Any] = None, time_range: Optional[Tuple[datetime, datetime]] = None, return_dataframe: bool = True, ) -> Union[List[Tuple[Any, ...]], pd.DataFrame]: """ Query the vector database for similar embeddings based on input text. - More info: - https://github.com/timescale/docs/blob/latest/ai/python-interface-for-pgvector-and-timescale-vector.md - Args: query: The input text to search for. limit: The maximum number of results to return. metadata_filter: A dictionary or list of dictionaries for equality-based metadata filtering. - predicates: A Predicates object for complex metadata filtering. - - Predicates objects are defined by the name of the metadata key, an operator, and a value. - - Operators: ==, !=, >, >=, <, <= - - & is used to combine multiple predicates with AND operator. - - | is used to combine multiple predicates with OR operator. + predicates: Deprecated compatibility parameter. Use metadata_filter. time_range: A tuple of (start_date, end_date) to filter results by time. return_dataframe: Whether to return results as a DataFrame (default: True). @@ -176,41 +214,51 @@ def semantic_search( Search with metadata filter: vector_store.semantic_search("Shipping options", metadata_filter={"category": "Shipping"}) - Predicates Examples: - Search with predicates: - vector_store.semantic_search("Pricing", predicates=client.Predicates("price", ">", 100)) - Search with complex combined predicates: - complex_pred = (client.Predicates("category", "==", "Electronics") & client.Predicates("price", "<", 1000)) | \ - (client.Predicates("category", "==", "Books") & client.Predicates("rating", ">=", 4.5)) - vector_store.semantic_search("High-quality products", predicates=complex_pred) - Time-based filtering: Search with time range: vector_store.semantic_search("Recent updates", time_range=(datetime(2024, 1, 1), datetime(2024, 1, 31))) """ query_embedding = self.get_embedding(query) - search_args = { - "limit": limit, - } - - if metadata_filter: - search_args["filter"] = metadata_filter - if predicates: - search_args["predicates"] = predicates + raise ValueError("Predicate objects are not supported; use metadata_filter instead.") + conditions: List[str] = [] + params: List[Any] = [self._vector_literal(query_embedding)] + if metadata_filter: + filters = metadata_filter if isinstance(metadata_filter, list) else [metadata_filter] + conditions.append( + "(" + " OR ".join(["metadata @> %s::jsonb"] * len(filters)) + ")" + ) + params.extend(json.dumps(item) for item in filters) if time_range: - start_date, end_date = time_range - search_args["uuid_time_filter"] = client.UUIDTimeRange(start_date, end_date) - - results = self.vec_client.search(query_embedding, **search_args) + conditions.append("created_at BETWEEN %s AND %s") + params.extend(time_range) + + where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else "" + params.extend([self._vector_literal(query_embedding), limit]) + sql = f""" + SELECT id, metadata, contents, embedding, + embedding <=> %s::vector AS distance + FROM {self.table_name} + {where_clause} + ORDER BY embedding <=> %s::vector + LIMIT %s + """ + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + cursor.execute(sql, params) + results = cursor.fetchall() if return_dataframe: return self._create_dataframe_from_results(results) else: return results + @staticmethod + def _vector_literal(vector: Any) -> str: + return "[" + ",".join(str(float(value)) for value in vector) + "]" + def _create_dataframe_from_results( self, results: List[Tuple[Any, ...]], @@ -335,18 +383,24 @@ def delete( ) if delete_all: - self.vec_client.delete_all() + query = f"DELETE FROM {self.table_name}" + params: Tuple[Any, ...] = () logging.info(f"Deleted all records from {self.vector_settings.table_name}") elif ids: - self.vec_client.delete_by_ids(ids) + query = f"DELETE FROM {self.table_name} WHERE id::text = ANY(%s)" + params = (ids,) logging.info( f"Deleted {len(ids)} records from {self.vector_settings.table_name}" ) elif metadata_filter: - self.vec_client.delete_by_metadata(metadata_filter) + query = f"DELETE FROM {self.table_name} WHERE metadata @> %s::jsonb" + params = (json.dumps(metadata_filter),) logging.info( f"Deleted records matching metadata filter from {self.vector_settings.table_name}" ) + with psycopg2.connect(self.database_url) as conn: + with conn.cursor() as cursor: + cursor.execute(query, params) def keyword_search( self, query: str, limit: int = 5, return_dataframe: bool = True diff --git a/app/start.sh b/app/start.sh index 80970e9..193bea4 100755 --- a/app/start.sh +++ b/app/start.sh @@ -4,4 +4,4 @@ if [ "${DOCULENS_RUN_MIGRATIONS:-false}" = "true" ]; then alembic -c /workspace/app/alembic.ini upgrade head fi -exec uvicorn app.main:app --host 0.0.0.0 --port 8080 --proxy-headers +exec uvicorn app.main:app --host 0.0.0.0 --port "${PORT:-8080}" --proxy-headers diff --git a/app/tasks/tasks.py b/app/tasks/tasks.py index 266a7f3..0174f30 100644 --- a/app/tasks/tasks.py +++ b/app/tasks/tasks.py @@ -11,7 +11,6 @@ from app.database.event import Event from app.database.models import DocumentClassificationHistory from app.database.repository import GenericRepository -from app.pipelines.registry import PipelineRegistry from app.services.classification_audit import record_classification_result from app.services.classification_service import get_classification_service from app.services.label_service import LabelService @@ -49,6 +48,8 @@ def process_incoming_event(event_id: str): raise ValueError(f"Event with id {event_id} not found") event = event_schema_adapter.validate_python(db_event.data) + from app.pipelines.registry import PipelineRegistry + pipeline = PipelineRegistry.get_pipeline(event) task_context = pipeline.run(event) @@ -101,8 +102,12 @@ def _schedule_post_ingestion_jobs(session, task_context) -> None: summary_event = Event(data=payload) session.add(summary_event) session.commit() - celery_app.send_task("process_incoming_event", args=[str(summary_event.id)]) - logger.info("Queued automatic summary event %s for document %s", summary_event.id, document_id) + if settings.task_mode == "local": + process_incoming_event.run(str(summary_event.id)) + logger.info("Processed automatic summary event %s locally", summary_event.id) + else: + celery_app.send_task("process_incoming_event", args=[str(summary_event.id)]) + logger.info("Queued automatic summary event %s for document %s", summary_event.id, document_id) def _auto_classify_from_summary(session, task_context) -> None: diff --git a/app/utils/insert_vectors.py b/app/utils/insert_vectors.py index b788742..152d517 100644 --- a/app/utils/insert_vectors.py +++ b/app/utils/insert_vectors.py @@ -6,10 +6,10 @@ import json # noqa: E402 from datetime import datetime # noqa: E402 +from uuid import uuid1 # noqa: E402 import pandas as pd # noqa: E402 from app.services.vector_store import VectorStore # noqa: E402 -from timescale_vector.client import uuid_from_time # noqa: E402 # Initialize VectorStore vec = VectorStore(local=True) @@ -33,28 +33,13 @@ def load_data(): def prepare_record(row): """Prepare a record for insertion into the vector store. - This function creates a record with a UUID version 1 as the ID, which captures - the current time or a specified time. - - Note: - - By default, this function uses the current time for the UUID. - - To use a specific time: - 1. Import the datetime module. - 2. Create a datetime object for your desired time. - 3. Use uuid_from_time(your_datetime) instead of uuid_from_time(datetime.now()). - - Example: - from datetime import datetime - specific_time = datetime(2023, 1, 1, 12, 0, 0) - id = str(uuid_from_time(specific_time)) - - This is useful when your content already has an associated datetime. + This function creates a time-ordered UUID version 1 identifier. """ content = f"Question: {row['question']}\nAnswer: {row['answer']}" embedding = vec.get_embedding(content) return pd.Series( { - "id": str(uuid_from_time(datetime.now())), + "id": str(uuid1()), "metadata": { "category": row["category"], "created_at": datetime.now().isoformat(), @@ -72,5 +57,5 @@ def prepare_record(row): # Create tables and insert data vec.create_tables() -vec.create_index() # DiskAnnIndex +vec.create_index() # pgvector HNSW vec.upsert(records_df) diff --git a/docker/Dockerfile.celery b/docker/Dockerfile.celery index 85ce31a..d10e1f9 100644 --- a/docker/Dockerfile.celery +++ b/docker/Dockerfile.celery @@ -23,7 +23,7 @@ WORKDIR /workspace COPY pyproject.toml README.md ./ COPY app ./app -RUN pip install --no-cache-dir . +RUN pip install --no-cache-dir '.[ocr]' RUN chown -R celery:celery /workspace diff --git a/docker/Dockerfile.railway b/docker/Dockerfile.railway new file mode 100644 index 0000000..0b1f8e1 --- /dev/null +++ b/docker/Dockerfile.railway @@ -0,0 +1,36 @@ +FROM node:20-alpine AS frontend + +WORKDIR /workspace/frontend +COPY frontend/package.json frontend/package-lock.json ./ +RUN npm ci +COPY frontend/ ./ +RUN npm run build + +FROM python:3.12.8-slim-bookworm AS runtime + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + build-essential \ + curl \ + gcc \ + libpq-dev \ + python3-dev \ + && rm -rf /var/lib/apt/lists/* + +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 +ENV DOCULENS_SERVE_FRONTEND=true +ENV DOCULENS_FRONTEND_DIST_PATH=/workspace/frontend_dist + +WORKDIR /workspace + +COPY pyproject.toml README.md ./ +COPY app ./app +COPY --from=frontend /workspace/frontend/dist ./frontend_dist + +RUN pip install --no-cache-dir . +RUN chmod +x /workspace/app/start.sh + +EXPOSE 8080 + +CMD ["/workspace/app/start.sh"] diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 1063918..67a47ca 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -27,6 +27,10 @@ services: restart: unless-stopped env_file: - ../.env + environment: + DATABASE_HOST: database + REDIS_HOST: redis + REDIS_URL: redis://redis:6379/0 healthcheck: test: ["CMD", "curl", "--fail", "http://localhost:8080/health/live"] interval: 10s @@ -61,8 +65,11 @@ services: env_file: - ../.env environment: - - EASYOCR_HOME=/home/celery/.EasyOCR - - DOCLING_CACHE_DIR=/home/celery/.cache/docling + DATABASE_HOST: database + REDIS_HOST: redis + REDIS_URL: redis://redis:6379/0 + EASYOCR_HOME: /home/celery/.EasyOCR + DOCLING_CACHE_DIR: /home/celery/.cache/docling volumes: - ingestion_data:/workspace/data/ingestion - easyocr_cache:/home/celery/.EasyOCR diff --git a/docs/deploy-railway.md b/docs/deploy-railway.md new file mode 100644 index 0000000..b5a8d27 --- /dev/null +++ b/docs/deploy-railway.md @@ -0,0 +1,257 @@ +# Deploy DocuLens on Railway + +DocuLens supports two Railway profiles from the same codebase: + +| Profile | Railway resources | Best for | Limitation | +| --- | ---: | --- | --- | +| Free portfolio | 1 web service | recruiter demos and occasional personal use | in-process jobs; heavy scanned-PDF OCR may exceed the Free memory limit | +| Full Railway | 5 project resources | durable asynchronous processing | realistically requires the Hobby plan or higher | + +The existing Cloudflare Pages showcase remains a separate, always-available, +read-only product tour. Deploying either profile does not replace it. + +## What the repository now provides + +- `docker/Dockerfile.railway`: builds React and FastAPI into one same-origin image +- `railway.toml`: web build, migrations, health check, and restart policy +- `railway.worker.toml`: optional Celery worker configuration +- an optional `ocr` dependency extra, installed only in the worker image +- `DATABASE_URL` support for Railway or external managed PostgreSQL +- portable pgvector + HNSW retrieval without TimescaleDB-specific extensions +- S3-compatible document storage for Supabase, Railway Buckets, R2, or S3 +- Supabase Auth with passwordless email and optional Google OAuth +- secure HTTP-only application sessions +- a bounded LangGraph investigation agent with PostgreSQL checkpoints +- streaming execution events, an evidence ledger, and citation validation +- `DOCULENS_TASK_MODE=local|celery` to switch deployment profiles + +## Profile A: Railway Free + +This is the recommended starting point. Railway runs one sleeping web service; +Supabase provides free PostgreSQL/pgvector and private document storage. + +```mermaid +flowchart LR + Browser --> Web["Railway: React + FastAPI"] + Web --> Auth["Supabase Auth: email link + Google"] + Web --> Agent["LangGraph investigation agent"] + Agent --> Supabase[("Supabase PostgreSQL + pgvector + checkpoints")] + Web --> Storage[("Supabase Storage")] + Web --> Providers["OpenAI / Anthropic"] + Web --> LocalJobs["In-process background jobs"] +``` + +### 1. Create Supabase data services + +1. Create a Supabase project. +2. In its SQL editor, run `CREATE EXTENSION IF NOT EXISTS vector;`. +3. Copy its pooled PostgreSQL connection string as `DATABASE_URL`. +4. Create a private Storage bucket such as `doculens-documents`. +5. Enable the Storage S3 protocol and generate server-side S3 access keys. +6. Copy the direct S3 endpoint, project region, Access Key ID, and Secret Access Key. + +The application runs Alembic before each Railway deployment and creates the +portable vector table and HNSW index idempotently during startup. + +The bucket must stay private. Supabase's S3 access keys bypass row-level +security and therefore belong only in Railway's server-side secret variables. + +### 2. Create the Railway service + +1. Create an empty Railway project. +2. Add a service from `codewithmoin/doculens-ai`. +3. Select the branch containing the Railway deployment work. +4. Railway will read `railway.toml` and build `docker/Dockerfile.railway`. +5. Generate a public Railway domain. +6. Enable **Settings → Deploy → Serverless** so the service sleeps while idle. +7. Set a hard usage limit in the Railway workspace usage settings. + +Use these variables: + +```dotenv +DOCULENS_ENVIRONMENT=production +DOCULENS_LOG_LEVEL=INFO +DOCULENS_INITIALIZE_DATABASE=true +DOCULENS_SEED_DEMO_USERS=false +DOCULENS_SEED_DEMO_WORKSPACE=false +DOCULENS_SHOWCASE_READ_ONLY=false +DOCULENS_REQUIRE_AUTH=true +DOCULENS_AUTH_SECRET= +DOCULENS_AUTH_COOKIE_SECURE=true +DOCULENS_TASK_MODE=local +DOCULENS_EXTRACTION_BACKEND=lightweight +DOCULENS_CORS_ORIGINS=["https://YOUR-SERVICE.up.railway.app"] +DOCULENS_AGENT_ENABLED=true +DOCULENS_AGENT_PROVIDER=openai +DOCULENS_AGENT_MODEL= +DOCULENS_AGENT_CHECKPOINT_BACKEND=postgres +DOCULENS_AGENT_MAX_STEPS=10 +DOCULENS_AGENT_MAX_EVIDENCE=24 +DOCULENS_AGENT_SEARCH_LIMIT=6 + +DATABASE_URL= + +DOCULENS_STORAGE_BACKEND=s3 +DOCULENS_S3_ENDPOINT_URL=https://.storage.supabase.co/storage/v1/s3 +DOCULENS_S3_BUCKET_NAME=doculens-documents +DOCULENS_S3_ACCESS_KEY_ID= +DOCULENS_S3_SECRET_ACCESS_KEY= +DOCULENS_S3_REGION= +DOCULENS_S3_FORCE_PATH_STYLE=true + +OPENAI_API_KEY= +ANTHROPIC_API_KEY= +OPEN_ROUTER_API_KEY= + +SUPABASE_URL=https://.supabase.co +SUPABASE_PUBLISHABLE_KEY= +DOCULENS_SUPABASE_ALLOWED_EMAILS=["your-email@example.com"] +DOCULENS_SUPABASE_ALLOWED_DOMAINS=[] +DOCULENS_ALLOW_PUBLIC_SUPABASE_SIGNIN=false +DOCULENS_SUPABASE_GOOGLE_ENABLED=true +``` + +`DOCULENS_TASK_MODE=local` means the API returns `202 Accepted` and processes the +job in a Starlette background task inside the same container. It avoids paying +for Redis and a worker, but an unexpected container restart can interrupt an +in-flight job. The event remains stored and can be resubmitted. + +`DOCULENS_EXTRACTION_BACKEND=lightweight` extracts embedded text from PDFs and +keeps page-aware citations without loading an OCR model. Scanned/image-only PDFs +return an explicit error and require the full Docling worker profile. + +`DOCULENS_AGENT_CHECKPOINT_BACKEND=postgres` is required for writable production +deployments. LangGraph stores each completed graph step in the same Supabase +PostgreSQL database, so a run can be inspected or resumed after a process +restart. Use Supabase's session-compatible pooled connection string for +`DATABASE_URL`. + +The Free profile executes an investigation through its active SSE request. +Checkpoints protect completed steps, but unattended execution after the browser +disconnects requires the optional worker profile. + +### 3. Configure Supabase Auth + +1. In **Supabase → Project Settings → API**, copy the Project URL and + publishable key into Railway as `SUPABASE_URL` and + `SUPABASE_PUBLISHABLE_KEY`. The legacy `anon` key also works. +2. In **Supabase → Authentication → URL Configuration**, set the Site URL to + `https://YOUR-SERVICE.up.railway.app`. +3. Add these Redirect URLs: + `https://YOUR-SERVICE.up.railway.app/login` and + `http://localhost:5173/login`. +4. Email authentication is enabled by default. DocuLens sends a passwordless + sign-in link and exchanges the resulting Supabase session for its own secure, + HTTP-only application session. +5. Keep the deployment invite-only with + `DOCULENS_SUPABASE_ALLOWED_EMAILS=["your-email@example.com"]`. + +To enable **Continue with Google**: + +1. Create a Google Cloud OAuth client of type **Web application**. +2. Add exactly this Authorized redirect URI: + `https://.supabase.co/auth/v1/callback`. +3. In **Supabase → Authentication → Providers → Google**, enable Google and + paste the Google Client ID and Client Secret. +4. Set `DOCULENS_SUPABASE_GOOGLE_ENABLED=true` in Railway. +5. While the Google app remains in **Testing**, add each allowed Google account + under **Google Auth Platform → Audience → Test users**. Publish the app only + when you intentionally want non-test accounts to use Google sign-in. + +The Google client secret belongs in Supabase, not Railway or Git. Supabase +verifies the provider identity; FastAPI validates the Supabase access token, +applies the DocuLens allowlist, provisions the local profile, and issues the +DocuLens session cookie. + +### 4. Verify + +```bash +curl --fail https://YOUR-SERVICE.up.railway.app/health/live +curl --fail https://YOUR-SERVICE.up.railway.app/health/ready +``` + +Then verify: + +1. `/` renders the landing page. +2. `/login` displays email-link sign-in and, when enabled, Google sign-in. +3. Both enabled sign-in methods reach `/app`. +4. A small text PDF uploads and progresses to Ready. +5. Search returns the document. +6. Ask DocuLens returns an answer with citations. +7. **Investigate** streams tool activity, builds an evidence ledger, and returns + a report with verified citation coverage. +8. Refreshing the page keeps the completed investigation in Recent investigations. +9. Logging out invalidates the browser session. + +## Profile B: everything on Railway + +Use this when the system must keep durable Celery semantics and all +infrastructure must live in one Railway project. + +```mermaid +flowchart LR + Browser --> Web["Railway web"] + Web --> DB[("Railway PostgreSQL + pgvector")] + Web --> Redis[("Railway Redis")] + Redis --> Worker["Railway Celery worker"] + Web --> Bucket[("Railway Bucket")] + Worker --> DB + Worker --> Bucket +``` + +Create exactly five resources: + +1. **Web** — GitHub repository using `/railway.toml` +2. **Worker** — same repository, config file `/railway.worker.toml` +3. **PostgreSQL with pgvector** — Railway's pgvector template +4. **Redis** — Railway managed Redis +5. **Bucket** — Railway private S3-compatible bucket + +Reference the managed variables in both Web and Worker: + +```dotenv +DATABASE_URL=${{Postgres.DATABASE_URL}} +REDIS_URL=${{Redis.REDIS_URL}} +DOCULENS_TASK_MODE=celery +DOCULENS_EXTRACTION_BACKEND=docling +DOCULENS_AGENT_ENABLED=true +DOCULENS_AGENT_CHECKPOINT_BACKEND=postgres +DOCULENS_STORAGE_BACKEND=s3 +DOCULENS_S3_ENDPOINT_URL= +DOCULENS_S3_BUCKET_NAME= +DOCULENS_S3_ACCESS_KEY_ID= +DOCULENS_S3_SECRET_ACCESS_KEY= +DOCULENS_S3_REGION=auto +DOCULENS_S3_FORCE_PATH_STYLE=false +``` + +Give the Web service the authentication variables and public domain. Give the +Worker the provider keys, database, Redis, and storage variables. The Worker +does not need a public domain. + +This profile keeps late acknowledgement, worker-loss retries, task timeouts, +and durable broker queues. The cost tradeoff is real: PostgreSQL and Redis are +continuously allocated services and will consume more than the Free credit. + +## Cost and operational notes + +- Railway Free currently includes $1 of monthly usage and limits a service to + 0.5 GB RAM. Enable Serverless and a hard spend cap. +- Docling OCR is the heaviest part of the image and runtime. Small, text-based + PDFs are appropriate for the Free demo. Scanned documents should use the full + worker profile with more memory. +- Supabase Auth and Google Identity have free entry tiers. OpenAI/Anthropic requests are separate, + usage-based costs. +- Never put provider, database, storage, or signing secrets in Vite variables or + commit them to Git. +- Keep the Cloudflare static showcase linked from GitHub as the reliable fallback + if the Railway service is asleep or its monthly credit is exhausted. + +## Rollback + +Railway keeps deployment history. If a release fails: + +1. Roll the Web service back to the previous healthy image. +2. Do not downgrade the database automatically. +3. Inspect Alembic history before applying any schema downgrade. +4. Keep the Supabase/Railway bucket unchanged; uploads are independent of app images. diff --git a/frontend/package-lock.json b/frontend/package-lock.json index aac9eb7..c0c04c3 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -9,11 +9,11 @@ "version": "0.0.0", "dependencies": { "@radix-ui/react-slot": "^1.1.0", + "@supabase/supabase-js": "^2.112.2", "@tanstack/react-query": "^5.66.2", "@tanstack/react-query-devtools": "^5.66.2", "class-variance-authority": "^0.7.0", "classnames": "^2.5.1", - "framer-motion": "^12.23.24", "lucide-react": "^0.475.0", "react": "^19.1.1", "react-dom": "^19.1.1", @@ -1370,6 +1370,98 @@ "win32" ] }, + "node_modules/@supabase/auth-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/auth-js/-/auth-js-2.112.2.tgz", + "integrity": "sha512-l1InCp4j98d09LZ6+RgubgF4eVPGBGXcLEhFusLg1qUCHJ2IEkYu5FohKK+eaFmIOwEk0kqG/j/lycw5e15mcQ==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/functions-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/functions-js/-/functions-js-2.112.2.tgz", + "integrity": "sha512-oMuSWN0ERmrG9S6kOM0bwhHmESGVl3kMtkZl2dNCU/r89hMiziX4GfD1omNo9QcBDele4N0GwSZ7hdbpuiA35A==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/phoenix": { + "version": "0.4.5", + "resolved": "https://registry.npmjs.org/@supabase/phoenix/-/phoenix-0.4.5.tgz", + "integrity": "sha512-aAn9H9ovVyeApKy11OWOrrOGq8DV68yWeH4ud2lN9fzn4aO8Zb5GLL9m1pUg9nLqIcT+ZDfAcsZe0E/nqdv2lw==", + "license": "MIT" + }, + "node_modules/@supabase/postgrest-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/postgrest-js/-/postgrest-js-2.112.2.tgz", + "integrity": "sha512-ewhhtRny/HFRGhUTTg/PsqIatsl8OhW8Eha/Tz4S+SRAXBnuhKei9ZpsQTgL/3XcH9UEwuPQyQgQ9itq7nRQeg==", + "license": "MIT", + "dependencies": { + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/realtime-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/realtime-js/-/realtime-js-2.112.2.tgz", + "integrity": "sha512-cd9/CEUJ6Go13FxtfiuC5rYELJtuQzVzTXlGG+XjSppjDS+anq+xo++WQe7ZRUNTuHOCeyKRwmx9Hw/OQJ04ig==", + "license": "MIT", + "dependencies": { + "@supabase/phoenix": "0.4.5", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/storage-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/storage-js/-/storage-js-2.112.2.tgz", + "integrity": "sha512-6jyBq/J1iXOHNpbjCZS7gFcDk49iM1MCJUVkDl71gLd/+XnLDzpUBs8icGebtwiHpl4kVszxIRDYAosbF4Rsig==", + "license": "MIT", + "dependencies": { + "iceberg-js": "^0.8.1", + "tslib": "2.8.1" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@supabase/supabase-js": { + "version": "2.112.2", + "resolved": "https://registry.npmjs.org/@supabase/supabase-js/-/supabase-js-2.112.2.tgz", + "integrity": "sha512-UyI1epU9B4X51HvNpkmlwTdF20fEcz2vyvrcDKVzFN4jZN41f5iQRqsiIQjAY5OVJD6ljqA/1g9JQeOTvFHpkA==", + "license": "MIT", + "dependencies": { + "@supabase/auth-js": "2.112.2", + "@supabase/functions-js": "2.112.2", + "@supabase/postgrest-js": "2.112.2", + "@supabase/realtime-js": "2.112.2", + "@supabase/storage-js": "2.112.2" + }, + "engines": { + "node": ">=22.0.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.0.0" + }, + "peerDependenciesMeta": { + "@opentelemetry/api": { + "optional": true + } + } + }, "node_modules/@tailwindcss/forms": { "version": "0.5.10", "resolved": "https://registry.npmjs.org/@tailwindcss/forms/-/forms-0.5.10.tgz", @@ -2860,33 +2952,6 @@ "url": "https://github.com/sponsors/rawify" } }, - "node_modules/framer-motion": { - "version": "12.23.24", - "resolved": "https://registry.npmjs.org/framer-motion/-/framer-motion-12.23.24.tgz", - "integrity": "sha512-HMi5HRoRCTou+3fb3h9oTLyJGBxHfW+HnNE25tAXOvVx/IvwMHK0cx7IR4a2ZU6sh3IX1Z+4ts32PcYBOqka8w==", - "license": "MIT", - "dependencies": { - "motion-dom": "^12.23.23", - "motion-utils": "^12.23.6", - "tslib": "^2.4.0" - }, - "peerDependencies": { - "@emotion/is-prop-valid": "*", - "react": "^18.0.0 || ^19.0.0", - "react-dom": "^18.0.0 || ^19.0.0" - }, - "peerDependenciesMeta": { - "@emotion/is-prop-valid": { - "optional": true - }, - "react": { - "optional": true - }, - "react-dom": { - "optional": true - } - } - }, "node_modules/fsevents": { "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", @@ -3019,6 +3084,15 @@ "node": ">= 0.4" } }, + "node_modules/iceberg-js": { + "version": "0.8.1", + "resolved": "https://registry.npmjs.org/iceberg-js/-/iceberg-js-0.8.1.tgz", + "integrity": "sha512-1dhVQZXhcHje7798IVM+xoo/1ZdVfzOMIc8/rgVSijRK38EDqOJoGula9N/8ZI5RD8QTxNQtK/Gozpr+qUqRRA==", + "license": "MIT", + "engines": { + "node": ">=20.0.0" + } + }, "node_modules/ignore": { "version": "5.3.2", "resolved": "https://registry.npmjs.org/ignore/-/ignore-5.3.2.tgz", @@ -3375,21 +3449,6 @@ "node": ">=16 || 14 >=14.17" } }, - "node_modules/motion-dom": { - "version": "12.23.23", - "resolved": "https://registry.npmjs.org/motion-dom/-/motion-dom-12.23.23.tgz", - "integrity": "sha512-n5yolOs0TQQBRUFImrRfs/+6X4p3Q4n1dUEqt/H58Vx7OW6RF+foWEgmTVDhIWJIMXOuNNL0apKH2S16en9eiA==", - "license": "MIT", - "dependencies": { - "motion-utils": "^12.23.6" - } - }, - "node_modules/motion-utils": { - "version": "12.23.6", - "resolved": "https://registry.npmjs.org/motion-utils/-/motion-utils-12.23.6.tgz", - "integrity": "sha512-eAWoPgr4eFEOFfg2WjIsMoqJTW6Z8MTUCgn/GZ3VRpClWBdnbjryiA3ZSNLyxCTmCQx4RmYX6jX1iWHbenUPNQ==", - "license": "MIT" - }, "node_modules/ms": { "version": "2.1.3", "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", diff --git a/frontend/package.json b/frontend/package.json index 24a0cbf..3e414c0 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -11,6 +11,7 @@ }, "dependencies": { "@radix-ui/react-slot": "^1.1.0", + "@supabase/supabase-js": "^2.112.2", "@tanstack/react-query": "^5.66.2", "@tanstack/react-query-devtools": "^5.66.2", "class-variance-authority": "^0.7.0", diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 0d4d6fa..b11ff73 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -15,6 +15,9 @@ import type { LabelRequestPayload, LabelResponse, LabelsResponse, + InvestigationDetail, + InvestigationStreamEvent, + InvestigationSummary, RuntimeConfig, SearchHistoryEntry, UploadResponse, @@ -105,22 +108,29 @@ async function handleResponse(response: Response): Promise { return (await response.json()) as T; } +function apiFetch(input: RequestInfo | URL, init?: RequestInit): Promise { + return fetch(input, { + ...init, + credentials: 'include', + }); +} + export async function fetchDocuments(limit = 20): Promise { - const response = await fetch(resolveUrl('/events/documents', { limit }), { + const response = await apiFetch(resolveUrl('/events/documents', { limit }), { headers: buildHeaders(), }); return handleResponse(response); } export async function fetchEvents(limit = 20): Promise { - const response = await fetch(resolveUrl('/events', { limit }), { + const response = await apiFetch(resolveUrl('/events', { limit }), { headers: buildHeaders(), }); return handleResponse(response); } export async function archiveDocument(documentId: string, reason?: string): Promise { - const response = await fetch(resolveUrl(`/events/documents/${documentId}/archive`), { + const response = await apiFetch(resolveUrl(`/events/documents/${documentId}/archive`), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(reason ? { reason } : {}), @@ -132,7 +142,7 @@ export async function deleteDocument( documentId: string, options?: { reason?: string; purgeVectors?: boolean }, ): Promise { - const response = await fetch( + const response = await apiFetch( resolveUrl(`/events/documents/${documentId}`, { reason: options?.reason, purge_vectors: options?.purgeVectors ?? true, @@ -146,7 +156,7 @@ export async function deleteDocument( } export async function restoreDocument(documentId: string, reason?: string): Promise { - const response = await fetch(resolveUrl(`/events/documents/${documentId}/restore`), { + const response = await apiFetch(resolveUrl(`/events/documents/${documentId}/restore`), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(reason ? { reason } : {}), @@ -158,7 +168,7 @@ export async function fetchDocumentChunks( documentId: string, limit: number, ): Promise { - const response = await fetch( + const response = await apiFetch( resolveUrl(`/events/documents/${documentId}/chunks`, { limit }), { headers: buildHeaders(), @@ -168,35 +178,90 @@ export async function fetchDocumentChunks( } export async function fetchQaHistory(limit = 20): Promise { - const response = await fetch(resolveUrl('/events/qa/history', { limit }), { + const response = await apiFetch(resolveUrl('/events/qa/history', { limit }), { headers: buildHeaders(), }); return handleResponse(response); } export async function fetchSearchHistory(limit = 20): Promise { - const response = await fetch(resolveUrl('/events/search/history', { limit }), { + const response = await apiFetch(resolveUrl('/events/search/history', { limit }), { headers: buildHeaders(), }); return handleResponse(response); } export async function fetchRuntimeConfig(): Promise { - const response = await fetch(resolveUrl('/events/config'), { + const response = await apiFetch(resolveUrl('/events/config'), { headers: buildHeaders(), }); return handleResponse(response); } export async function fetchDashboardInsights(): Promise { - const response = await fetch(resolveUrl('/events/insights/dashboard'), { + const response = await apiFetch(resolveUrl('/events/insights/dashboard'), { headers: buildHeaders(), }); return handleResponse(response); } +export async function fetchInvestigations(limit = 20): Promise { + const response = await apiFetch(resolveUrl('/investigations', { limit }), { + headers: buildHeaders(), + }); + return handleResponse(response); +} + +export async function fetchInvestigation(id: string): Promise { + const response = await apiFetch(resolveUrl(`/investigations/${id}`), { + headers: buildHeaders(), + }); + return handleResponse(response); +} + +export async function streamInvestigation( + payload: { goal: string; document_ids?: string[] }, + onEvent: (event: InvestigationStreamEvent) => void, + signal?: AbortSignal, +): Promise { + const response = await apiFetch(resolveUrl('/investigations/stream'), { + method: 'POST', + headers: buildHeaders({ Accept: 'text/event-stream' }, true), + body: JSON.stringify(payload), + signal, + }); + if (!response.ok) { + await handleResponse(response); + } + if (!response.body) { + throw new Error('The server did not provide an investigation stream.'); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + while (true) { + const { done, value } = await reader.read(); + buffer += decoder.decode(value, { stream: !done }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + for (const frame of frames) { + const eventLine = frame.split(/\r?\n/).find((line) => line.startsWith('event:')); + const dataLines = frame + .split(/\r?\n/) + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice(5).trimStart()); + if (!eventLine || !dataLines.length) continue; + const type = eventLine.slice(6).trim(); + const event = { type, payload: JSON.parse(dataLines.join('\n')) } as InvestigationStreamEvent; + onEvent(event); + } + if (done) break; + } +} + export async function postEvent(payload: Record): Promise { - const response = await fetch(resolveUrl('/events'), { + const response = await apiFetch(resolveUrl('/events'), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(payload), @@ -218,7 +283,7 @@ export async function uploadDocument(options: { form.append('metadata', JSON.stringify(options.metadata)); } - const response = await fetch(resolveUrl('/events/documents/upload'), { + const response = await apiFetch(resolveUrl('/events/documents/upload'), { method: 'POST', headers: buildHeaders(undefined, false), body: form, @@ -227,7 +292,7 @@ export async function uploadDocument(options: { } export async function login(credentials: { email: string; password: string }): Promise { - const response = await fetch(resolveUrl('/auth/login'), { + const response = await apiFetch(resolveUrl('/auth/login'), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(credentials), @@ -235,8 +300,25 @@ export async function login(credentials: { email: string; password: string }): P return handleResponse(response); } +export async function loginWithSupabase(accessToken: string): Promise { + const response = await apiFetch(resolveUrl('/auth/supabase'), { + method: 'POST', + headers: buildHeaders(undefined, true), + body: JSON.stringify({ access_token: accessToken }), + }); + return handleResponse(response); +} + +export async function logout(): Promise { + const response = await apiFetch(resolveUrl('/auth/logout'), { + method: 'POST', + headers: buildHeaders(), + }); + await handleResponse(response); +} + export async function fetchProfile(): Promise { - const response = await fetch(resolveUrl('/auth/me'), { + const response = await apiFetch(resolveUrl('/auth/me'), { headers: buildHeaders(), }); return handleResponse(response); @@ -246,7 +328,7 @@ export async function classifyDocument( documentId: string, payload: DocumentClassificationRequest = {}, ): Promise { - const response = await fetch(resolveUrl(`/events/documents/${documentId}/classify`), { + const response = await apiFetch(resolveUrl(`/events/documents/${documentId}/classify`), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(payload), @@ -255,14 +337,14 @@ export async function classifyDocument( } export async function fetchLabels(): Promise { - const response = await fetch(resolveUrl('/events/labels'), { + const response = await apiFetch(resolveUrl('/events/labels'), { headers: buildHeaders(), }); return handleResponse(response); } export async function createLabel(payload: LabelRequestPayload): Promise { - const response = await fetch(resolveUrl('/events/labels'), { + const response = await apiFetch(resolveUrl('/events/labels'), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(payload), @@ -271,7 +353,7 @@ export async function createLabel(payload: LabelRequestPayload): Promise): Promise { - const response = await fetch(resolveUrl(`/events/labels/${labelId}`), { + const response = await apiFetch(resolveUrl(`/events/labels/${labelId}`), { method: 'PATCH', headers: buildHeaders(undefined, true), body: JSON.stringify(payload), @@ -280,7 +362,7 @@ export async function updateLabel(labelId: string, payload: Partial { - const response = await fetch(resolveUrl(`/events/labels/${labelId}`, { force }), { + const response = await apiFetch(resolveUrl(`/events/labels/${labelId}`, { force }), { method: 'DELETE', headers: buildHeaders(), }); @@ -288,7 +370,7 @@ export async function deleteLabel(labelId: string, force = false): Promise } export async function fetchClassificationHistory(documentId: string): Promise { - const response = await fetch(resolveUrl(`/events/documents/${documentId}/classification-history`), { + const response = await apiFetch(resolveUrl(`/events/documents/${documentId}/classification-history`), { headers: buildHeaders(), }); return handleResponse(response); @@ -298,7 +380,7 @@ export async function overrideClassification( documentId: string, payload: ClassificationOverrideRequest, ): Promise { - const response = await fetch(resolveUrl(`/events/documents/${documentId}/classification-history`), { + const response = await apiFetch(resolveUrl(`/events/documents/${documentId}/classification-history`), { method: 'POST', headers: buildHeaders(undefined, true), body: JSON.stringify(payload), diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 6012e03..40060a9 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -107,6 +107,8 @@ export interface UserProfile { persona: string; role: string; access_level: string; + avatar_url?: string | null; + auth_provider: string; } export interface AuthResponse { @@ -210,12 +212,94 @@ export interface RuntimeConfig { search_preview_limit: number; chunk_preview_limit: number; auth_required: boolean; + session_auth_required: boolean; + supabase_url?: string | null; + supabase_publishable_key?: string | null; + supabase_google_enabled?: boolean; showcase_read_only: boolean; api_key_header: string; persona_options?: string[]; role_definitions?: Record; + agent_enabled: boolean; + agent_max_steps: number; + agent_max_evidence: number; } +export interface InvestigationActivity { + activity_id: string; + kind: 'planning' | 'tool' | 'evidence' | 'verification' | 'clarification' | 'report' | 'system'; + title: string; + detail?: string | null; + status: 'running' | 'completed' | 'warning' | 'failed'; + tool?: string | null; + created_at: string; +} + +export interface InvestigationEvidence { + evidence_id: string; + reference: string; + document_id: string; + filename: string; + text: string; + chunk_id?: string | null; + chunk_index?: number | null; + page_number?: number | null; + relevance_score?: number | null; + source_query?: string | null; +} + +export interface InvestigationFinding { + claim: string; + significance: 'low' | 'medium' | 'high' | 'critical'; + explanation: string; + citation_refs: string[]; +} + +export interface InvestigationReport { + title: string; + executive_summary: string; + findings: InvestigationFinding[]; + recommended_actions: string[]; + limitations: string[]; +} + +export interface InvestigationVerification { + valid: boolean; + citation_coverage: number; + unsupported_claims: string[]; + unknown_citations: string[]; +} + +export interface InvestigationSummary { + id: string; + goal: string; + status: 'queued' | 'running' | 'needs_input' | 'completed' | 'partial' | 'failed'; + created_at: string; + updated_at: string; + evidence_count: number; + step_count: number; + report_title?: string | null; +} + +export interface InvestigationDetail extends InvestigationSummary { + state: { + activities: InvestigationActivity[]; + evidence: InvestigationEvidence[]; + report?: InvestigationReport; + verification?: InvestigationVerification; + clarification_question?: string; + error?: string; + }; +} + +export type InvestigationStreamEvent = + | { type: 'investigation'; payload: { id: string; goal: string; status: string; max_steps: number } } + | { type: 'activity'; payload: InvestigationActivity } + | { type: 'evidence'; payload: InvestigationEvidence } + | { type: 'report'; payload: { report: InvestigationReport; verification: InvestigationVerification } } + | { type: 'done'; payload: { id: string; status: InvestigationSummary['status']; step_count: number; evidence_count: number; clarification_question?: string } } + | { type: 'error'; payload: { id: string; message: string } }; + export interface DashboardInsights { total_documents: number; summarised_documents: number; diff --git a/frontend/src/auth/AuthProvider.tsx b/frontend/src/auth/AuthProvider.tsx index 3ce5689..581332a 100644 --- a/frontend/src/auth/AuthProvider.tsx +++ b/frontend/src/auth/AuthProvider.tsx @@ -1,16 +1,22 @@ -import { useCallback, useEffect, useMemo, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { useNavigate } from 'react-router-dom'; -import { clearAuthToken, fetchProfile, login as apiLogin, setAuthToken } from '../api/client'; -import type { RoleDefinition, UserProfile } from '../api/types'; +import { + clearAuthToken, + fetchProfile, + loginWithSupabase as apiLoginWithSupabase, + logout as apiLogout, +} from '../api/client'; +import type { AuthResponse, RoleDefinition, UserProfile } from '../api/types'; +import { useSettings } from '../settings/useSettings'; import { DEFAULT_PERSONAS } from '../settings/types'; import { AuthContext } from './context'; +import { getSupabaseClient } from './supabase'; import type { AuthContextValue } from './types'; const AUTH_STORAGE_KEY = 'doculens.auth'; interface StoredAuthPayload { - token: string; personas?: string[]; roles?: Record; } @@ -20,7 +26,9 @@ export function AuthProvider({ children }: { children: React.ReactNode }) { const [isLoading, setIsLoading] = useState(true); const [personas, setPersonas] = useState(DEFAULT_PERSONAS); const [roles, setRoles] = useState>({}); + const lastExchangedToken = useRef(undefined); const navigate = useNavigate(); + const { serverConfig, isLoaded: isSettingsLoaded } = useSettings(); const persistAuth = useCallback((payload: StoredAuthPayload | null) => { if (!payload) { @@ -30,87 +38,140 @@ export function AuthProvider({ children }: { children: React.ReactNode }) { } }, []); - const performLogout = useCallback( + const applyAuthResponse = useCallback( + (response: AuthResponse) => { + setUser(response.user); + setPersonas(response.personas?.length ? response.personas : DEFAULT_PERSONAS); + setRoles(response.roles ?? {}); + persistAuth({ personas: response.personas, roles: response.roles }); + return response; + }, + [persistAuth], + ); + + const clearLocalSession = useCallback( (redirect = true) => { setUser(null); clearAuthToken(); persistAuth(null); setPersonas(DEFAULT_PERSONAS); setRoles({}); - if (redirect) { - navigate('/login', { replace: true }); - } + lastExchangedToken.current = undefined; + if (redirect) navigate('/login', { replace: true }); }, [navigate, persistAuth], ); useEffect(() => { const raw = localStorage.getItem(AUTH_STORAGE_KEY); - if (!raw) { - setIsLoading(false); - return; - } try { - const parsed = JSON.parse(raw) as StoredAuthPayload; - if (parsed.token) { - setAuthToken(parsed.token); - if (parsed.personas?.length) { - setPersonas(parsed.personas); - } - if (parsed.roles) { - setRoles(parsed.roles); - } - void fetchProfile() - .then((profile) => { - setUser(profile); - }) - .catch(() => { - setUser(null); - clearAuthToken(); - persistAuth(null); - setPersonas(DEFAULT_PERSONAS); - setRoles({}); - }) - .finally(() => setIsLoading(false)); - return; - } + const parsed = raw ? (JSON.parse(raw) as StoredAuthPayload) : undefined; + if (parsed?.personas?.length) setPersonas(parsed.personas); + if (parsed?.roles) setRoles(parsed.roles); } catch { - // ignore parse errors; treat as logged out + persistAuth(null); } - setIsLoading(false); }, [persistAuth]); - const handleLogin = useCallback( - async (email: string, password: string) => { - const response = await apiLogin({ email, password }); - setUser(response.user); - setAuthToken(response.access_token); - if (response.personas?.length) { - setPersonas(response.personas); + useEffect(() => { + if (!isSettingsLoaded) return; + + const supabaseUrl = serverConfig?.supabase_url; + const publishableKey = serverConfig?.supabase_publishable_key; + let cancelled = false; + let unsubscribe: (() => void) | undefined; + + const exchangeToken = async (accessToken: string) => { + if (lastExchangedToken.current === accessToken) return; + lastExchangedToken.current = accessToken; + try { + const response = await apiLoginWithSupabase(accessToken); + if (!cancelled) applyAuthResponse(response); + } catch (error) { + lastExchangedToken.current = undefined; + throw error; } - if (response.roles) { - setRoles(response.roles); + }; + + const initialise = async () => { + try { + const profile = await fetchProfile(); + if (!cancelled) setUser(profile); + } catch { + if (supabaseUrl && publishableKey) { + const supabase = getSupabaseClient(supabaseUrl, publishableKey); + const { data } = await supabase.auth.getSession(); + if (data.session?.access_token) await exchangeToken(data.session.access_token); + } + } finally { + if (!cancelled) setIsLoading(false); } - persistAuth({ - token: response.access_token, - personas: response.personas, - roles: response.roles, + + if (supabaseUrl && publishableKey) { + const supabase = getSupabaseClient(supabaseUrl, publishableKey); + const { data } = supabase.auth.onAuthStateChange((event, session) => { + if ((event === 'SIGNED_IN' || event === 'TOKEN_REFRESHED') && session?.access_token) { + void exchangeToken(session.access_token).catch(() => clearLocalSession(false)); + } + if (event === 'SIGNED_OUT') clearLocalSession(false); + }); + unsubscribe = () => data.subscription.unsubscribe(); + } + }; + + void initialise(); + return () => { + cancelled = true; + unsubscribe?.(); + }; + }, [applyAuthResponse, clearLocalSession, isSettingsLoaded, serverConfig]); + + const requireSupabase = useCallback(() => { + const url = serverConfig?.supabase_url; + const key = serverConfig?.supabase_publishable_key; + if (!url || !key) throw new Error('Supabase authentication is not configured.'); + return getSupabaseClient(url, key); + }, [serverConfig]); + + const requestMagicLink = useCallback( + async (email: string) => { + const { error } = await requireSupabase().auth.signInWithOtp({ + email, + options: { emailRedirectTo: `${window.location.origin}/login` }, }); - return response; + if (error) throw error; }, - [persistAuth], + [requireSupabase], ); + const loginWithGoogle = useCallback(async () => { + const { error } = await requireSupabase().auth.signInWithOAuth({ + provider: 'google', + options: { redirectTo: `${window.location.origin}/login` }, + }); + if (error) throw error; + }, [requireSupabase]); + + const performLogout = useCallback(() => { + const supabaseUrl = serverConfig?.supabase_url; + const publishableKey = serverConfig?.supabase_publishable_key; + const supabaseLogout = supabaseUrl && publishableKey + ? getSupabaseClient(supabaseUrl, publishableKey).auth.signOut() + : Promise.resolve(); + void Promise.allSettled([apiLogout(), supabaseLogout]).finally(() => clearLocalSession(true)); + }, [clearLocalSession, serverConfig]); + const value = useMemo( () => ({ user, isLoading, personas, roles, - login: handleLogin, - logout: () => performLogout(true), + requestMagicLink, + loginWithGoogle, + logout: performLogout, }), - [user, isLoading, personas, roles, handleLogin, performLogout], + [user, isLoading, personas, roles, requestMagicLink, loginWithGoogle, performLogout], ); return {children}; diff --git a/frontend/src/auth/ProtectedRoute.tsx b/frontend/src/auth/ProtectedRoute.tsx index a161c81..d7f2b78 100644 --- a/frontend/src/auth/ProtectedRoute.tsx +++ b/frontend/src/auth/ProtectedRoute.tsx @@ -21,7 +21,7 @@ export function ProtectedRoute({ children }: { children: ReactNode }) { ); } - if (serverConfig?.showcase_read_only) { + if (serverConfig?.showcase_read_only || !serverConfig?.session_auth_required) { return <>{children}; } diff --git a/frontend/src/auth/supabase.ts b/frontend/src/auth/supabase.ts new file mode 100644 index 0000000..6838fb1 --- /dev/null +++ b/frontend/src/auth/supabase.ts @@ -0,0 +1,19 @@ +import { createClient, type SupabaseClient } from '@supabase/supabase-js'; + +let client: SupabaseClient | undefined; +let clientIdentity = ''; + +export function getSupabaseClient(url: string, publishableKey: string): SupabaseClient { + const identity = `${url}:${publishableKey}`; + if (!client || clientIdentity !== identity) { + client = createClient(url, publishableKey, { + auth: { + detectSessionInUrl: true, + persistSession: true, + autoRefreshToken: true, + }, + }); + clientIdentity = identity; + } + return client; +} diff --git a/frontend/src/auth/types.ts b/frontend/src/auth/types.ts index 176e9aa..2eac7bf 100644 --- a/frontend/src/auth/types.ts +++ b/frontend/src/auth/types.ts @@ -1,10 +1,11 @@ -import type { AuthResponse, RoleDefinition, UserProfile } from '../api/types'; +import type { RoleDefinition, UserProfile } from '../api/types'; export interface AuthContextValue { user: UserProfile | null; isLoading: boolean; personas: string[]; roles: Record; - login: (email: string, password: string) => Promise; + requestMagicLink: (email: string) => Promise; + loginWithGoogle: () => Promise; logout: () => void; } diff --git a/frontend/src/components/layout/AppShell.tsx b/frontend/src/components/layout/AppShell.tsx index 1851faa..f70e6ea 100644 --- a/frontend/src/components/layout/AppShell.tsx +++ b/frontend/src/components/layout/AppShell.tsx @@ -2,6 +2,7 @@ import { type FormEvent, type ReactNode, useEffect, useState } from 'react'; import { NavLink, useLocation, useNavigate } from 'react-router-dom'; import { Activity, + BrainCircuit, ChevronsUpDown, Command, FileStack, @@ -40,6 +41,7 @@ interface AppShellProps { const PRIMARY_NAV = [ { to: '/app', label: 'Overview', icon: LayoutDashboard, end: true }, + { to: '/app/investigate', label: 'Investigate', icon: BrainCircuit }, { to: '/app/qa', label: 'Ask DocuLens', icon: MessageSquareText }, { to: '/app/pipeline', label: 'Documents', icon: FileStack }, { to: '/app/work-queues', label: 'Work queues', icon: FolderKanban }, @@ -53,6 +55,7 @@ const SECONDARY_NAV = [ const PAGE_META: Record = { '/app': { title: 'Workspace', description: 'Your documents, decisions, and active work in one place.' }, '/app/qa': { title: 'Ask DocuLens', description: 'Explore your documents with answers grounded in source evidence.' }, + '/app/investigate': { title: 'Investigate', description: 'Give the agent an outcome and watch it build a verified evidence trail.' }, '/app/pipeline': { title: 'Documents', description: 'Review summaries, classifications, source content, and processing status.' }, '/app/work-queues': { title: 'Work queues', description: 'Route documents to the right team and keep decisions moving.' }, '/app/settings': { title: 'Workspace settings', description: 'Configure models, retrieval, access, and preferences.' }, @@ -218,9 +221,9 @@ function CommandEmpty({ onNavigate }: { onNavigate: (path: string) => void }) { return (

Quick actions

-