fix: auto-download patent PDF in analyze_single_patent before reading

When the PDF is not on disk, analyze_single_patent now looks up the
cached PDF link from the database and downloads it automatically.
If no link is cached, a clear FileNotFoundError is raised. Also adds
a GET /analyze/patent/{patent_id} API endpoint that exposes this
functionality and returns 404 when the PDF cannot be obtained.

Closes leeworks-agents/SPARC#36

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

SPARC/analyzer.py

@@ -108,12 +108,10 @@ class CompanyAnalyzer:
     def analyze_single_patent(self, patent_id: str, company_name: str) -> str:
         """Analyze a single patent by ID.
 
-        Prerequisite:
-            The patent PDF must already exist at ``patents/{patent_id}.pdf``
-            before calling this method. PDFs are downloaded automatically when
-            using the batch analysis pipeline (``analyze_company`` or the
-            ``/analyze/batch`` API endpoint). For standalone usage, download
-            the PDF manually or call ``SERP.save_patents()`` first.
+        If the patent PDF is not already on disk, this method attempts to
+        download it automatically by looking up the PDF link in the database
+        cache. If the link is not cached either, a ``FileNotFoundError`` is
+        raised with instructions on how to obtain the PDF.
 
         Args:
           patent_id: Publication ID of the patent (e.g. "US-11234567-B2")
@@ -123,7 +121,7 @@ class CompanyAnalyzer:
           Analysis of the specific patent's innovation quality
 
         Raises:
-          FileNotFoundError: If the patent PDF is not found at the expected path.
+          FileNotFoundError: If the patent PDF cannot be found or downloaded.
         """
         import os
         logger.info("Analyzing patent %s for %s...", patent_id, company_name)
@@ -131,10 +129,22 @@ class CompanyAnalyzer:
         patent_path = f"patents/{patent_id}.pdf"
 
         if not os.path.exists(patent_path):
-            raise FileNotFoundError(
-                f"Patent PDF not found at '{patent_path}'. "
-                f"Download the PDF first using SERP.save_patents() or the batch analysis pipeline."
-            )
+            # Attempt to download the PDF automatically from cached metadata
+            cached = self.db.get_cached_patent(patent_id)
+            pdf_link = cached.get("pdf_link") if cached else None
+
+            if pdf_link:
+                logger.info("PDF not on disk; downloading %s from cached link", patent_id)
+                patent = SERP.save_patents(
+                    Patent(patent_id=patent_id, pdf_link=pdf_link)
+                )
+                patent_path = patent.pdf_path
+            else:
+                raise FileNotFoundError(
+                    f"Patent PDF not found at '{patent_path}' and no download link is "
+                    f"cached for '{patent_id}'. Run a company analysis first to populate "
+                    f"the cache, or call SERP.save_patents() with the patent's PDF link."
+                )
 
         try:
             sections = SERP.parse_patent_pdf(patent_path)

SPARC/api.py

@@ -77,13 +77,6 @@ class JobStatus(BaseModel):
     error: str | None = None
 
 
-class PaginatedJobsResponse(BaseModel):
-    """Paginated response for job listings."""
-
-    items: list["JobStatus"]
-    next_cursor: str | None = None
-
-
 class HealthResponse(BaseModel):
     """Health check response."""
 
