Add session persistence, sharing, web search grounding

agents/research.py

@@ -2,12 +2,19 @@
 
 Constitutional role: Strictly neutral. Authorship is hidden from adversarial agents.
 The research package is the ONLY shared context between Prosecutor and Defense.
+
+Grounding: When available, the agent performs lightweight web retrieval via
+DuckDuckGo Instant Answers and (optionally) Tavily Search API to ground
+claims in current factual data rather than relying solely on LLM training data.
 """
 
+import asyncio
 import json
 import logging
+import os
 from typing import Callable, Optional
 
+import httpx
 from langchain_groq import ChatGroq
 from langchain_core.messages import SystemMessage, HumanMessage
 
@@ -16,6 +23,71 @@ from models.schemas import StreamEvent
 
 logger = logging.getLogger(__name__)
 
+
+# ---------------------------------------------------------------------------
+# Lightweight web retrieval for factual grounding
+# ---------------------------------------------------------------------------
+
+async def _web_search_grounding(query: str, max_results: int = 3) -> list[str]:
+    """Retrieve web search snippets to ground LLM research in current facts.
+
+    Strategy:
+    1. Try Tavily Search API if TAVILY_API_KEY is set (best quality).
+    2. Fall back to DuckDuckGo Instant Answers API (no key required).
+    3. Return empty list on failure — research proceeds with LLM-only.
+    """
+    snippets = []
+
+    # Try Tavily first (higher quality, structured results)
+    tavily_key = os.getenv("TAVILY_API_KEY")
+    if tavily_key:
+        try:
+            async with httpx.AsyncClient(timeout=8.0) as client:
+                resp = await client.post(
+                    "https://api.tavily.com/search",
+                    json={
+                        "api_key": tavily_key,
+                        "query": query,
+                        "max_results": max_results,
+                        "search_depth": "basic",
+                    },
+                )
+                if resp.status_code == 200:
+                    data = resp.json()
+                    for r in data.get("results", [])[:max_results]:
+                        snippet = r.get("content", "")[:300]
+                        source = r.get("url", "")
+                        if snippet:
+                            snippets.append(f"{snippet} (source: {source})")
+                    if snippets:
+                        logger.info("Tavily grounding: %d snippets for '%s'", len(snippets), query[:50])
+                        return snippets
+        except Exception as e:
+            logger.warning("Tavily search failed, falling back to DuckDuckGo: %s", e)
+
+    # Fallback: DuckDuckGo Instant Answers (no API key needed)
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            resp = await client.get(
+                "https://api.duckduckgo.com/",
+                params={"q": query, "format": "json", "no_html": "1", "skip_disambig": "1"},
+            )
+            if resp.status_code == 200:
+                data = resp.json()
+                # Abstract text (Wikipedia-sourced summary)
+                if data.get("AbstractText"):
+                    snippets.append(data["AbstractText"][:400])
+                # Related topics
+                for topic in data.get("RelatedTopics", [])[:max_results]:
+                    if isinstance(topic, dict) and topic.get("Text"):
+                        snippets.append(topic["Text"][:200])
+                if snippets:
+                    logger.info("DuckDuckGo grounding: %d snippets for '%s'", len(snippets), query[:50])
+    except Exception as e:
+        logger.warning("DuckDuckGo search failed: %s", e)
+
+    return snippets
+
 RESEARCH_SYSTEM_PROMPT = """You are a neutral research analyst producing anonymous briefing material for an adversarial review process.
 
 CONSTITUTIONAL DIRECTIVE: Remain strictly neutral. Do not advocate for or against any outcome.
@@ -98,24 +170,42 @@ class ResearchAgent:
         ]
 
         try:
