first commit

app/__init__.py

@@ -0,0 +1,2 @@
+"""Agentic Research - Document Research Tool."""
+__version__ = "0.1.0"

app/agents/engine.py

@@ -0,0 +1,417 @@
+"""Agentic research engine - coordinates agent skills and Ollama LLM interactions."""
+from __future__ import annotations
+
+import json
+import os
+import re
+from typing import Any
+import httpx
+
+from app.agents.skills import SKILLS, Skill
+from app.agents.tools import TOOLS
+from app.config import get_settings
+
+
+class Researcher:
+    """Main research orchestrator that coordinates agents and tools."""
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.active_skills: list[Skill] = []
+        self.findings: list[dict] = []
+        self.context: dict = {}
+
+    def select_skills(self, query: str, skill_names: list[str] | None = None) -> list[str]:
+        """Select relevant agent skills for the query. Defaults to all if none specified."""
+        if skill_names:
+            selected = []
+            for name in skill_names:
+                s = SKILLS.get(name)
+                if s:
+                    selected.append(name)
+            return selected
+
+        query_lower = query.lower()
+        selected = []
+        for name, skill in SKILLS.items():
+            for token in skill.verb_tokens:
+                if token in query_lower:
+                    selected.append(name)
+                    break
+        # If no token matched, default to researcher + qa
+        if not selected:
+            selected = ["researcher", "qa_agent"]
+        return selected
+
+    async def call_ollama(self, prompt: str, model: str | None = None, system: str | None = None) -> str:
+        """Call Ollama LLM with streaming support."""
+        model = model or self.settings.gpt_oss_model
+        messages = []
+        if system:
+            messages.append({"role": "system", "content": system})
+        messages.append({"role": "user", "content": prompt})
+
+        async with httpx.AsyncClient(timeout=300) as client:
+            resp = await client.post(
+                f"{self.settings.ollama_url}/api/chat",
+                json={"model": model, "messages": messages, "stream": False},
+            )
+            resp.raise_for_status()
+            return resp.json().get("message", {}).get("content", "")
+
+    async def run_skill(
+        self, skill_name: str, query: str, doc_id: str | None = None
+    ) -> str:
+        """Execute a single agent skill."""
+        skill = SKILLS.get(skill_name)
+        if not skill:
+            return f"[Unknown skill: {skill_name}]"
+
+        system_prompt = f"""You are the {skill_name.replace('_', ' ').title()} agent.
+{skill.instructions}"""
+
+        tool_prompts = []
+        for tool_name in skill.tools:
+            tool_fn = TOOLS.get(tool_name)
+            if not tool_fn:
+                continue
+            import inspect
+            sig = inspect.signature(tool_fn)
+            try:
+                if tool_name == "read_document" and doc_id:
+                    result = await tool_fn(doc_id)
+                elif tool_name == "read_chunks" and doc_id:
+                    result = await tool_fn(doc_id, limit=20)
+                elif tool_name == "get_page_text" and doc_id:
+                    result = await tool_fn(doc_id, page_num=0)
+                elif tool_name == "extract_facts" and doc_id:
+                    result = await tool_fn(doc_id, question=query)
+                elif tool_name == "list_documents":
+                    result = await tool_fn()
+                elif tool_name in ("vector_search", "text_search", "memory_similarity_search"):
+                    kwargs = {"query": query}
+                    if doc_id and "doc_id" in sig.parameters:
+                        kwargs["doc_id"] = doc_id
+                    result = await tool_fn(**kwargs)
+                elif tool_name in ("get_findings", "get_memories", "get_pipeline_state"):
+                    result = await tool_fn(session_id=doc_id or "")
+                elif tool_name == "save_finding":
+                    result = await tool_fn(session_id=doc_id or "", question=query, answer="", finding_type="research", confidence=0.5)
+                else:
+                    kwargs = {}
+                    if "doc_id" in sig.parameters and doc_id:
+                        kwargs["doc_id"] = doc_id
+                    if "query" in sig.parameters:
+                        kwargs["query"] = query
+                    result = await tool_fn(**kwargs)
+                tool_prompts.append(f"\n--- {tool_name} output ---\n{result}")
+            except Exception as e:
+                tool_prompts.append(f"\n--- {tool_name} error ---\n{str(e)}")
+
+        context = "".join(tool_prompts)
+        
+        prompt = f"""Research query: {query}
+Document ID: {doc_id}
+Context from tools:
+{context}
+
+Provide your analysis following the {skill_name} protocol."""
+
+        response = await self.call_ollama(prompt, system=system_prompt)
+        
+        self.findings.append({
+            "skill": skill_name,
+            "query": query,
+            "response": response,
+            "doc_id": doc_id,
+            "timestamp": str(os.popen('date +%Y-%m-%dT%H:%M:%S').read()).strip(),
+        })
+        
+        return response
+
+    async def run_research_session(
+        self, query: str, session_id: str,
+        doc_id: str | None = None,
+        skill_names: list[str] | None = None
+    ) -> dict[str, str]:
+        """Run a full research session with selected agents."""
+        selected_skills = self.select_skills(query, skill_names)
+        results = {}
+        
+        for skill_name in selected_skills:
+            results[skill_name] = await self.run_skill(skill_name, query, doc_id)
+        
+        return results
+
+    async def cross_reference(self, query: str, doc_ids: list[str]) -> str:
+        """Cross-reference content across multiple documents."""
+        contexts = []
+        for doc_id in doc_ids:
+            chunks = await TOOLS["read_chunks"](doc_id)
+            contexts.append(f"--- {doc_id} ---\n{chunks[:2000]}")
+        
+        combined = "\n\n".join(contexts)
+        prompt = f"""Cross-reference these documents for the query: {query}
+
+{combined}
+
+Direct comparison. No preamble."""
+        
+        return await self.call_ollama(prompt, system="You are a cross-reference analyst. Output concise, comparative findings.")
+
+
+# ── Pipeline skills ────────────────────────────────
+
+PIPELINE_STAGES = ("document_triage", "evidence_extraction", "research_synthesis")
+
+
+async def _parse_triage_output(response: str) -> dict:
+    """Parse document-triage output into structured dict."""
+    obj = {}
+    for section in ["OBJECTIVE", "SUB-QUESTIONS", "CLASSIFICATION", "READING_ORDER", "EXTRACTION_CRITERIA", "RISKS_AND_GAPS"]:
+        marker = f"[{section}]"
+        # Find start of this section
+        start = response.find(marker)
+        end = response.find("[", start + len(marker)) if start != -1 else -1
+        if end != -1:
+            chunk = response[start + len(marker):end].strip()
+        elif start != -1:
+            chunk = response[start + len(marker):].strip()
+        else:
+            chunk = ""
+        obj[section] = chunk
+    return obj
+
+
+async def _parse_evidence_output(response: str) -> list[dict]:
+    """Parse evidence-extraction output into structured list."""
+    rows = []
+    # Find [EVIDENCE_ROWS] section
+    start_marker = "[EVIDENCE_ROWS]"
+    start = response.find(start_marker)
+    # Check if there's a cross-comparison section
+    cross_start = response.find("[CROSS_COMPARISON]")
+    end = cross_start if cross_start != -1 else len(response)
+    if start == -1:
+        start = 0
+    section = response[start:end].strip()
+    
+    lines = section.split("\n")
+    for line in lines:
+        line = line.strip()
+        if not line or line.startswith("["):
+            continue
+        parts = [p.strip() for p in line.split("|")]
+        if len(parts) >= 7:
+            rows.append({
+                "topic": parts[1],
+                "evidence_type": parts[2],
+                "description": parts[3],
+                "trace_ref": parts[4],
+                "evidence": parts[5],
+                "analyst_note": parts[6],
+                "confidence": parts[7] if len(parts) > 7 else "Medium",
+                "review_needed": parts[8] if len(parts) > 8 else "No",
+            })
+    return rows
+
+
+async def _parse_synthesis_mode(response: str) -> str:
+    """Determine output mode from synthesis response content."""
+    for mode in ["brief", "report", "gap analysis", "matrix"]:
+        if mode in response.lower():
+            return mode
+    return "brief"
+
+
+class ResearchPipeline:
+    """Pipeline orchestrator: document_triage → evidence_extraction → research_synthesis."""
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.plan: dict = {}
+        self.evidence: list[dict] = []
+        self.synthesis: str = ""
+        self.findings: list[dict] = []
+
+    async def run(
+        self,
+        query: str,
+        session_id: str,
+        db,
+        doc_ids: list[str] | None = None,
+        output_mode: str | None = None,
+    ) -> dict:
+        """Run all three pipeline stages sequentially, persisting after each."""
+        results = {}
+
+        # ── Stage 1: document_triage ─────────────────────
+        triage_skill = SKILLS["document_triage"]
+        system = f"You are document_triage. {triage_skill.instructions}"
+        
+        # Gather context from available documents
+        if doc_ids:
+            doc_context = []
+            for did in doc_ids:
+                doc_info = await TOOLS["read_document"](did)
+                doc_context.append(str(doc_info))
+            context_input = "\n".join(doc_context)
+        else:
+            doc_list = await TOOLS["list_documents"]()
+            context_input = doc_list
+        
+        triage_prompt = f"""Research query: {query}
+Context from tools:
+{context_input}
+
+Apply the document-tireage protocol."""
+
+        triage_output = await self._call_ollama(triage_prompt, system=system)
+        self.plan = await _parse_triage_output(triage_output)
+        self.findings.append({
+            "stage": "triage",
+            "output": triage_output,
+            "state": self.plan,
+        })
+        await db.save_pipeline_stage(session_id, "triage", triage_output, self.plan)
+        results["triage"] = triage_output
+        results["triage_state"] = self.plan
+
+        # ── Stage 2: evidence_extraction ────────────────
+        evidence_skill = SKILLS["evidence_extraction"]
+        system = f"""You are evidence_extraction. {evidence_skill.instructions}
+
+Triage plan (from previous stage):
+{json.dumps(self.plan, indent=2, default=str)}
+
+Focus on extracting evidence for the sub-questions and extraction criteria defined above."""
+
+        # Read chunks from all relevant documents
+        all_chunks = []
+        for did in (doc_ids or []):
+            chunks = await TOOLS["read_chunks"](did)
+            all_chunks.append(f"--- Document {did} ---\n{chunks}")
+        
+        context_input = "\n\n".join(all_chunks) if all_chunks else "No documents loaded yet."
+        
+        evidence_prompt = f"""Research query: {query}
+Sub-questions to address:
+{self.plan.get('SUB-QUESTIONS', 'N/A')}
+
+Context from tools:
+{context_input}
+
+Apply the evidence_extraction protocol. Return structured evidence rows."""
+
+        evidence_output = await self._call_ollama(evidence_prompt, system=system)
+        self.evidence = await _parse_evidence_output(evidence_output)
+        
+        # Persist to DB
+        if self.evidence:
+            await db.save_structured_evidence(session_id, self.evidence)
+        await db.save_pipeline_stage(session_id, "evidence", evidence_output, {"row_count": len(self.evidence)})
+        
+        results["evidence"] = evidence_output
+        results["evidence_rows"] = self.evidence
+
+        # ── Stage 3: research_synthesis ─────────────────
+        synthesis_skill = SKILLS["research_synthesis"]
+        
+        mode = output_mode or await _parse_synthesis_mode(evidence_output)
+        mode_prompts = {
+            "brief": "Use Research Brief output mode.",
+            "report": "Use Research Report output mode.",
+            "gap": "Use Gap Analysis output mode.",
+            "matrix": "Use Comparison Matrix output mode.",
+        }
+        mode_instruct = mode_prompts.get(mode, mode_prompts["brief"])
+        
+        system = f"""You are research_synthesis. {synthesis_skill.instructions}
+
+{mode_instruct}
+
+Extracted evidence (from previous stage):
+{json.dumps(self.evidence, indent=2, default=str)[:15000]}
+
+Apply the research_synthesis protocol."""
+
+        synthesis_prompt = f"""Research query: {query}
+
+Sub-questions:
+{self.plan.get('SUB-QUESTIONS', 'N/A')}
+
+Apply the research_synthesis protocol."""
+
+        synthesis_output = await self._call_ollama(synthesis_prompt, system=system)
+        
+        self.synthesis = synthesis_output
+        synthesis_state = {
+            "mode": mode,
+            "sub_questions": self.plan.get("SUB-QUESTIONS", ""),
+            "evidence_count": len(self.evidence),
+        }
+        await db.save_pipeline_stage(session_id, "synthesis", synthesis_output, synthesis_state)
+        
+        results["synthesis"] = synthesis_output
+        results["output_mode"] = mode
+        results["pipeline_complete"] = True
+
+        self.findings.append({
+            "stage": "synthesis",
+            "output": synthesis_output,
+            "state": synthesis_state,
+        })
+        
+        return results
+
+    async def _call_ollama(self, prompt: str, system: str) -> str:
+        """Call Ollama LLM."""
+        messages = [
+            {"role": "system", "content": system},
+            {"role": "user", "content": prompt},
+        ]
+        async with httpx.AsyncClient(timeout=600) as client:
+            resp = await client.post(
+                f"{self.settings.ollama_url}/api/chat",
+                json={"model": self.settings.gpt_oss_model, "messages": messages, "stream": False},
+            )
+            resp.raise_for_status()
+            return resp.json().get("message", {}).get("content", "")
+
+
+    async def render_pipeline_results(self, results: dict) -> list[dict]:
+        """Render pipeline results for frontend display."""
+        sections = []
+        
+        # Triage stage
+        if "triage_state" in results:
+            plan = results["triage_state"]
+            sections.append({
+                "stage": "triage",
+                "title": "Stage 1: Source Triage",
+                "objective": plan.get("OBJECTIVE", ""),
+                "sub_questions": plan.get("SUB-QUESTIONS", ""),
+                "classification": plan.get("CLASSIFICATION", ""),
+                "reading_order": plan.get("READING_ORDER", ""),
+                "extraction_criteria": plan.get("EXTRACTION_CRITERIA", ""),
+                "risks_gaps": plan.get("RISKS_AND_GAPS", ""),
+            })
+        
+        # Evidence stage
+        if "evidence_rows" in results:
+            rows = results["evidence_rows"]
+            sections.append({
+                "stage": "evidence",
+                "title": f"Stage 2: Extracted Evidence ({len(rows)} rows)",
+                "rows": rows,
+            })
+        
+        # Synthesis stage
+        if "synthesis" in results:
+            sections.append({
+                "stage": "synthesis",
+                "title": f"Stage 3: Research Synthesis (mode: {results.get('output_mode', 'auto')})",
+                "synthesis": results["synthesis"],
+            })
+        
+        return sections