@@ -436,6 +429,38 @@ async def analyze_company(
     return _convert_result(result)
 
 
+@app.get(
+    "/analyze/patent/{patent_id}",
+    tags=["Analysis"],
+)
+async def analyze_single_patent(
+    patent_id: str,
+    company_name: str = Query(description="Company name for analysis context"),
+    _: UserResponse = Depends(get_current_user),
+):
+    """Analyze a single patent by its publication ID.
+
+    If the patent PDF is not already cached locally, the system will attempt
+    to download it automatically from a previously cached link. If no link
+    is available, a 404 error is returned.
+
+    Args:
+        patent_id: Patent publication ID (e.g. "US-11234567-B2")
+        company_name: Company name for analysis context
+
+    Returns:
+        Analysis text for the patent
+    """
+    if not _analyzer:
+        raise HTTPException(status_code=503, detail="Analyzer not initialized")
+
+    try:
+        analysis = _analyzer.analyze_single_patent(patent_id, company_name)
+        return {"patent_id": patent_id, "company_name": company_name, "analysis": analysis}
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
 @app.post(
     "/analyze/batch",
     response_model=BatchAnalysisResponse,
@@ -584,51 +609,24 @@ async def get_job_status(
     return _job_row_to_status(job_row)
 
 
-@app.get("/jobs", response_model=PaginatedJobsResponse, tags=["Jobs"])
+@app.get("/jobs", response_model=list[JobStatus], tags=["Jobs"])
 async def list_jobs(
     status: Annotated[
         str | None,
         Query(description="Filter by status: pending, running, completed, failed"),
     ] = None,
     limit: Annotated[int, Query(ge=1, le=100)] = 10,
-    cursor: Annotated[
-        str | None,
-        Query(description="Opaque cursor from a previous response's next_cursor field"),
-    ] = None,
     _: UserResponse = Depends(get_current_user),
 ):
-    """List analysis jobs with cursor-based pagination.
-
-    Pass ``limit`` to control page size. The response includes a ``next_cursor``
-    field; pass it back as the ``cursor`` query parameter to fetch the next page.
-    When ``next_cursor`` is ``null``, there are no more results.
-
-    Existing clients that use only ``limit`` (without ``cursor``) continue to
-    work without modification.
+    """List all analysis jobs.
 
     Args:
         status: Optional filter by job status
         limit: Maximum number of jobs to return (default 10, max 100)
-        cursor: Opaque pagination cursor from a previous response
 
     Returns:
-        Paginated list of job statuses
+        List of job statuses
     """
     db = _get_job_db()
-    # Fetch one extra to determine if there is a next page
-    job_rows = db.list_jobs(status=status, limit=limit + 1, cursor=cursor)
-
-    has_next = len(job_rows) > limit
-    if has_next:
-        job_rows = job_rows[:limit]
-
-    items = [_job_row_to_status(row) for row in job_rows]
-
-    next_cursor = None
-    if has_next and job_rows:
-        last = job_rows[-1]
-        created = last["created_at"]
-        ts = created.isoformat() if hasattr(created, "isoformat") else str(created)
-        next_cursor = f"{ts}|{last['job_id']}"
-
-    return PaginatedJobsResponse(items=items, next_cursor=next_cursor)
+    job_rows = db.list_jobs(status=status, limit=limit)
+    return [_job_row_to_status(row) for row in job_rows]

SPARC/database.py

@@ -568,45 +568,20 @@ class DatabaseClient:
         self,
         status: Optional[str] = None,
         limit: int = 10,
-        cursor: Optional[str] = None,
     ) -> List[Dict]:
-        """List jobs with optional status filter and cursor-based pagination.
-
-        Args:
-            status: Optional status filter (pending, running, completed, failed).
-            limit: Maximum number of jobs to return.
-            cursor: Opaque cursor (``created_at|job_id``) from a previous
-                response. When provided, only jobs older than the cursor are
-                returned.
-
-        Returns:
-            List of job dicts ordered by created_at descending.
-        """
-        conditions: list[str] = []
-        params: list = []
-
-        if status:
-            conditions.append("status = %s")
-            params.append(status)
-
-        if cursor:
-            try:
-                ts_str, cursor_job_id = cursor.rsplit("|", 1)
-                conditions.append("(created_at, job_id) < (%s, %s)")
-                params.extend([ts_str, cursor_job_id])
-            except ValueError:
-                pass  # Ignore malformed cursors; return from start
-
+        """List jobs, optionally filtered by status."""
         query = "SELECT * FROM jobs"
-        if conditions:
-            query += " WHERE " + " AND ".join(conditions)
-        query += " ORDER BY created_at DESC, job_id DESC LIMIT %s"
+        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 cur:
-                cur.execute(query, params)
-                return [dict(row) for row in cur.fetchall()]
+            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'.