+            # Phase 1: Web retrieval for factual grounding
+            if stream_callback:
+                await stream_callback(StreamEvent(
+                    event_type="research_start", agent="research",
+                    content="Retrieving current web data for factual grounding...\n",
+                ))
+
+            web_snippets = await _web_search_grounding(decision_question)
+
+            if web_snippets and stream_callback:
+                await stream_callback(StreamEvent(
+                    event_type="research_start", agent="research",
+                    content=f"Found {len(web_snippets)} grounding sources. Analyzing...\n",
+                ))
+
+            # Inject web grounding into the prompt so LLM has current facts
+            if web_snippets:
+                grounding_text = "\n".join(f"- {s}" for s in web_snippets)
+                messages.append(HumanMessage(
+                    content=f"Web-retrieved grounding data (use these current facts to supplement your analysis):\n{grounding_text}"
+                ))
+
+            # Phase 2: LLM analysis with grounding context
             thinking_phases = [
                 "Scanning market landscape and competitive environment...",
-                "Gathering relevant data points and statistics...",
-                "Analyzing historical precedents and case studies...",
-                "Identifying key stakeholders and risk factors...",
-                "Compiling research synthesis...",
+                "Analyzing data points and historical precedents...",
+                "Identifying stakeholders and risk factors...",
+                "Compiling grounded research synthesis...",
             ]
 
             for phase in thinking_phases:
                 if stream_callback:
-                    await stream_callback(
-                        StreamEvent(
-                            event_type="research_start",
-                            agent="research",
-                            content=phase + "\n",
-                        )
-                    )
-                    import asyncio
+                    await stream_callback(StreamEvent(
+                        event_type="research_start", agent="research",
+                        content=phase + "\n",
+                    ))
                     await asyncio.sleep(0.3)
 
             response = await self.llm.ainvoke(messages)

api/routes.py

@@ -1,10 +1,13 @@
 """FastAPI routes for the Verdict API — REST + WebSocket endpoints."""
 
 import asyncio
+import hashlib
 import json
 import logging
+import os
 import uuid
 from datetime import datetime
+from pathlib import Path
 from typing import Optional, Literal
 
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException
@@ -20,9 +23,59 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api/verdict")
 
-# In-memory session store
+# ---------------------------------------------------------------------------
+# Session persistence — JSON file store with in-memory cache
+# ---------------------------------------------------------------------------
+
+SESSION_DIR = Path(os.getenv("SESSION_DIR", "data/sessions"))
+SESSION_DIR.mkdir(parents=True, exist_ok=True)
+
+# In-memory session cache backed by JSON file persistence
 sessions: dict[str, dict] = {}
 
+
+def _persist_session(session_id: str, session: dict) -> None:
+    """Write session to disk as JSON for persistence across restarts."""
+    try:
+        path = SESSION_DIR / f"{session_id}.json"
+        with open(path, "w") as f:
+            json.dump(session, f, default=str)
+    except Exception as e:
+        logger.warning("Failed to persist session %s: %s", session_id, e)
+
+
+def _load_session(session_id: str) -> Optional[dict]:
+    """Load a session from disk if not in memory cache."""
+    path = SESSION_DIR / f"{session_id}.json"
+    if path.exists():
+        try:
+            with open(path) as f:
+                session = json.load(f)
+            sessions[session_id] = session  # Warm the cache
+            return session
+        except Exception as e:
+            logger.warning("Failed to load session %s: %s", session_id, e)
+    return None
+
+
+def _get_session(session_id: str) -> Optional[dict]:
+    """Get session from cache or disk."""
+    if session_id in sessions:
+        return sessions[session_id]
+    return _load_session(session_id)
+
+
+def _load_all_sessions() -> None:
+    """Load all persisted sessions into memory on startup."""
+    for path in SESSION_DIR.glob("*.json"):
+        sid = path.stem
+        if sid not in sessions:
+            _load_session(sid)
+
+
+# Load persisted sessions on module import
+_load_all_sessions()
+
 # Valid output formats and their descriptions
 OUTPUT_FORMATS = {
     "executive": "High-level summary for executives and decision-makers",
@@ -163,6 +216,7 @@ async def start_verdict(request: StartRequest):
         "created_at": datetime.utcnow().isoformat(),
     }
     sessions[decision.id] = session
+    _persist_session(decision.id, session)
 
     logger.info(
         "Session created: %s — %s [format=%s, domain=%s]",
@@ -218,7 +272,7 @@ async def get_history():
 @router.get("/{session_id}/status")
 async def get_status(session_id: str):
     """Get the current status of a verdict session."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
 
@@ -233,7 +287,7 @@ async def get_status(session_id: str):
 @router.get("/{session_id}/result")
 async def get_result(session_id: str):
     """Get the complete result of a verdict session."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
 
@@ -257,7 +311,7 @@ async def stream_verdict(websocket: WebSocket, session_id: str):
     await websocket.accept()
     logger.info("WebSocket connected: %s", session_id)
 
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         await websocket.send_json({"error": "Session not found"})
         await websocket.close()
@@ -298,6 +352,7 @@ async def stream_verdict(websocket: WebSocket, session_id: str):
             }
 
             session["status"] = "complete"