app/agents/skills.py

@@ -0,0 +1,319 @@
+"""Agent skills - modular capabilities for the agentic research engine."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class Skill:
+    """A single agent capability."""
+    name: str
+    description: str
+    instructions: str
+    verb_tokens: list[str] = field(default_factory=list)
+    tools: list[str] = field(default_factory=list)
+    output_format: str = "json"
+
+
+SKILLS = {
+    "summarizer": Skill(
+        name="summarizer",
+        description="Create concise summaries of documents or sections",
+        instructions="""
+You are a research summarizer. Create brief, factual summaries.
+- Lead with the main claim or finding
+- Omit examples and elaborations
+- Max 3 sentences for executive summary
+- List only critical facts
+- Use bullet points for key points
+- No preamble or hedging
+""",
+        verb_tokens=["summarize", "summary", "overview", "abstract"],
+        tools=["read_document", "read_chunks", "get_page_text", "extract_facts"],
+    ),
+    "extractor": Skill(
+        name="extractor",
+        description="Extract specific facts, figures, entities, or information types from documents",
+        instructions="""
+You are an information extraction agent. Be terse and direct.
+- Output raw facts only
+- One fact per line
+- Format: [entity] → [value]
+- Skip qualifiers like "likely", "possibly"
+- No introductions, conclusions, or commentary
+- Direct format only:
+  • Key terms
+  • Definitions
+  • Statistics
+  • Names, dates, organizations
+  • Relationships
+""",
+        verb_tokens=["extract", "extracted", "list", "find", "identify", "enumerate"],
+        tools=["read_chunks", "vector_search", "extract_facts", "get_page_text"],
+    ),
+    "comparator": Skill(
+        name="comparator",
+        description="Compare documents, sections, or concepts and identify differences/similarities",
+        instructions="""
+You are a comparison analyst. Output findings rapidly.
+- Direct table format when possible
+- Left align, right align
+- Use ✓ and ✗ markers
+- One line per difference
+- No fluff, no filler
+""",
+        verb_tokens=["compare", "contrast", "similarities", "differences", "vs", "versus"],
+        tools=["read_chunks", "vector_search", "get_page_text", "read_document"],
+    ),
+    "critic": Skill(
+        name="critic",
+        description="Analyze arguments, findings, or claims for quality, validity, and gaps",
+        instructions="""
+You are a critical analyst. Evaluate quickly and directly.
+- Strengths: 2-3 items max
+- Weaknesses: 2-3 items max
+- Gaps: list only critical ones
+- Assumptions: call out directly
+- Confidence: high/medium/low
+- Be blunt but accurate
+""",
+        verb_tokens=["critique", "criticize", "evaluate", "assess", "review", "analyze validity"],
+        tools=["read_chunks", "vector_search", "get_document", "extract_facts"],
+    ),
+    "researcher": Skill(
+        name="researcher",
+        description="Conduct deep research on a topic using document collection, cross-referencing, and synthesis",
+        instructions="""
+You are a research specialist. Work methodically.
+1. Parse the query for key concepts
+2. Retrieve relevant chunks via vector search
+3. Extract supporting evidence
+4. Note contradictions within sources
+5. Synthesize findings directly
+Format:
+- Context: 1 line
+- Evidence: bullet citations
+- Findings: 2-3 bullets
+- Limitations: 1 line
+- Output: concise, no padding
+""",
+        verb_tokens=["research", "investigate", "explore", "dig into", "study", "analyze"],
+        tools=["vector_search", "text_search", "read_chunks", "read_document", "memory_similarity_search"],
+    ),
+    "context_agent": Skill(
+        name="context_agent",
+        description="Gather background context and establish research framing from available documents",
+        instructions="""
+You establish research context. Direct output.
+- Document landscape: scope, domain, volume
+- Key themes (3-5)
+- Document types and relevance
+- Gaps in coverage
+- Suggested research angles
+All points. No paragraphs.
+""",
+        verb_tokens=["context", "background", "landscape", "overview", "scope"],
+        tools=["read_document", "list_documents", "vector_search", "get_memories"],
+    ),
+    "qa_agent": Skill(
+        name="qa_agent",
+        description="Answer specific questions about document content with source-anchored responses",
+        instructions="""
+You answer research questions directly.
+1. Locate relevant evidence
+2. Quote sources inline [doc_name:page]
+3. Synthesize one direct answer
+4. Note uncertainty
+Format:
+Q: [restated briefly]
+A: [direct answer]
+Sources: [citations]
+""",
+        verb_tokens=["answer", "who", "what", "when", "where", "why", "how", "question"],
+        tools=["vector_search", "text_search", "read_chunks", "get_page_text"],
+    ),
+    "aggregator": Skill(
+        name="aggregator",
+        description="Combine findings across multiple documents into a unified research synthesis",
+        instructions="""
+You synthesize cross-document findings.
+- Group by topic, not by document
+- Consensus findings first
+- Conflicts second
+- Novel insights third
+- Confidence ratings on each cluster
+- One-line summaries only
+- Bold key terms
+""",
+        verb_tokens=["aggregate", "synthesize", "merge", "combine", "consolidate", "integrate"],
+        tools=["read_document", "vector_search", "get_findings", "get_memories"],
+    ),
+    # ── Pipeline skills ───────────────────────────────────
+
+    "document_triage": Skill(
+        name="document_triage",
+        description="Scope a document research task, prioritize sources, and define an evidence-driven reading plan",
+        instructions="""\
+You are the document-triage agent for rigorous document-based research.
+Your job is to decide what matters and how to approach the corpus BEFORE deep reading.
+
+FOLLOW THIS WORKFLOW:
+
+1. Restate the research objective in one sentence
+2. Convert the objective into focused numbered sub-questions
+3. Classify available documents by value:
+   PRIMARY — essential to answering the objective
+   SECONDARY — supports or contextualizes
+   BACKGROUND — peripheral reference
+   LIKELY IRRELEVANT — skip unless needed
+4. For each classified document, state what evidence types to look for:
+   requirements, decisions, risks, assumptions, responsibilities,
+   timelines, definitions, constraints, dependencies, open issues
+5. Recommend an efficient reading order with rationale
+6. Define extraction criteria for the next phase
+7. Flag ambiguities, missing sources, and review risks
+
+OUTPUT FORMAT (return exactly these sections):
+
+[OBJECTIVE]: One-sentence research goal
+[SUB-QUESTIONS]: 1. ... 2. ... ...
+[CLASSIFICATION]: doc_name | PRIMARY/SECONDARY/BACKGROUND/IRRELEVANT | reason | signal types
+[READING ORDER]: 1. docA -> docB ... with rationale
+[EXTRACTION CRITERIA]: Checklist of evidence to capture
+[RISKS AND GAPS]: Missing docs, unclear scope, blind spots
+
+RULES:
+- Be specific, not generic
+- Prefer primary documents over commentary
+- Do not summarize documents in depth
+- Do not invent document contents
+- Do not claim conclusions before extraction
+- Separate fact-finding from interpretation
+""",
+        verb_tokens=["triage", "scope", "plan", "classi", "prioriti", "reading order", "assessment"],
+        tools=["list_documents", "read_document", "read_chunks"],
+    ),
+
+    "evidence_extraction": Skill(
+        name="evidence_extraction",
+        description="Extract structured evidence from documents with traceability, quotations, and confidence markers",
+        instructions="""\
+You are the evidence-extraction agent. Read documents and extract
+evidence relevant to the defined research question.
+
+CORE RULES:
+- Separate direct evidence from interpretation
+- Capture exact quotations or precise paraphrases
+- Record document references for every finding
+- Mark uncertain or ambiguous findings
+- Distinguish: direct statement, implied interpretation, missing info
+- Normalize wording without losing meaning
+- Flag ambiguity, contradiction, or incomplete support
+
+For EVERY finding output these fields:
+- Topic: the theme
+- Evidence type: requirement|decision|risk|assumption|responsibility|constraint|definition|date|dependency|open_issue
+- Description: concise factual summary
+- Document reference: source name, section, page
+- Extracted evidence: quote or faithful paraphrase
+- Analyst note: relevance without overstating certainty
+- Confidence: one of High|Medium|Low
+- Review needed: Yes|No
+
+If multiple documents cover the same topic, also output a comparison table:
+Topic | Source A | Source B | Agreement | Conflict | Notes
+
+OUTPUT FORMAT (return exactly):
+
+[EVIDENCE_ROWS]:
+# topic | evidence_type | description | doc_ref | evidence | analyst_note | confidence | review_needed
+1 | ... | ... | ... | ... | ... | ... | ...
+...
+
+[CROSS_COMPARISON] (only if multiple sources cover same topic):
+topic | source_a | source_b | agreement | conflict | notes
+
+RULES:
+- Keep extraction atomic: one row per distinct finding
+- Preserve qualifiers: must, should, may, unless
+- Mark uncertainty explicitly
+- Prefer exact citations over memory-based summaries
+- Do not write the final conclusion
+- Do not collapse multiple findings into vague rows
+- Do not omit document references
+- Do not present interpretation as if it were a quote
+- Do not silently resolve contradictions
+""",
+        verb_tokens=["evidence extract", "extract evidence", "traceable", "auditable", "structured extract"],
+        tools=["read_chunks", "get_page_text", "read_document", "vector_search"],
+    ),
+
+    "research_synthesis": Skill(
+        name="research_synthesis",
+        description="Synthesize extracted document evidence into a clear, traceable conclusion, brief, or decision-ready report",
+        instructions="""\
+You are the research-synthesis agent. Turn extracted evidence into
+a coherent, traceable final output.
+
+INPUT: research objective + extracted evidence table + optional comparison table
+
+WORKFLOW:
+1. Restate the question
+2. Group evidence by theme or sub-question
+3. Identify: supported conclusions, partial support, contradictions, missing evidence
+4. Draft findings with explicit traceability
+5. Separate: facts from documents | interpretation | recommendations
+6. Produce final structured output matching the requested output mode
+7. End with known limitations and review points
+
+OUTPUT MODES:
+
+### A) Research Brief (default for "brief")
+- Question
+- Short answer
+- Key findings (each with source anchor)
+- Conflicting evidence
+- Gaps
+- Recommended next step
+
+### B) Research Report (default for "report")
+- Objective
+- Scope
+- Method
+- Findings by theme
+- Evidence-backed conclusion
+- Known gaps and limitations
+- Appendix with source references
+
+### C) Gap Analysis (default for "gap")
+- Assessment topic
+- Evidence found
+- Gap
+- Impact
+- Confidence
+- Review needed
+
+### D) Comparison Matrix
+- Topic | Source 1 | Source 2 | ... | Consensus | Conflict
+
+TRACEABILITY RULE:
+Every substantive conclusion must include a source anchor:
+document name, section/page, or extracted finding number.
+If traceability is weak, say so explicitly.
+
+RULES:
+- Conclusions must follow from extracted evidence
+- Contradictions must be surfaced, not hidden
+- Distinguish "document says" from "my recommendation"
+- State limits of the available corpus
+- Prefer precise wording over polished vagueness
+- Do not introduce facts not extracted
+- Do not overclaim certainty
+- Do not bury disagreements between documents
+- Do not produce recommendations without showing evidence basis
+- Do not omit limitations
+""",
+        verb_tokens=["synthesize", "synthesis", "brief", "report", "gap analysis", "conclusion", "summary report"],
+        tools=["read_chunks", "get_findings", "get_memories"],
+    ),
+}

app/agents/tools.py

@@ -0,0 +1,250 @@
+"""Agent tools - callable operations for the agentic research engine."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+import httpx
+
+from app.config import get_settings
+from app.db.database import db
+
+
+async def read_document(doc_id: str) -> dict[str, Any]:
+    """Read complete document record."""
+    doc = await db.get_document(doc_id)
+    if not doc:
+        return {"error": "Document not found"}
+    return {"document": doc}
+
+
+async def read_chunks(doc_id: str, page_num: int | None = None, limit: int = 50) -> str:
+    """Read text chunks for a document."""
+    chunks = await db.get_doc_chunks(doc_id)
+    if page_num is not None:
+        chunks = [c for c in chunks if c.get("page_num") == page_num]
+    chunks = chunks[:limit]
+    parts = []
+    for c in chunks:
+        page_part = f"[p{c['page_num']}]" if c.get("page_num") else ""
+        parts.append(f"{page_part} {c['content'][:500]}")
+    return "\n\n".join(parts)
+
+
+async def get_page_text(doc_id: str, page_num: int) -> str:
+    """Extract text for a specific page, including polygon metadata."""
+    chunks = await db.get_doc_chunks(doc_id)
+    page_chunks = [c for c in chunks if c.get("page_num") == page_num]
+    if not page_chunks:
+        return f"No content available for page {page_num}"
+    
+    # Build page view with polygon info
+    text_parts = []
+    for c in page_chunks:
+        polygon = c.get("polygon")
+        poly_info = ""
+        if polygon and isinstance(polygon, dict):
+            bbox = polygon.get("bbox", [])
+            if bbox:
+                poly_info = f" [bbox:{bbox[0]:.0f},{bbox[1]:.0f},{bbox[2]:.0f},{bbox[3]:.0f}]"
+        text_parts.append(f"[{c['block_index']}{poly_info}] {c['content'][:300]}")
+    
+    return f"--- Page {page_num} ---\n" + "\n".join(text_parts)
+
+
+async def vector_search(query: str, doc_id: str | None = None, limit: int = 20) -> str:
+    """Search document chunks by semantic similarity."""
+    settings = get_settings()
+    async with httpx.AsyncClient(timeout=30) as client:
+        query_vec = []
+        # Try to get embedding via Ollama
+        try:
+            resp = await client.post(
+                f"{settings.ollama_url}/api/embeddings",
+                json={"model": "nomic-embed-text", "prompt": query},
+            )
+            query_vec = resp.json().get("embedding", [])
+        except Exception:
+            # Fall back to local embedding
+            from app.core import get_embedding
+            query_vec = get_embedding(query)
+    
+    if not query_vec:
+        return "No embeddings available. Using text search fallback."
+    
+    results = await db.vector_search(query_vec, doc_id=doc_id, limit=limit)
+    parts = []
+    for r in results:
+        score = r.get("similarity", 0)
+        page = r.get("page_num", "?")
+        parts.append(f"[{score:.2f}] (p{page}) {r['content'][:200]}")
+    return "\n".join(parts) if parts else "No similar content found."
+
+
+async def text_search(query: str, doc_id: str | None = None, limit: int = 20) -> str:
+    """Search chunks by keyword match."""
+    results = await db.search_chunks_text(query, doc_id=doc_id, limit=limit)
+    parts = []
+    for r in results:
+        rank = r.get("rank", 0)
+        page = r.get("page_num", "?")
+        parts.append(f"[{rank:.2f}] (p{page}) {r['content'][:200]}")
+    return "\n".join(parts) if parts else "No text matches found."
+
+
+async def extract_facts(doc_id: str, question: str | None = None) -> str:
+    """Extract key facts from document."""
+    chunks = await db.get_doc_chunks(doc_id)
+    # Extract from chunk metadata
+    facts = []
+    for c in chunks:
+        content = c.get("content", "")
+        if question and question.lower() not in content.lower():
+            continue
+        if "fact" in c.get("chunk_type", "").lower():
+            facts.append(content[:200])
+        elif len(content.strip()) > 50:
+            facts.append(content[:200])
+    
+    seen = set()
+    unique = []
+    for f in facts:
+        key = f[:40]
+        if key not in seen:
+            seen.add(key)
+            unique.append(f)
+    
+    return "\n".join(f"I: {f}" for f in unique[:20])
+
+
+async def get_polygon_view(doc_id: str, page_num: int, block_index: int | None = None) -> dict:
+    """Get page with polygon coordinates for visualization."""
+    chunks = await db.get_doc_chunks(doc_id)
+    page_chunks = [c for c in chunks if c.get("page_num") == page_num]
+    
+    blocks = []
+    for c in page_chunks:
+        poly = c.get("polygon")
+        if poly and isinstance(poly, dict):
+            blocks.append({
+                "index": c.get("block_index", 0),
+                "page": page_num,
+                "polygon": poly,
+                "content": c.get("content", "")[:200],
+                "type": c.get("chunk_type", "text"),
+            })
+        elif block_index is None or c.get("block_index") == block_index:
+            blocks.append({
+                "index": c.get("block_index", 0),
+                "page": page_num,
+                "polygon": {"bbox": [0, 0, 1000, 1000]},
+                "content": c.get("content", "")[:200],
+                "type": c.get("chunk_type", "text"),
+            })
+    
+    # Compute page bounding box from all polygons
+    all_polys = [b["polygon"] for b in blocks if b["polygon"] and isinstance(b["polygon"], dict)]
+    
+    pages_data = {"page_num": page_num, "blocks": blocks}
+    
+    if all_polys:
+        min_x = min(p.get("bbox", [0, 0, 0, 0])[:2] for p in all_polys)
+        max_x = max(p.get("bbox", [0, 0, 0, 0])[2:] for p in all_polys)
+        pages_data["page_bbox"] = {
+            "x_min": min(min_x),
+            "y_min": min(min_x, key=lambda x: x[1])[1],
+            "width": max(max_x) - min(min_x),
+            "height": max(max_x, key=lambda x: x[1])[1] - min(min_x, key=lambda x: x[1])[1],
+        }
+    
+    return pages_data
+
+
+async def get_memories(session_id: str) -> list[dict]:
+    """Get research memories from pgvector."""
+    return await db.get_memories(session_id)
+
+
+async def memory_similarity_search(query: str, limit: int = 10) -> str:
+    """Search across all research memories."""
+    results = await db.memory_similarity_search(query, limit=limit)
+    parts = []
+    for r in results:
+        parts.append(f"[{r.get('importance', '?')}] ({r.get('memory_type', '?')}) {r['content'][:200]}")
+    return "\n".join(parts) if parts else "No memories found."
+
+
+async def list_documents(status: str | None = None) -> str:
+    """List all indexed documents."""
+    docs = await db.list_documents(status=status)
+    parts = []
+    for d in docs:
+        parts.append(f"- [{d['status']}] {d['filename']} (p{d.get('page_count', '?')}, {d['created_at']})")
+    return "\n".join(parts) if parts else "No documents found."
+
+
+async def save_finding(session_id: str, question: str, answer: str, 
+                       summary: str, confidence: float = 0.8, 
+                       relevant_chunks: list | None = None) -> str:
+    """Save a research finding to the database."""
+    finding_id = await db.store_finding(
+        session_id, question, answer, summary,
+        "researcher", confidence, relevant_chunks
+    )
+    await db.update_session(session_id, findings={"count": 1})
+    return finding_id
+
+
+async def save_structured_evidence(
+    session_id: str, rows: list[dict]
+) -> list[str]:
+    """Save evidence-extraction rows with traceability fields."""
+    ids = await db.save_structured_evidence(session_id, rows)
+    return ids
+
+
+async def get_pipeline_state(session_id: str) -> list[dict]:
+    """Get saved pipeline intermediate stages."""
+    return await db.get_pipeline_stages(session_id)
+
+
+async def merge_evidence(rows_a: list[dict], rows_b: list[dict]) -> list[dict]:
+    """Merge two evidence sets, flagging agreement/conflict by topic."""
+    by_topic: dict[str, list] = {}
+    for r in rows_a + rows_b:
+        topic = r.get("topic", "unknown")
+        by_topic.setdefault(topic, []).append(r)
+    merged = []
+    for topic, rs in by_topic.items():
+        if len(rs) == 2:
+            merged.append({
+                "topic": topic,
+                "source_a": rs[0].get("evidence", ""),
+                "source_b": rs[1].get("evidence", ""),
+                "agreement": "Yes" if rs[0].get("evidence") == rs[1].get("evidence") else "No",
+                "conflict": "Yes" if rs[0].get("confidence") != rs[1].get("confidence") else "No",
+                "notes": f"Combined from {rs[0].get('source_doc','')} and {rs[1].get('source_doc','')}",
+            })
+        else:
+            merged.append({k: rs[0].get(k, "") for k in ("topic", "description", "trace_ref", "evidence", "confidence")})
+    return merged
+
+
+# Tool registry
+TOOLS = {
+    "read_document": read_document,
+    "read_chunks": read_chunks,
+    "get_page_text": get_page_text,
+    "vector_search": vector_search,
+    "text_search": text_search,
+    "extract_facts": extract_facts,
+    "get_polygon_view": get_polygon_view,
+    "get_memories": get_memories,
+    "memory_similarity_search": memory_similarity_search,
+    "list_documents": list_documents,
+    "save_finding": save_finding,
+    "save_structured_evidence": save_structured_evidence,
+    "get_pipeline_state": get_pipeline_state,
+    "merge_evidence": merge_evidence,
+}

