feat(jobs): persist async batch job state in PostgreSQL

- Add jobs table to database schema (job_id, status, progress, result_json, etc.)
- Add DatabaseClient methods: create_job, update_job, get_job, list_jobs
- Add mark_stale_jobs_failed() called at startup to handle interrupted jobs
- Refactor _run_batch_job and job endpoints to read/write from PostgreSQL
- Remove in-memory _jobs dict; job state now survives API restarts
- Update init_database.py to list all tables in output

Closes leeworks-agents/SPARC#8

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

ROADMAP.md

@@ -0,0 +1,122 @@
+# SPARC Roadmap
+
+Semiconductor Patent & Analytics Report Core -- development priorities.
+
+## Current State
+
+SPARC is a patent analysis platform with a working end-to-end pipeline:
+Python/FastAPI backend, React/TypeScript frontend, PostgreSQL for persistence
+and caching, Docker Compose for local development, and Gitea Actions CI/CD for
+image builds. Core features (patent retrieval via SerpAPI, PDF parsing, LLM
+analysis via OpenRouter/Claude, batch processing, JWT authentication, analytics
+dashboard) are all implemented and functional.
+
+---
+
+## P1 -- High Priority
+
+These items address correctness, security, and reliability gaps that should be
+resolved before broader production use.
+
+### Security hardening
+
+- **Rotate default JWT secret.** `auth.py` ships a fallback
+  `sparc-secret-key-change-in-production` that will be used if `JWT_SECRET` is
+  unset. Add a startup check that refuses to start with the default secret in
+  non-development environments.
+- **CORS allow-origins are hardcoded.** `api.py` only permits
+  `localhost:3000` and `localhost:5173`. Make the allowed origins configurable
+  via environment variable so the dashboard works when deployed behind a real
+  domain.
+- **Database credentials in docker-compose.yml.** The compose file embeds
+  `postgres:postgres` in plain text. Reference a `.env` file or Docker secrets
+  instead.
+
+### Error handling and resilience
+
+- **`get_db_client()` in `auth.py` creates a new `DatabaseClient` on every
+  call.** This bypasses the connection pool and can exhaust database
+  connections under load. Refactor to share a single pooled client.
+- **`_jobs` dict is in-memory only.** Job state is lost on API restart. Persist
+  job status in PostgreSQL or Redis so async batch results survive restarts.
+- **No rate limiting on auth endpoints.** `/auth/login` and `/auth/register`
+  are unprotected against brute-force or abuse. Add rate limiting middleware.
+
+### Test coverage for auth and admin
+
+- The existing API tests (`tests/test_api.py`) bypass authentication entirely.
+  Add tests that exercise the JWT flow: registration, login, protected-route
+  access, token refresh, and admin-only endpoints.
+
+---
+
+## P2 -- Medium Priority
+
+Improvements to usability, performance, and developer experience.
+
+### Backend
+
+- **Add structured logging.** Replace `print()` calls throughout `analyzer.py`,
+  `serp_api.py`, and `llm.py` with Python `logging` so log levels and
+  formatting are consistent.
+- **Make LLM model configurable.** `llm.py` hardcodes
+  `anthropic/claude-3.5-sonnet`. Accept a `MODEL` environment variable to allow
+  switching models without code changes.
+- **SERP cache TTL is hardcoded to 24 hours.** Expose `SERP_CACHE_TTL_HOURS`
+  as an environment variable in `config.py`.
+- **Patent PDF storage.** PDFs are saved to a local `patents/` directory. For
+  containerized deployments, consider object storage (S3/MinIO) or at minimum
+  document the volume mount requirement more prominently.
+- **`analyze_single_patent` assumes local file path.** The method constructs
+  `patents/{patent_id}.pdf` and reads from disk, but does not download the PDF
+  first. Either integrate the download step or document the prerequisite.
+- **`Patent.patent_id` typed as `int` in `types.py` but used as `str`
+  everywhere.** Fix the type annotation to `str`.
+
+### Frontend
+
+- **No loading/error states on several pages.** The Batch and Analytics pages
+  would benefit from skeleton loaders and user-friendly error messages.
+- **No dark mode.** Tailwind is configured but no dark variant is applied.
+- **Missing `package-lock.json` or `pnpm-lock.yaml`.** The frontend has no
+  lockfile committed, leading to non-reproducible builds.
+
+### CI/CD
+
+- **No test stage in the Gitea Actions workflow.** `build.yaml` builds and
+  pushes images but never runs `pytest`. Add a test job that gates the build.
+- **No linting or type checking.** Add `ruff` (Python) and `tsc --noEmit`
+  (TypeScript) to CI.
+
+---
+
+## P3 -- Nice to Have
+
+Lower-urgency enhancements and future features.
+
+- **Export analysis reports.** Allow users to download analysis results as PDF
+  or CSV from the dashboard.
+- **Comparison view.** Side-by-side comparison of two companies' patent
+  portfolios.
+- **Scheduled/recurring analysis.** Periodically re-analyze tracked companies
+  and alert on significant changes.
+- **Webhook/notification support.** Send alerts (Slack, Discord, email) when
+  batch jobs complete or when a company's innovation score changes
+  significantly.
+- **Multi-model support.** Let users choose between LLM providers per analysis
+  (e.g., GPT-4o, Gemini, Claude) and compare outputs.
+- **Patent trend charts.** Visualize patent filing frequency and technology
+  category distribution over time in the Analytics page.
+- **API pagination.** The `/analyze/batch` and `/jobs` endpoints could benefit
+  from cursor-based pagination for large result sets.
+- **OpenAPI client generation.** Auto-generate the TypeScript API client from
+  the FastAPI OpenAPI spec to keep frontend types in sync.
+
+---
+
+## Infrastructure and Deployment
+
+Kubernetes manifests, Helm charts, and cluster-level concerns (MetalLB,
+storage, FluxCD sync) are tracked in the
+[Talos](https://10.0.1.10/leeworks-agents/Talos) repository. File
+infrastructure-related issues there, not here.

SPARC/analyzer.py

@@ -4,26 +4,33 @@ This module ties together patent retrieval, parsing, and LLM analysis
 to provide company performance estimation based on patent portfolios.
 """
 
+import hashlib
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Callable
 
+from SPARC import config
+from SPARC.database import DatabaseClient
 from SPARC.serp_api import SERP
 from SPARC.llm import LLMAnalyzer
-from SPARC.types import Patent, CompanyAnalysisResult, BatchAnalysisResult
+from SPARC.types import Patent, Patents, CompanyAnalysisResult, BatchAnalysisResult
 
 
 class CompanyAnalyzer:
     """Orchestrates end-to-end company performance analysis via patents."""
 
-    def __init__(self, openrouter_api_key: str | None = None):
+    def __init__(self, openrouter_api_key: str | None = None, db_client: DatabaseClient | None = None):
         """Initialize the company analyzer.
 
         Args:
           openrouter_api_key: Optional OpenRouter API key. If None, loads from config.
+          db_client: Optional DatabaseClient for patent caching. Created automatically if None.
         """
         self.llm_analyzer = LLMAnalyzer(api_key=openrouter_api_key)
+        self.db = db_client or DatabaseClient(config.database_url)
+        self.db.connect()
+        self.db.initialize_schema()
 
-    def analyze_company(self, company_name: str) -> str:
+    def analyze_company(self, company_name: str, patents: "Patents | None" = None) -> str:
         """Analyze a company's performance based on their patent portfolio.
 
         This is the main entry point that orchestrates the full pipeline:
@@ -35,40 +42,52 @@ class CompanyAnalyzer:
 
         Args:
           company_name: Name of the company to analyze
+          patents: Optional pre-fetched Patents result to avoid duplicate API calls
 
         Returns:
           Comprehensive analysis of company's innovation and performance outlook
         """
+        if patents is None:
+            # Check SERP query cache first
+            query_hash = hashlib.sha256(company_name.lower().encode()).hexdigest()
+            cached_ids = self.db.get_cached_serp_query(query_hash)
+            if cached_ids is not None:
+                print(f"Using cached SERP results for {company_name} ({len(cached_ids)} patents)")
+                patents = Patents(patents=[
+                    Patent(patent_id=pid, pdf_link="")
+                    for pid in cached_ids
+                ])
+            else:
                 print(f"Retrieving patents for {company_name}...")
                 patents = SERP.query(company_name)
+                # Cache the SERP results
+                if patents.patents:
+                    self.db.store_serp_query(
+                        company_name=company_name,
+                        query_hash=query_hash,
+                        patent_ids=[p.patent_id for p in patents.patents],
+                    )
 
         if not patents.patents:
             return f"No patents found for {company_name}"
 
         print(f"Found {len(patents.patents)} patents. Processing...")
 
-        # Download and parse each patent
+        # Download, parse, and minimize patents in parallel
         processed_patents = []
-        for idx, patent in enumerate(patents.patents, 1):
+        with ThreadPoolExecutor(max_workers=config.patent_thread_workers) as executor:
-            print(f"Processing patent {idx}/{len(patents.patents)}: {patent.patent_id}")
+            future_to_patent = {
-
+                executor.submit(self._process_single_patent, patent, company_name, self.db): patent
+                for patent in patents.patents
+            }
+            for future in as_completed(future_to_patent):
+                patent = future_to_patent[future]
                 try:
-                # Download PDF
+                    result = future.result()
-                patent = SERP.save_patents(patent)
+                    if result:
-
+                        processed_patents.append(result)
-                # Parse sections from PDF
-                sections = SERP.parse_patent_pdf(patent.pdf_path)
-
-                # Minimize for LLM (remove bloat)
-                minimized_content = SERP.minimize_patent_for_llm(sections)
-
-                processed_patents.append(
-                    {"patent_id": patent.patent_id, "content": minimized_content}
-                )
-
                 except Exception as e:
                     print(f"Warning: Failed to process {patent.patent_id}: {e}")
-                continue
 
         if not processed_patents:
             return f"Failed to process any patents for {company_name}"
@@ -113,6 +132,46 @@ class CompanyAnalyzer:
         except Exception as e:
             return f"Failed to analyze patent {patent_id}: {e}"
 
+    @staticmethod
+    def _process_single_patent(
+        patent: Patent,
+        company_name: str = "",
+        db: DatabaseClient | None = None,
+    ) -> dict | None:
+        """Download, parse, and minimize a single patent. Thread-safe.
+
+        Checks DB cache before downloading. Stores results after processing.
+
+        Returns:
+            Dict with patent_id and minimized content, or None on failure.
+        """
+        try:
+            # Check DB cache first
+            if db:
+                cached = db.get_cached_patent(patent.patent_id)
+                if cached and cached.get("minimized_content"):
+                    return {"patent_id": patent.patent_id, "content": cached["minimized_content"]}
+
+            # Full processing: download, parse, minimize
+            patent = SERP.save_patents(patent)
+            sections = SERP.parse_patent_pdf(patent.pdf_path)
+            minimized_content = SERP.minimize_patent_for_llm(sections)
+
+            # Store in DB cache
+            if db:
+                db.store_patent(
+                    patent_id=patent.patent_id,
+                    company_name=company_name,
+                    pdf_link=patent.pdf_link,
+                    raw_sections=sections,
+                    minimized_content=minimized_content,
+                )
+
+            return {"patent_id": patent.patent_id, "content": minimized_content}
+        except Exception as e:
+            print(f"Warning: Failed to process {patent.patent_id}: {e}")
+            return None
+
     def _analyze_company_safe(self, company_name: str) -> CompanyAnalysisResult:
         """Internal wrapper that catches exceptions and returns structured result.
 
@@ -123,11 +182,14 @@ class CompanyAnalyzer:
             CompanyAnalysisResult with success/failure status
         """
         try:
-            patents = SERP.query(company_name)
+            # Delegate to analyze_company which handles SERP/patent caching
-            patent_count = len(patents.patents) if patents.patents else 0
-
             analysis = self.analyze_company(company_name)
 
+            # Determine patent count from cached SERP query
+            query_hash = hashlib.sha256(company_name.lower().encode()).hexdigest()
+            cached_ids = self.db.get_cached_serp_query(query_hash)
+            patent_count = len(cached_ids) if cached_ids else 0
+
             # Check if analysis indicates failure
             if analysis.startswith("No patents found") or analysis.startswith(
                 "Failed to process"

SPARC/api.py

@@ -114,8 +114,7 @@ class AnalyticsResponse(BaseModel):
     period_days: int
 
 
-# In-memory job storage (for demo; production would use Redis/DB)
+# Job counter for generating unique IDs (the actual state is in PostgreSQL)
-_jobs: dict[str, JobStatus] = {}
 _job_counter = 0
 
 
@@ -148,9 +147,19 @@ _analyzer: CompanyAnalyzer | None = None
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Initialize resources on startup."""
+    """Initialize resources on startup, clean up on shutdown."""
     global _analyzer
     _analyzer = CompanyAnalyzer()
+    # Mark any jobs that were running/pending before the restart as failed
+    from SPARC.database import DatabaseClient
+    _db = DatabaseClient(config.database_url)
+    _db.connect()
+    _db.initialize_schema()
+    stale = _db.mark_stale_jobs_failed()
+    if stale:
+        import logging
+        logging.getLogger(__name__).warning("Marked %d stale jobs as failed on startup", stale)
+    _db.close()
     yield
     # Cleanup if needed
     _analyzer = None
@@ -422,20 +431,52 @@ async def analyze_companies_batch(
     return _convert_batch_result(result)
 
 
+def _get_job_db() -> "DatabaseClient":
+    """Get a DatabaseClient for job persistence."""
+    from SPARC.database import DatabaseClient
+    db = DatabaseClient(config.database_url)
+    return db
+
+
+def _job_row_to_status(row: dict) -> JobStatus:
+    """Convert a database job row to a JobStatus model."""
+    import json as _json
+    result = None
+    if row.get("result_json"):
+        result_data = row["result_json"]
+        if isinstance(result_data, str):
+            result_data = _json.loads(result_data)
+        result = BatchAnalysisResponse(**result_data)
+    return JobStatus(
+        job_id=row["job_id"],
+        status=row["status"],
+        progress=row["progress"],
+        total_companies=row["total_companies"],
+        completed_companies=row["completed_companies"],
+        result=result,
+        error=row.get("error"),
+    )
+
+
 def _run_batch_job(job_id: str, companies: list[str], max_workers: int):
     """Background task for batch analysis."""
-    global _jobs, _analyzer
+    import json as _json
+    global _analyzer
+
+    db = _get_job_db()
 
     if not _analyzer:
-        _jobs[job_id].status = "failed"
+        db.update_job(job_id, status="failed", error="Analyzer not initialized")
-        _jobs[job_id].error = "Analyzer not initialized"
         return
 
-    _jobs[job_id].status = "running"
+    db.update_job(job_id, status="running")
 
     def progress_callback(company: str, completed: int, total: int):
-        _jobs[job_id].completed_companies = completed
+        db.update_job(
-        _jobs[job_id].progress = int((completed / total) * 100)
+            job_id,
+            completed_companies=completed,
+            progress=int((completed / total) * 100),
+        )
 
     try:
         result = _analyzer.analyze_companies(
@@ -443,12 +484,15 @@ def _run_batch_job(job_id: str, companies: list[str], max_workers: int):
             max_workers=max_workers,
             progress_callback=progress_callback,
         )
-        _jobs[job_id].status = "completed"
+        batch_response = _convert_batch_result(result)
-        _jobs[job_id].progress = 100
+        db.update_job(
-        _jobs[job_id].result = _convert_batch_result(result)
+            job_id,
+            status="completed",
+            progress=100,
+            result_json=_json.dumps(batch_response.model_dump(), default=str),
+        )
     except Exception as e:
-        _jobs[job_id].status = "failed"
+        db.update_job(job_id, status="failed", error=str(e))
-        _jobs[job_id].error = str(e)
 
 
 @app.post("/analyze/batch/async", response_model=JobStatus, tags=["Analysis"])
@@ -473,19 +517,14 @@ async def analyze_companies_async(
     _job_counter += 1
     job_id = f"job_{_job_counter}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
 
-    _jobs[job_id] = JobStatus(
+    db = _get_job_db()
-        job_id=job_id,
+    job_row = db.create_job(job_id=job_id, total_companies=len(request.companies))
-        status="pending",
-        progress=0,
-        total_companies=len(request.companies),
-        completed_companies=0,
-    )
 
     background_tasks.add_task(
         _run_batch_job, job_id, request.companies, request.max_workers
     )
 
-    return _jobs[job_id]
+    return _job_row_to_status(job_row)
 
 
 @app.get("/jobs/{job_id}", response_model=JobStatus, tags=["Jobs"])
@@ -501,10 +540,13 @@ async def get_job_status(
     Returns:
         Current job status including progress and results when complete
     """
-    if job_id not in _jobs:
+    db = _get_job_db()
+    job_row = db.get_job(job_id)
+
+    if not job_row:
         raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
 
-    return _jobs[job_id]
+    return _job_row_to_status(job_row)
 
 
 @app.get("/jobs", response_model=list[JobStatus], tags=["Jobs"])
@@ -525,12 +567,6 @@ async def list_jobs(
     Returns:
         List of job statuses
     """
-    jobs = list(_jobs.values())
+    db = _get_job_db()
-
+    job_rows = db.list_jobs(status=status, limit=limit)
-    if status:
+    return [_job_row_to_status(row) for row in job_rows]
-        jobs = [j for j in jobs if j.status == status]
-
-    # Return most recent first
-    jobs.sort(key=lambda j: j.job_id, reverse=True)
-
-    return jobs[:limit]

SPARC/config.py

@@ -26,6 +26,10 @@ use_cache = os.getenv("USE_CACHE", "true").lower() in ("true", "1", "yes")
 # This variable is kept for backwards compatibility but has no effect
 use_database = os.getenv("USE_DATABASE", "false").lower() in ("true", "1", "yes")
 
+# Patent search configuration
+patent_search_days = int(os.getenv("PATENT_SEARCH_DAYS", "90"))
+patent_thread_workers = int(os.getenv("PATENT_THREAD_WORKERS", "5"))
+
 # Root path for running behind a reverse proxy (e.g., "/api" when served at /api/)
 # This ensures OpenAPI docs work correctly when accessed via the proxy
 root_path = os.getenv("ROOT_PATH", "")

SPARC/database.py

@@ -1,9 +1,11 @@
 """Database client for storing and retrieving LLM messages and user authentication."""
 
+import contextlib
 import psycopg2
+from psycopg2.pool import ThreadedConnectionPool
 from psycopg2.extras import RealDictCursor
 from typing import Dict, List, Optional
-from datetime import datetime
+from datetime import datetime, timedelta
 import json
 import hashlib
 import bcrypt
@@ -12,24 +14,49 @@ import bcrypt
 class DatabaseClient:
     """Handles database operations for message storage and retrieval."""
 
-    def __init__(self, database_url: str):
+    def __init__(self, database_url: str, minconn: int = 2, maxconn: int = 10):
         """Initialize the database client.
 
         Args:
             database_url: PostgreSQL connection string
+            minconn: Minimum connections in the pool
+            maxconn: Maximum connections in the pool
         """
         self.database_url = database_url
+        self._pool: ThreadedConnectionPool | None = None
+        self._minconn = minconn
+        self._maxconn = maxconn
+        # Legacy single connection kept for backwards compatibility
         self.conn = None
 
+    def _ensure_pool(self):
+        """Create the connection pool if it doesn't exist yet."""
+        if self._pool is None or self._pool.closed:
+            self._pool = ThreadedConnectionPool(
+                self._minconn, self._maxconn, self.database_url
+            )
+
+    @contextlib.contextmanager
+    def get_conn(self):
+        """Check out a connection from the pool. Returns it on exit."""
+        self._ensure_pool()
+        conn = self._pool.getconn()
+        try:
+            yield conn
+        finally:
+            self._pool.putconn(conn)
+
     def connect(self):
-        """Establish database connection."""
+        """Establish database connection (legacy single-connection path)."""
         if not self.conn or self.conn.closed:
             self.conn = psycopg2.connect(self.database_url)
 
     def close(self):
-        """Close database connection."""
+        """Close database connection and pool."""
         if self.conn and not self.conn.closed:
             self.conn.close()
+        if self._pool and not self._pool.closed:
+            self._pool.closeall()
 
     def initialize_schema(self):
         """Create database tables if they don't exist."""
@@ -110,6 +137,60 @@ class DatabaseClient:
                 ON users(email)
             """)
 
+            # Create patents cache table
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS patents (
+                    patent_id VARCHAR(64) PRIMARY KEY,
+                    company_name VARCHAR(255),
+                    pdf_link TEXT,
+                    raw_sections JSONB,
+                    minimized_content TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
+            cursor.execute("""
+                CREATE INDEX IF NOT EXISTS idx_patents_company
+                ON patents(company_name)
+            """)
+
+            # Create SERP query cache table
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS serp_queries (
+                    id SERIAL PRIMARY KEY,
+                    company_name VARCHAR(255),
+                    query_hash VARCHAR(64) UNIQUE,
+                    result_patent_ids TEXT[],
+                    expires_at TIMESTAMP NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
+            cursor.execute("""
+                CREATE INDEX IF NOT EXISTS idx_serp_queries_hash
+                ON serp_queries(query_hash)
+            """)
+
+            # Create jobs table for persisting async batch job state
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS jobs (
+                    job_id VARCHAR(128) PRIMARY KEY,
+                    status VARCHAR(20) NOT NULL DEFAULT 'pending',
+                    progress INTEGER NOT NULL DEFAULT 0,
+                    total_companies INTEGER NOT NULL DEFAULT 0,
+                    completed_companies INTEGER NOT NULL DEFAULT 0,
+                    result_json JSONB,
+                    error TEXT,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
+            cursor.execute("""
+                CREATE INDEX IF NOT EXISTS idx_jobs_status
+                ON jobs(status)
+            """)
+
             self.conn.commit()
 
     @staticmethod
@@ -320,6 +401,212 @@ class DatabaseClient:
                 "period_days": days,
             }
 
+    # Patent Cache Methods
+
+    def get_cached_patent(self, patent_id: str) -> Optional[Dict]:
+        """Look up a cached patent by ID.
+
+        Returns:
+            Dict with raw_sections and minimized_content, or None.
+        """
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute(
+                    "SELECT * FROM patents WHERE patent_id = %s",
+                    (patent_id,),
+                )
+                row = cursor.fetchone()
+                return dict(row) if row else None
+
+    def store_patent(
+        self,
+        patent_id: str,
+        company_name: str,
+        pdf_link: str,
+        raw_sections: Dict,
+        minimized_content: str,
+    ) -> None:
+        """Store a processed patent in the cache."""
+        with self.get_conn() as conn:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO patents (patent_id, company_name, pdf_link, raw_sections, minimized_content)
+                    VALUES (%s, %s, %s, %s, %s)
+                    ON CONFLICT (patent_id) DO UPDATE SET
+                        raw_sections = EXCLUDED.raw_sections,
+                        minimized_content = EXCLUDED.minimized_content
+                    """,
+                    (patent_id, company_name, pdf_link, json.dumps(raw_sections), minimized_content),
+                )
+            conn.commit()
+
+    def get_cached_serp_query(self, query_hash: str) -> Optional[List[str]]:
+        """Look up cached SERP query results.
+
+        Returns:
+            List of patent IDs if cache hit and not expired, None otherwise.
+        """
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute(
+                    """
+                    SELECT result_patent_ids FROM serp_queries
+                    WHERE query_hash = %s AND expires_at > NOW()
+                    """,
+                    (query_hash,),
+                )
+                row = cursor.fetchone()
+                return row["result_patent_ids"] if row else None
+
+    def store_serp_query(
+        self,
+        company_name: str,
+        query_hash: str,
+        patent_ids: List[str],
+        ttl_hours: int = 24,
+    ) -> None:
+        """Store SERP query results in the cache."""
+        expires_at = datetime.now() + timedelta(hours=ttl_hours)
+        with self.get_conn() as conn:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO serp_queries (company_name, query_hash, result_patent_ids, expires_at)
+                    VALUES (%s, %s, %s, %s)
+                    ON CONFLICT (query_hash) DO UPDATE SET
+                        result_patent_ids = EXCLUDED.result_patent_ids,
+                        expires_at = EXCLUDED.expires_at
+                    """,
+                    (company_name, query_hash, patent_ids, expires_at),
+                )
+            conn.commit()
+
+    # Job Persistence Methods
+
+    def create_job(
+        self,
+        job_id: str,
+        total_companies: int,
+    ) -> Dict:
+        """Create a new job record.
+
+        Args:
+            job_id: Unique job identifier
+            total_companies: Number of companies in the batch
+
+        Returns:
+            Job dict
+        """
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO jobs (job_id, status, progress, total_companies, completed_companies)
+                    VALUES (%s, 'pending', 0, %s, 0)
+                    RETURNING *
+                    """,
+                    (job_id, total_companies),
+                )
+                job = cursor.fetchone()
+            conn.commit()
+            return dict(job)
+
+    def update_job(
+        self,
+        job_id: str,
+        status: Optional[str] = None,
+        progress: Optional[int] = None,
+        completed_companies: Optional[int] = None,
+        result_json: Optional[str] = None,
+        error: Optional[str] = None,
+    ) -> Optional[Dict]:
+        """Update a job's state.
+
+        Only non-None fields are updated.
+        """
+        updates = []
+        params = []
+        if status is not None:
+            updates.append("status = %s")
+            params.append(status)
+        if progress is not None:
+            updates.append("progress = %s")
+            params.append(progress)
+        if completed_companies is not None:
+            updates.append("completed_companies = %s")
+            params.append(completed_companies)
+        if result_json is not None:
+            updates.append("result_json = %s")
+            params.append(result_json)
+        if error is not None:
+            updates.append("error = %s")
+            params.append(error)
+
+        if not updates:
+            return self.get_job(job_id)
+
+        updates.append("updated_at = CURRENT_TIMESTAMP")
+        params.append(job_id)
+
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute(
+                    f"UPDATE jobs SET {', '.join(updates)} WHERE job_id = %s RETURNING *",
+                    params,
+                )
+                job = cursor.fetchone()
+            conn.commit()
+            return dict(job) if job else None
+
+    def get_job(self, job_id: str) -> Optional[Dict]:
+        """Get a job by ID."""
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute("SELECT * FROM jobs WHERE job_id = %s", (job_id,))
+                job = cursor.fetchone()
+                return dict(job) if job else None
+
+    def list_jobs(
+        self,
+        status: Optional[str] = None,
+        limit: int = 10,
+    ) -> List[Dict]:
+        """List jobs, optionally filtered by status."""
+        query = "SELECT * FROM jobs"
+        params: list = []
+        if status:
+            query += " WHERE status = %s"
+            params.append(status)
+        query += " ORDER BY created_at DESC LIMIT %s"
+        params.append(limit)
+
+        with self.get_conn() as conn:
+            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                cursor.execute(query, params)
+                return [dict(row) for row in cursor.fetchall()]
+
+    def mark_stale_jobs_failed(self) -> int:
+        """Mark any jobs in 'running' or 'pending' state as 'failed'.
+
+        Called at startup to clean up jobs that were interrupted by a restart.
+
+        Returns:
+            Number of jobs marked as failed.
+        """
+        with self.get_conn() as conn:
+            with conn.cursor() as cursor:
+                cursor.execute(
+                    """
+                    UPDATE jobs SET status = 'failed', error = 'Interrupted by server restart',
+                    updated_at = CURRENT_TIMESTAMP
+                    WHERE status IN ('running', 'pending')
+                    """
+                )
+                count = cursor.rowcount
+            conn.commit()
+            return count
+
     # User Authentication Methods
 
     @staticmethod

SPARC/serp_api.py

@@ -1,17 +1,20 @@
+import os
 import serpapi
 from SPARC import config
 import re
 import pdfplumber  # pip install pdfplumber
 import requests
+from datetime import datetime, timedelta
 from typing import Dict
 from SPARC.types import Patents, Patent
 
 class SERP:
-  def query(company: str) -> Patents:
+  def query(company: str, days_back: int = None) -> Patents:
     """Query Google Patents for a company's recent patents.
 
     Args:
         company: Name of the company to search for
+        days_back: Number of days to look back for patents (default from config)
 
     Returns:
         Patents object containing list of patents with PDF links
@@ -23,13 +26,19 @@ class SERP:
         patents with restricted access). The returned count may be lower
         than the requested number of results.
     """
+    if days_back is None:
+        days_back = config.patent_search_days
+    end_date = datetime.now()
+    start_date = end_date - timedelta(days=days_back)
+    date_filter = f"cdr:1,cd_min:{start_date.strftime('%-m/%-d/%Y')},cd_max:{end_date.strftime('%-m/%-d/%Y')}"
+
     # Make API call
     params = {
       "engine": "google_patents",
       "q": company,
       "num": 10,
       "filter": 1,
-      "tbs": "cdr:1,cd_min:10/28/2025,cd_max:11/4/2025",
+      "tbs": date_filter,
       "api_key": config.api_key,
     }
     search = serpapi.search(params)
@@ -46,7 +55,7 @@ class SERP:
 
   def save_patents(patent: Patent) -> Patent:
     """
-    Save the patent PDF to the patents folder
+    Save the patent PDF to the patents folder, skipping download if already cached.
 
     Args:
       patent: Patent object
@@ -54,12 +63,15 @@ class SERP:
     Returns:
       Patent object with updated PDF path
     """
+    pdf_path = f"patents/{patent.patent_id}.pdf"
+    os.makedirs("patents", exist_ok=True)
+
+    if not (os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 0):
       response = requests.get(patent.pdf_link)
-    print(patent.pdf_link)
+      with open(pdf_path, "wb") as f:
-    with open(f"patents/{patent.patent_id}.pdf", "wb") as f:
         f.write(response.content)
 
-    patent.pdf_path = f"patents/{patent.patent_id}.pdf"
+    patent.pdf_path = pdf_path
     return patent
 
   def parse_patent_pdf(pdf_path: str) -> Dict:

scripts/init_database.py

@@ -40,6 +40,9 @@ def main():
         print("\nTables created:")
         print("  - llm_messages: Stores all LLM prompts and responses")
         print("  - users: Stores user accounts")
+        print("  - jobs: Stores async batch job state")
+        print("  - patents: Patent PDF cache")
+        print("  - serp_queries: SERP query result cache")
         print("\nIndexes created:")
         print("  - idx_messages_timestamp: For time-based queries")
         print("  - idx_messages_company: For company-specific queries")

tests/test_analyzer.py

@@ -1,11 +1,22 @@
 """Tests for the high-level company analyzer orchestration."""
 
 import pytest
-from unittest.mock import Mock, patch, call
+from unittest.mock import Mock, patch, call, MagicMock
 from SPARC.analyzer import CompanyAnalyzer
 from SPARC.types import Patent, Patents, CompanyAnalysisResult, BatchAnalysisResult
 
 
+@pytest.fixture(autouse=True)
+def mock_db(mocker):
+    """Mock DatabaseClient for all tests so no real DB connection is needed."""
+    mock_db_cls = mocker.patch("SPARC.analyzer.DatabaseClient")
+    mock_db_instance = MagicMock()
+    mock_db_instance.get_cached_patent.return_value = None
+    mock_db_instance.get_cached_serp_query.return_value = None
+    mock_db_cls.return_value = mock_db_instance
+    return mock_db_instance
+
+
 class TestCompanyAnalyzer:
     """Test the CompanyAnalyzer orchestration logic."""
 
@@ -17,7 +28,7 @@ class TestCompanyAnalyzer:
 
         mock_llm.assert_called_once_with(api_key="test-key")
 
-    def test_analyze_company_full_pipeline(self, mocker):
+    def test_analyze_company_full_pipeline(self, mocker, mock_db):
         """Test complete company analysis pipeline."""
         # Mock all the dependencies
         mock_query = mocker.patch("SPARC.analyzer.SERP.query")
@@ -178,6 +189,180 @@ class TestCompanyAnalyzer:
         assert "PDF not found" in result
 
 
+class TestSingleQueryBugFix:
+    """Test that SERP.query is only called once per company analysis."""
+
+    def test_analyze_company_safe_calls_query_once(self, mocker, mock_db):
+        """_analyze_company_safe should call SERP.query exactly once."""
+        mock_query = mocker.patch("SPARC.analyzer.SERP.query")
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+        mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
+        mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
+        mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+        mock_query.return_value = Patents(patents=[patent])
+
+        def save_side_effect(p):
+            p.pdf_path = "patents/US123.pdf"
+            return p
+
+        mock_save.side_effect = save_side_effect
+        mock_parse.return_value = {"abstract": "Test"}
+        mock_minimize.return_value = "Content"
+
+        mock_llm_instance = Mock()
+        mock_llm_instance.analyze_patent_portfolio.return_value = "Analysis"
+        mock_llm.return_value = mock_llm_instance
+
+        analyzer = CompanyAnalyzer()
+        analyzer._analyze_company_safe("TestCorp")
+
+        # The key assertion: SERP.query called exactly once, not twice
+        mock_query.assert_called_once_with("TestCorp")
+
+    def test_analyze_company_with_prefetched_patents_skips_query(self, mocker):
+        """analyze_company should not call SERP.query when patents are provided."""
+        mock_query = mocker.patch("SPARC.analyzer.SERP.query")
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+        mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
+        mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
+        mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+        prefetched = Patents(patents=[patent])
+
+        def save_side_effect(p):
+            p.pdf_path = "patents/US123.pdf"
+            return p
+
+        mock_save.side_effect = save_side_effect
+        mock_parse.return_value = {"abstract": "Test"}
+        mock_minimize.return_value = "Content"
+
+        mock_llm_instance = Mock()
+        mock_llm_instance.analyze_patent_portfolio.return_value = "Analysis"
+        mock_llm.return_value = mock_llm_instance
+
+        analyzer = CompanyAnalyzer()
+        analyzer.analyze_company("TestCorp", patents=prefetched)
+
+        # SERP.query should never be called
+        mock_query.assert_not_called()
+
+
+class TestPatentCaching:
+    """Test patent-level DB caching in the pipeline."""
+
+    def test_process_single_patent_uses_db_cache(self, mocker, mock_db):
+        """_process_single_patent returns cached content when available."""
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+
+        mock_db.get_cached_patent.return_value = {
+            "patent_id": "US123",
+            "minimized_content": "Cached minimized content",
+        }
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+        result = CompanyAnalyzer._process_single_patent(patent, "TestCorp", mock_db)
+
+        assert result == {"patent_id": "US123", "content": "Cached minimized content"}
+        # Should NOT download since cache hit
+        mock_save.assert_not_called()
+
+    def test_process_single_patent_stores_to_db_cache(self, mocker, mock_db):
+        """_process_single_patent stores result in DB after processing."""
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+        mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
+        mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
+
+        # No cache hit
+        mock_db.get_cached_patent.return_value = None
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+
+        def save_side_effect(p):
+            p.pdf_path = "patents/US123.pdf"
+            return p
+
+        mock_save.side_effect = save_side_effect
+        mock_parse.return_value = {"abstract": "Test abstract"}
+        mock_minimize.return_value = "Minimized content"
+
+        result = CompanyAnalyzer._process_single_patent(patent, "TestCorp", mock_db)
+
+        assert result == {"patent_id": "US123", "content": "Minimized content"}
+        mock_db.store_patent.assert_called_once_with(
+            patent_id="US123",
+            company_name="TestCorp",
+            pdf_link="http://example.com/test.pdf",
+            raw_sections={"abstract": "Test abstract"},
+            minimized_content="Minimized content",
+        )
+
+    def test_serp_query_cache_hit_skips_api(self, mocker, mock_db):
+        """When SERP query is cached, API call is skipped."""
+        mock_query = mocker.patch("SPARC.analyzer.SERP.query")
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+        mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
+        mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
+        mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
+
+        # Simulate SERP cache hit
+        mock_db.get_cached_serp_query.return_value = ["US123"]
+        # Simulate patent cache hit too
+        mock_db.get_cached_patent.return_value = {
+            "patent_id": "US123",
+            "minimized_content": "Cached content",
+        }
+
+        mock_llm_instance = Mock()
+        mock_llm_instance.analyze_patent_portfolio.return_value = "Analysis"
+        mock_llm.return_value = mock_llm_instance
+
+        analyzer = CompanyAnalyzer()
+        result = analyzer.analyze_company("TestCorp")
+
+        assert result == "Analysis"
+        # SERP.query should NOT be called
+        mock_query.assert_not_called()
+        # No downloads should happen
+        mock_save.assert_not_called()
+
+    def test_serp_query_cache_miss_stores_result(self, mocker, mock_db):
+        """When SERP query cache misses, result is stored after API call."""
+        mock_query = mocker.patch("SPARC.analyzer.SERP.query")
+        mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
+        mock_parse = mocker.patch("SPARC.analyzer.SERP.parse_patent_pdf")
+        mock_minimize = mocker.patch("SPARC.analyzer.SERP.minimize_patent_for_llm")
+        mock_llm = mocker.patch("SPARC.analyzer.LLMAnalyzer")
+
+        mock_db.get_cached_serp_query.return_value = None
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+        mock_query.return_value = Patents(patents=[patent])
+
+        def save_side_effect(p):
+            p.pdf_path = "patents/US123.pdf"
+            return p
+
+        mock_save.side_effect = save_side_effect
+        mock_parse.return_value = {"abstract": "Test"}
+        mock_minimize.return_value = "Content"
+
+        mock_llm_instance = Mock()
+        mock_llm_instance.analyze_patent_portfolio.return_value = "Analysis"
+        mock_llm.return_value = mock_llm_instance
+
+        analyzer = CompanyAnalyzer()
+        analyzer.analyze_company("TestCorp")
+
+        mock_db.store_serp_query.assert_called_once()
+        call_kwargs = mock_db.store_serp_query.call_args[1]
+        assert call_kwargs["company_name"] == "TestCorp"
+        assert call_kwargs["patent_ids"] == ["US123"]
+
+
 class TestBatchProcessing:
     """Test multi-company batch processing functionality."""
 
@@ -316,7 +501,7 @@ class TestBatchProcessing:
 
         assert callback.call_count == 2
 
-    def test_company_analysis_result_structure(self, mocker):
+    def test_company_analysis_result_structure(self, mocker, mock_db):
         """Test CompanyAnalysisResult has correct structure."""
         mock_query = mocker.patch("SPARC.analyzer.SERP.query")
         mock_save = mocker.patch("SPARC.analyzer.SERP.save_patents")
@@ -327,6 +512,9 @@ class TestBatchProcessing:
         patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
         mock_query.return_value = Patents(patents=[patent])
 
+        # Simulate DB caching: after store, subsequent get returns the IDs
+        mock_db.get_cached_serp_query.side_effect = [None, ["US123"]]
+
         def save_side_effect(p):
             p.pdf_path = "patents/US123.pdf"
             return p

tests/test_api.py

@@ -5,7 +5,7 @@ from datetime import datetime
 from unittest.mock import Mock, patch
 from fastapi.testclient import TestClient
 
-from SPARC.api import app, _analyzer, _jobs
+from SPARC.api import app
 from SPARC.types import CompanyAnalysisResult, BatchAnalysisResult
 
 

tests/test_serp_api.py

@@ -1,7 +1,11 @@
 """Tests for SERP API patent retrieval and parsing functionality."""
 
+import os
 import pytest
+from unittest.mock import patch, Mock
+from datetime import datetime, timedelta
 from SPARC.serp_api import SERP
+from SPARC.types import Patent
 
 
 class TestTextCleaning:
@@ -176,3 +180,89 @@ class TestPatentMinimization:
 
         # Sections should be separated by double newlines
         assert "\n\n" in result
+
+
+class TestDynamicDateRange:
+    """Test dynamic date range computation in SERP.query."""
+
+    def test_query_uses_rolling_date_window(self, mocker):
+        """Verify the date filter uses a rolling window, not hardcoded dates."""
+        mock_search = mocker.patch("SPARC.serp_api.serpapi.search")
+        mock_search.return_value = {"organic_results": []}
+        mocker.patch("SPARC.serp_api.config.api_key", "fake-key")
+        mocker.patch("SPARC.serp_api.config.patent_search_days", 90)
+
+        SERP.query("TestCorp")
+
+        call_params = mock_search.call_args[0][0]
+        tbs = call_params["tbs"]
+        # Should contain "cdr:1,cd_min:" with a date, not the old hardcoded one
+        assert "cdr:1,cd_min:" in tbs
+        assert "10/28/2025" not in tbs  # old hardcoded date gone
+
+    def test_query_respects_days_back_param(self, mocker):
+        """Verify days_back parameter controls the date window."""
+        mock_search = mocker.patch("SPARC.serp_api.serpapi.search")
+        mock_search.return_value = {"organic_results": []}
+        mocker.patch("SPARC.serp_api.config.api_key", "fake-key")
+        mocker.patch("SPARC.serp_api.config.patent_search_days", 90)
+
+        now = datetime.now()
+        SERP.query("TestCorp", days_back=30)
+
+        call_params = mock_search.call_args[0][0]
+        tbs = call_params["tbs"]
+        expected_start = (now - timedelta(days=30)).strftime("%-m/%-d/%Y")
+        assert expected_start in tbs
+
+
+class TestFilesystemPDFCaching:
+    """Test that save_patents skips download for existing files."""
+
+    def test_save_patents_skips_download_when_cached(self, mocker, tmp_path):
+        """Already-downloaded PDFs should not be re-downloaded."""
+        mock_get = mocker.patch("SPARC.serp_api.requests.get")
+        mocker.patch("SPARC.serp_api.os.makedirs")
+
+        pdf_path = tmp_path / "US123.pdf"
+        pdf_path.write_bytes(b"%PDF-1.4 fake content")
+
+        mocker.patch("SPARC.serp_api.os.path.exists", return_value=True)
+        mocker.patch("SPARC.serp_api.os.path.getsize", return_value=100)
+
+        patent = Patent(patent_id="US123", pdf_link="http://example.com/test.pdf")
+        result = SERP.save_patents(patent)
+
+        mock_get.assert_not_called()
+        assert result.pdf_path == "patents/US123.pdf"
+
+    def test_save_patents_downloads_when_not_cached(self, mocker):
+        """Missing PDFs should be downloaded."""
+        mock_response = Mock()
+        mock_response.content = b"%PDF-1.4 content"
+        mock_get = mocker.patch("SPARC.serp_api.requests.get", return_value=mock_response)
+        mocker.patch("SPARC.serp_api.os.makedirs")
+        mocker.patch("SPARC.serp_api.os.path.exists", return_value=False)
+        mock_open = mocker.patch("builtins.open", mocker.mock_open())
+
+        patent = Patent(patent_id="US456", pdf_link="http://example.com/test.pdf")
+        result = SERP.save_patents(patent)
+
+        mock_get.assert_called_once_with("http://example.com/test.pdf")
+        assert result.pdf_path == "patents/US456.pdf"
+
+    def test_save_patents_redownloads_empty_files(self, mocker):
+        """Empty/corrupt PDFs (0 bytes) should be re-downloaded."""
+        mock_response = Mock()
+        mock_response.content = b"%PDF-1.4 content"
+        mock_get = mocker.patch("SPARC.serp_api.requests.get", return_value=mock_response)
+        mocker.patch("SPARC.serp_api.os.makedirs")
+        mocker.patch("SPARC.serp_api.os.path.exists", return_value=True)
+        mocker.patch("SPARC.serp_api.os.path.getsize", return_value=0)
+        mock_open = mocker.patch("builtins.open", mocker.mock_open())
+
+        patent = Patent(patent_id="US789", pdf_link="http://example.com/test.pdf")
+        result = SERP.save_patents(patent)
+
+        mock_get.assert_called_once()
+        assert result.pdf_path == "patents/US789.pdf"