+            _persist_session(session_id, session)
         except Exception as e:
             logger.error("Pipeline failed: %s", str(e))
             session["status"] = "error"
@@ -348,7 +403,7 @@ async def stream_verdict(websocket: WebSocket, session_id: str):
 @router.get("/{session_id}/export/markdown")
 async def export_markdown(session_id: str):
     """Export the verdict session as a markdown report."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     if not session.get("result"):
@@ -361,7 +416,7 @@ async def export_markdown(session_id: str):
 @router.get("/{session_id}/export/json")
 async def export_json(session_id: str):
     """Export the verdict session as structured JSON."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     if not session.get("result"):
@@ -374,7 +429,7 @@ async def export_json(session_id: str):
 @router.get("/{session_id}/export/pdf")
 async def export_pdf(session_id: str):
     """Export the verdict session as a formatted PDF report."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     if not session.get("result"):
@@ -391,7 +446,7 @@ async def export_pdf(session_id: str):
 @router.get("/{session_id}/export/docx")
 async def export_docx(session_id: str):
     """Export the verdict session as a formatted DOCX report."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     if not session.get("result"):
@@ -417,7 +472,7 @@ class FollowUpRequest(BaseModel):
 @router.post("/{session_id}/followup")
 async def followup_question(session_id: str, request: FollowUpRequest):
     """Ask a follow-up question about the verdict session."""
-    session = sessions.get(session_id)
+    session = _get_session(session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     if not session.get("result"):
@@ -467,3 +522,78 @@ Tailor your response to the {output_format} format and {domain} domain."""),
         return {"answer": response.content, "session_id": session_id}
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Failed to generate follow-up: {str(e)}")
+
+
+# ---------------------------------------------------------------------------
+# Verdict sharing endpoint
+# ---------------------------------------------------------------------------
+
+
+def _generate_share_token(session_id: str) -> str:
+    """Generate a short, URL-safe share token from a session ID."""
+    return hashlib.sha256(session_id.encode()).hexdigest()[:12]
+
+
+@router.get("/{session_id}/share")
+async def create_share_link(session_id: str):
+    """Generate a shareable link for a completed verdict session.
+
+    Returns a short share token that can be used to retrieve the session
+    results without needing the full session ID.
+    """
+    session = _get_session(session_id)
+    if not session:
+        raise HTTPException(status_code=404, detail="Session not found")
+    if not session.get("result"):
+        raise HTTPException(status_code=202, detail="Session not complete")
+
+    token = _generate_share_token(session_id)
+
+    # Store mapping from share token to session ID
+    session["share_token"] = token
+    _persist_session(session_id, session)
+
+    return {
+        "share_token": token,
+        "share_url": f"/shared/{token}",
+        "session_id": session_id,
+    }
+
+
+@router.get("/shared/{share_token}")
+async def get_shared_verdict(share_token: str):
+    """Retrieve a verdict session via its share token.
+
+    This allows anyone with the share link to view the verdict results
+    without needing the original session ID.
+    """
+    # Search for session with matching share token
+    for sid, session in sessions.items():
+        if session.get("share_token") == share_token:
+            if not session.get("result"):
+                raise HTTPException(status_code=202, detail="Session not complete")
+            return {
+                "session_id": sid,
+                "question": session["decision"].get("question", ""),
+                "domain": session.get("domain", "business"),
+                "output_format": session.get("output_format", "executive"),
+                "result": session["result"],
+            }
+
+    # Also check persisted sessions on disk
+    for path in SESSION_DIR.glob("*.json"):
+        try:
+            with open(path) as f:
+                session = json.load(f)
+            if session.get("share_token") == share_token:
+                return {
+                    "session_id": path.stem,
+                    "question": session.get("decision", {}).get("question", ""),
+                    "domain": session.get("domain", "business"),
+                    "output_format": session.get("output_format", "executive"),
+                    "result": session.get("result"),
+                }
+        except Exception:
+            continue
+
+    raise HTTPException(status_code=404, detail="Shared verdict not found")