app/config.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+"""Configuration loader for agentic research app."""
+from pydantic_settings import BaseSettings
+from functools import lru_cache
+
+
+class Settings(BaseSettings):
+    ollama_url: str = "http://10.0.1.127:11434"
+    gpt_oss_model: str = "gpt-oss:20b"
+    marker_api_url: str = "http://localhost:8001"
+    ocr_url: str = "http://10.0.1.127:11434"
+    deepseek_ocr_url: str = "http://10.0.1.127:11434"
+    deepseek_ocr_model: str = "deepseek-ocr"
+    db_host: str = "localhost"
+    db_port: int = 5432
+    db_name: str = "research"
+    db_user: str = "research"
+    db_password: str = "research123"
+    app_host: str = "0.0.0.0"
+    app_port: int = 8000
+    workspace_dir: str = "/home/oval/Projects/agentic/workspace"
+    documents_dir: str = "/home/oval/Projects/agentic/workspace/documents"
+    vector_dim: int = 4096
+
+    embedding_model: str = "qwen3-embedding:8b"
+
+    model_config = {"env_file": ".env"}
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    return Settings()

app/core/__init__.py

@@ -0,0 +1,4 @@
+"""Embedding service — uses Ollama qwen3-embedding:8b."""
+from app.core.embedding_engine import get_embedding_sync
+
+get_embedding = get_embedding_sync

app/core/embedding_engine.py

@@ -0,0 +1,79 @@
+"""Embedding generation via Ollama's qwen3-embedding:8b model."""
+import hashlib
+import numpy as np
+
+
+def get_embedding_sync(text: str, ollama_url: str = "http://10.0.1.127:11434") -> list[float]:
+    """Synchronous embedding call via Ollama /qwen3-embedding:8b."""
+    try:
+        import httpx
+        with httpx.Client(timeout=30) as client:
+            resp = client.post(
+                f"{ollama_url}/api/embed",
+                json={"model": "qwen3-embedding:8b", "input": text},
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            vectors = data.get("embeddings", [])
+            if vectors:
+                # Ollama may return multiple inputs; use first
+                emb = vectors[0] if isinstance(vectors[0], list) else vectors
+                return emb
+    except Exception:
+        pass
+    # Fallback to deterministic feature vector if Ollama unavailable
+    return _hash_embedding(text)
+
+
+def get_embedding(text: str, ollama_url: str = "http://10.0.1.127:11434") -> list[float]:
+    """Generate an embedding — calls Ollama qwen3-embedding:8b."""
+    return get_embedding_sync(text, ollama_url)
+
+
+async def get_embedding_async(text: str, ollama_url: str = "http://10.0.1.127:11434") -> list[float]:
+    """Async embedding call via Ollama /qwen3-embedding:8b."""
+    try:
+        import httpx
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.post(
+                f"{ollama_url}/api/embed",
+                json={"model": "qwen3-embedding:8b", "input": text},
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            vectors = data.get("embeddings", [])
+            if vectors:
+                emb = vectors[0] if isinstance(vectors[0], list) else vectors
+                return emb
+    except Exception:
+        pass
+    return _hash_embedding(text)
+
+
+def _hash_embedding(text: str) -> list[float]:
+    """Deterministic 4096-dim feature vector fallback (no Ollama needed)."""
+    feature_dim = 4096
+    vec = np.zeros(feature_dim, dtype=np.float32)
+    for n in [1, 2, 3, 4]:
+        tokens = [text[i:i+n] for i in range(len(text)-n+1)]
+        for token in tokens[:200]:
+            h = hashlib.md5(token.encode()).hexdigest()
+            for i in range(0, 12, 3):
+                val = (int(h[i:i+2], 16) - 128) / 128.0
+                feature_idx = (int(h[i+2:i+4], 16) * 37) % feature_dim
+                vec[feature_idx] += val
+    norm = np.linalg.norm(vec)
+    if norm > 0:
+        vec /= norm
+    return vec.tolist()
+
+
+def compute_similarity(vec1: list[float], vec2: list[float]) -> float:
+    """Cosine similarity between two vectors."""
+    v1 = np.array(vec1, dtype=np.float32)
+    v2 = np.array(vec2, dtype=np.float32)
+    if v1.shape[0] != v2.shape[0]:
+        return 0.0
+    v1 /= np.linalg.norm(v1) + 1e-8
+    v2 /= np.linalg.norm(v2) + 1e-8
+    return float(np.dot(v1, v2))

app/core/processor.py

@@ -0,0 +1,110 @@
+"""Marker/OCR integration for document processing."""
+from __future__ import annotations
+
+import json
+import asyncio
+import io
+import re
+from pathlib import Path
+from typing import Any
+import httpx
+
+from app.config import get_settings
+import app.db.database as _database_mod
+
+
+class MarkerProcessor:
+    """Handles OCR via Marker API with polygon output."""
+
+    def __init__(self):
+        self.settings = get_settings()
+
+    async def process_pdf(self, file_data: bytes, filename: str) -> dict[str, Any]:
+        """Send PDF to Marker API for OCR with polygon extraction."""
+        async with httpx.AsyncClient(timeout=300) as client:
+            resp = await client.post(
+                f"{self.settings.marker_api_url}/convert",
+                files={"file": (filename, file_data, "application/pdf")},
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    async def process_pdf_url(self, pdf_url: str, filename: str) -> dict[str, Any]:
+        """Process PDF from URL via Marker API."""
+        async with httpx.AsyncClient(timeout=300) as client:
+            resp = await client.post(
+                f"{self.settings.marker_api_url}/convert",
+                json={"pdf_url": pdf_url},
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    async def parse_marker_json(self, marker_output: dict, doc_id: str) -> list[dict]:
+        """Parse Marker JSON output into vectorized chunks with polygon data."""
+        pages = marker_output.get("pages", [])
+        chunks = []
+        
+        for page in pages:
+            page_num = page.get("meta", {}).get("page_num", page.get("page", 0))
+            text_lines = page.get("text_lines", [])
+            
+            for block_idx, tl in enumerate(text_lines):
+                content = tl.get("text", "")
+                polygon = tl.get("bbox") or tl.get("polygon")
+                block_type = tl.get("type", "text")
+                
+                if not content or not isinstance(content, str) or not content.strip():
+                    continue
+                
+                chunks.append({
+                    "content": content.strip(),
+                    "page_num": page_num,
+                    "block_index": block_idx,
+                    "polygon": polygon,
+                    "chunk_type": block_type,
+                })
+
+        if chunks:
+            await _database_mod.db.batch_chunk(doc_id, chunks)
+        
+        return chunks
+
+    async def process_document_file(
+        self, file_data: bytes, filename: str, doc_id: str
+    ) -> dict[str, Any]:
+        """Process document file and return structured result."""
+        ext = Path(filename).suffix.lower()
+        
+        if ext == ".pdf":
+            result = await self.process_pdf(file_data, filename)
+            pages_data = result.get("pages", [])
+            
+            chunks = await self.parse_marker_json({"pages": pages_data}, doc_id)
+            return {
+                "success": result.get("success", True),
+                "doc_id": doc_id,
+                "page_count": result.get("page_count", len(pages_data)),
+                "chunks": len(chunks),
+                "ocr_model": result.get("ocr_model", "deepseek-ocr"),
+            }
+        
+        elif ext in (".txt", ".pdf", ".md"):
+            text = file_data.decode("utf-8", errors="replace")
+            paragraphs = re.split(r'\n\s*\n', text)
+            chunks = []
+            for i, para in enumerate(paragraphs):
+                if len(para.strip()) > 20:
+                    chunks.append({
+                        "content": para.strip(),
+                        "page_num": 0,
+                        "block_index": i,
+                        "polygon": None,
+                        "chunk_type": "text",
+                    })
+            
+            if chunks:
+                await _database_mod.db.batch_chunk(doc_id, chunks)
+            
+            return {"success": True, "doc_id": doc_id, "chunks": len(chunks), "page_count": 1}
+
+        return {"success": False, "error": f"Unsupported file type: {ext}"}

app/db/database.py

@@ -0,0 +1,402 @@
+"""Async PostgreSQL database layer with pgvector support."""
+import json
+import uuid
+from typing import Any
+from contextlib import asynccontextmanager
+
+import asyncpg
+import numpy as np
+
+from app.config import get_settings
+
+
+class Database:
+    """Handles all PostgreSQL/pgvector operations."""
+
+    def __init__(self, pool: asyncpg.Pool):
+        self.pool = pool
+
+    @classmethod
+    async def create_pool(cls) -> asyncpg.Pool:
+        settings = get_settings()
+        pool = await asyncpg.create_pool(
+            host=settings.db_host,
+            port=settings.db_port,
+            database=settings.db_name,
+            user=settings.db_user,
+            password=settings.db_password,
+            min_size=2,
+            max_size=10,
+        )
+        return pool
+
+    @classmethod
+    @asynccontextmanager
+    async def connection(cls):
+        pool = await cls.create_pool()
+        async with pool.acquire() as conn:
+            yield conn
+        await pool.close()
+
+    async def init_schema(self):
+        """Run initial SQL schema from migrations."""
+        import os
+        migrations_path = os.path.join(
+            os.path.dirname(__file__), "..", "..", "migrations", "init.sql"
+        )
+        async with self.connection() as conn:
+            with open(migrations_path) as f:
+                await conn.execute(f.read())
+
+    async def upsert_document(
+        self, filename: str, doc_id: str | None, mime_type: str,
+        file_path: str, status: str, page_count: int,
+        full_text: str, metadata: dict
+    ) -> str:
+        async with self.connection() as conn:
+            pk = doc_id or str(uuid.uuid4())
+            await conn.execute(
+                """INSERT INTO documents (id, filename, doc_id, mime_type, file_path,
+                   status, page_count, full_text, metadata)
+                   VALUES ($1::uuid, $2, $1::text, $3, $4, $5, $6, $7, $8::jsonb)
+                   ON CONFLICT (filename) DO UPDATE SET
+                   status=EXCLUDED.status, page_count=EXCLUDED.page_count,
+                   full_text=EXCLUDED.full_text, metadata=EXCLUDED.metadata,
+                   updated_at=NOW()""",
+                str(pk), filename, mime_type, file_path,
+                status, page_count, full_text, json.dumps(metadata),
+            )
+            return pk
+
+    async def get_document(self, doc_id: str | uuid.UUID) -> dict | None:
+        async with self.connection() as conn:
+            row = await conn.fetchrow(
+                "SELECT * FROM documents WHERE id=$1", doc_id
+            )
+            return dict(row) if row else None
+
+    async def list_documents(self, status: str | None = None) -> list[dict]:
+        async with self.connection() as conn:
+            if status:
+                rows = await conn.fetch(
+                    "SELECT * FROM documents WHERE $1=ANY(string_to_array(status, ',')) ORDER BY created_at DESC",
+                    status,
+                )
+            else:
+                rows = await conn.fetch("SELECT * FROM documents ORDER BY created_at DESC")
+            return [dict(r) for r in rows]
+
+    async def chunk_document(
+        self, doc_id: str, content: str, polygon: dict | None,
+        page_num: int, block_index: int, chunk_type: str = "text"
+    ) -> str:
+        vec_str = "[" + ",".join(str(x) for x in self._extract_vector(content)) + "]"
+        chunk_id = str(uuid.uuid4())
+        async with self.connection() as conn:
+            await conn.execute(
+                """INSERT INTO chunks (id, doc_id, content, vector, page_num,
+                   block_index, polygon, chunk_type)
+                   VALUES ($1, $2, $3, $4::vector, $5, $6, $7::jsonb, $8)""",
+                chunk_id, str(doc_id), content, vec_str, page_num,
+                block_index, json.dumps(polygon) if polygon else None, chunk_type,
+            )
+            return chunk_id
+
+    async def batch_chunk(self, doc_id: str, data: list[dict]):
+        """Insert multiple chunks at once."""
+        vec_data = []
+        for d in data:
+            content = d.get("content", "")
+            vec_str = "[" + ",".join(str(x) for x in self._extract_vector(content)) + "]"
+            vec_data.append((
+                str(doc_id), content, vec_str,
+                d.get("page_num", 0), d.get("block_index", 0),
+                json.dumps(d.get("polygon")) if d.get("polygon") else None,
+                d.get("chunk_type", "text"),
+            ))
+        async with self.connection() as conn:
+            await conn.executemany(
+                """INSERT INTO chunks (doc_id, content, vector, page_num,
+                   block_index, polygon, chunk_type)
+                   VALUES ($1, $2, $3::vector, $4, $5, $6::jsonb, $7)""",
+                vec_data
+            )
+
+    async def vector_search(
+        self, query_vector: list[float], doc_id: str | None = None,
+        limit: int = 20, min_score: float = 0.0
+    ) -> list[dict]:
+        """Find similar chunks using pgvector cosine similarity."""
+        query_vec = "[" + ",".join(str(x) for x in query_vector) + "]"
+        async with self.connection() as conn:
+            if doc_id:
+                rows = await conn.fetch(
+                    """SELECT id, content, doc_id, page_num, polygon,
+                       (1 - (vector <-> $4::vector) / 2) as similarity
+                       FROM chunks WHERE doc_id = $1
+                       AND (1 - (vector <-> $4::vector) / 2) >= $3
+                       ORDER BY vector <-> $4 LIMIT $2""",
+                    str(doc_id), limit, min_score, query_vec,
+                )
+            else:
+                rows = await conn.fetch(
+                    """SELECT id, content, doc_id, page_num, polygon,
+                       (1 - (vector <-> $3::vector) / 2) as similarity
+                       FROM chunks
+                       WHERE (1 - (vector <-> $3::vector) / 2) >= $2
+                       ORDER BY vector <-> $3 LIMIT $1""",
+                    limit, min_score, query_vec,
+                )
+            return [dict(r) for r in rows]
+
+    async def semantic_search(
+        self, query_text: str, limit: int = 20, min_score: float = 0.0
+    ) -> list[dict]:
+        """Semantic search by embedding the query text."""
+        from app.core import get_embedding
+        query_vec = get_embedding(query_text)
+        return await self.vector_search(query_vec, limit=limit, min_score=min_score)
+
+    async def search_chunks_text(
+        self, query: str, doc_id: str | None = None, limit: int = 20
+    ) -> list[dict]:
+        """Text-based search using trigram similarity."""
+        async with self.connection() as conn:
+            if doc_id:
+                rows = await conn.fetch(
+                    """SELECT id, content, doc_id, page_num,
+                       ts_rank(to_tsvector('simple', content),
+                       plainto_tsquery('simple', $4)) as rank
+                       FROM chunks WHERE doc_id = $1
+                       AND to_tsvector('simple', content) @@ plainto_tsquery('simple', $4)
+                       ORDER BY rank DESC LIMIT $2""",
+                    str(doc_id), limit, query,
+                )
+            else:
+                rows = await conn.fetch(
+                    """SELECT id, content, doc_id, page_num,
+                       ts_rank(to_tsvector('simple', content),
+                       plainto_tsquery('simple', $3)) as rank
+                       FROM chunks
+                       WHERE to_tsvector('simple', content) @@ plainto_tsquery('simple', $3)
+                       ORDER BY rank DESC LIMIT $1""",
+                    limit, query,
+                )
+            return [dict(r) for r in rows]
+
+    async def store_memory(
+        self, session_id: str, content: str, memory_type: str = "fact",
+        importance: int = 3, source_doc_id: str | None = None
+    ) -> str:
+        vec = self._extract_vector(content)
+        mem_id = str(uuid.uuid4())
+        async with self.connection() as conn:
+            await conn.execute(
+                """INSERT INTO memories (id, session_id, content, vector,
+                   memory_type, importance, source_doc_id)
+                   VALUES ($1, $2, $3, $4, $5, $6, $7)""",
+                mem_id, str(session_id), content, vec, memory_type,
+                importance, str(source_doc_id) if source_doc_id else None,
+            )
+            return mem_id
+
+    async def get_memories(
+        self, session_id: str, limit: int = 50
+    ) -> list[dict]:
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                "SELECT * FROM memories WHERE session_id=$1 ORDER BY importance DESC, created_at DESC LIMIT $2",
+                str(session_id), limit,
+            )
+            return [dict(r) for r in rows]
+
+    async def memory_similarity_search(
+        self, query_text: str, limit: int = 10
+    ) -> list[dict]:
+        from app.core import get_embedding
+        query_vec = get_embedding(query_text)
+        query_str = "[" + ",".join(str(x) for x in query_vec) + "]"
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                """SELECT id, content, memory_type, importance,
+                   (1 - (vector <-> $2::vector) / 2) as similarity
+                   FROM memories ORDER BY vector <-> $2 LIMIT $1""",
+                limit, query_str,
+            )
+            return [dict(r) for r in rows]
+
+    async def store_finding(
+        self, session_id: str, question: str, answer: str,
+        summary: str, agent_name: str, confidence: float,
+        relevant_chunks: list | None = None
+    ) -> str:
+        question_vec_str = "[" + ",".join(str(x) for x in self._extract_vector(answer)) + "]"
+        finding_id = str(uuid.uuid4())
+        async with self.connection() as conn:
+            await conn.execute(
+                """INSERT INTO findings (id, session_id, question, answer,
+                   summary, vector, relevant_chunks, agent_name, confidence)
+                   VALUES ($1, $2, $3, $4, $5, $6::vector, $7::jsonb, $8, $9)""",
+                finding_id, str(session_id), question, answer,
+                summary, question_vec_str, json.dumps(relevant_chunks or []),
+                agent_name, confidence,
+            )
+            return finding_id
+
+    async def get_findings(self, session_id: str) -> list[dict]:
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                "SELECT * FROM findings WHERE session_id=$1 ORDER BY created_at DESC",
+                str(session_id),
+            )
+            return [dict(r) for r in rows]
+
+    async def create_session(self, query: str) -> str:
+        session_id = str(uuid.uuid4())
+        async with self.connection() as conn:
+            await conn.execute(
+                "INSERT INTO research_sessions (id, query, status, documents, findings) VALUES ($1, $2, $3, '[]', '[]')",
+                session_id, query, "running",
+            )
+        return session_id
+
+    async def update_session(
+        self, session_id: str, status: str | None = None,
+        documents: list | dict | None = None, findings: list | dict | None = None
+    ):
+        async with self.connection() as conn:
+            if status:
+                if documents or findings:
+                    await conn.execute(
+                        "UPDATE research_sessions SET status=$1, documents=$2::jsonb, findings=$3::jsonb, completed_at=NOW() WHERE id=$4",
+                        status, json.dumps(documents or []), json.dumps(findings or []),
+                        session_id,
+                    )
+                else:
+                    await conn.execute(
+                        "UPDATE research_sessions SET status=$1 WHERE id=$2",
+                        status, session_id,
+                    )
+            else:
+                await conn.execute(
+                    "UPDATE research_sessions SET documents=$1::jsonb, findings=$2::jsonb WHERE id=$3",
+                    json.dumps(documents or []), json.dumps(findings or []),
+                    session_id,
+                )
+
+    async def get_session(self, session_id: str) -> dict | None:
+        async with self.connection() as conn:
+            row = await conn.fetchrow(
+                "SELECT * FROM research_sessions WHERE id=$1", session_id
+            )
+            return dict(row) if row else None
+
+    async def get_chunk(self, chunk_id: str) -> dict | None:
+        async with self.connection() as conn:
+            row = await conn.fetchrow(
+                "SELECT * FROM chunks WHERE id=$1", chunk_id
+            )
+            return dict(row) if row else None
+
+    @staticmethod
+    def _extract_vector(text: str) -> list[float]:
+        """Generate a lightweight embedding vector directly (no external call for speed)."""
+        # Use a fast hash-based feature vector as fallback
+        # In production this calls the server; locally we use a fast approximation
+        import hashlib
+        feature_dim = 4096
+        vec = np.zeros(feature_dim, dtype=np.float32)
+        # Create deterministic features from character trigrams
+        trigrams = [text[i:i+3] for i in range(len(text)-2)]
+        for i, tri in enumerate(trigrams):
+            hash_val = hash(tri) & 0xFFFFFFFF
+            # convert to signed 32-bit
+            if hash_val >= 0x80000000:
+                hash_val -= 0x100000000
+            start_idx = (hash_val % feature_dim)
+            end_idx = min(start_idx + 5, feature_dim)
+            for j, byte in enumerate(hash_val.to_bytes(4, "big", signed=True)):
+                idx = (start_idx + j) % feature_dim
+                vec[idx] = (byte / 127.0) * np.sin(i * 0.1)
+        norm = np.linalg.norm(vec)
+        if norm > 0:
+            vec = vec / norm
+        return vec.tolist()
+
+    async def get_doc_chunks(self, doc_id: str) -> list[dict]:
+        """Get all chunks for a document."""
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                "SELECT * FROM chunks WHERE doc_id=$1 ORDER BY page_num, block_index",
+                str(doc_id),
+            )
+            return [dict(r) for r in rows]
+
+    # ── Pipeline persistence ───────────────────────────────
+
+    async def save_pipeline_stage(
+        self, session_id: str, stage: str, output: str, state: dict
+    ) -> str:
+        """Persist intermediate pipeline stage (triage/evidence/synthesis)."""
+        stage_id = str(uuid.uuid4())
+        async with self.connection() as conn:
+            await conn.execute(
+                """INSERT INTO pipeline_stages (id, session_id, stage, output, state)
+                   VALUES ($1, $2, $3, $4, $5::jsonb)""",
+                stage_id, str(session_id), stage, output, json.dumps(state),
+            )
+        return stage_id
+
+    async def get_pipeline_stages(self, session_id: str) -> list[dict]:
+        """Get all pipeline stages for a session in order."""
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                "SELECT * FROM pipeline_stages WHERE session_id=$1 ORDER BY stage, created_at",
+                str(session_id),
+            )
+            return [dict(r) for r in rows]
+
+    async def save_structured_evidence(
+        self, session_id: str,
+        rows: list[dict]
+    ) -> list[str]:
+        """Save evidence-extraction rows with full traceability fields."""
+        ids = []
+        for row in rows:
+            finding_id = str(uuid.uuid4())
+            async with self.connection() as conn:
+                await conn.execute(
+                    """INSERT INTO findings (id, session_id, question, answer, summary,
+                       agent_name, confidence, relevant_chunks)
+                       VALUES ($1, $2, $3, $4, $5, $6, $7, $8)""",
+                    finding_id, str(session_id),
+                    row.get("topic", ""),
+                    row.get("evidence", ""),
+                    row.get("description", ""),
+                    "evidence_extraction",
+                    {"High": 0.9, "Medium": 0.6, "Low": 0.3}.get(row.get("confidence", "Medium"), 0.6),
+                    json.dumps({
+                        "evidence_type": row.get("evidence_type"),
+                        "trace_ref": row.get("trace_ref"),
+                        "review_needed": row.get("review_needed", False),
+                        "confidence": row.get("confidence"),
+                        "source_doc": row.get("source_doc"),
+                    }),
+                )
+            ids.append(finding_id)
+        return ids
+
+    async def get_structured_findings(self, session_id: str) -> list[dict]:
+        """Get structured evidence findings for a session."""
+        async with self.connection() as conn:
+            rows = await conn.fetch(
+                """SELECT f.*, f.relevant_chunks::jsonb as meta
+                   FROM findings f
+                   WHERE f.session_id=$1 AND f.agent_name='evidence_extraction'
+                   ORDER BY f.created_at""",
+                str(session_id),
+            )
+            return [dict(r) for r in rows]
+
+
+db: Database | None = None