fix(api): paginate GET /task/{task_id}/result to prevent OOM on large scans

backend/secuscan/finding_intelligence.py

@@ -6,7 +6,7 @@
 import json
 import re
 from datetime import datetime, timezone
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Union
 from urllib.parse import urlparse
 
 
@@ -543,6 +543,125 @@ def build_asset_summary(
     return summary
 
 
+async def _as_async_iter(source):
+    """Normalize a sync or async iterable into an async iterator."""
+    if hasattr(source, "__anext__"):
+        async for item in source:
+            yield item
+    else:
+        for item in source:
+            yield item
+
+
+async def aggregate_findings_streaming(
+    findings: Union[Iterable[Dict[str, Any]], AsyncIterable[Dict[str, Any]]],
+    asset_services: List[Dict[str, Any]],
+) -> tuple:
+    """Compute severity_counts, finding_groups, and asset_summary in one pass.
+
+    `findings` may be a plain list/generator or an async generator. It is
+    iterated exactly once, so callers can pass an async generator that
+    streams rows from the database in bounded-size chunks -- the full
+    finding set never needs to be materialized in memory at once, and the
+    result is always exact (never a capped sample), regardless of how
+    large the scan is.
+    """
+    severity_counts: Dict[str, int] = {}
+    groups: Dict[str, Dict[str, Any]] = {}
+    assets: Dict[str, Dict[str, Any]] = {}
+
+    for service in asset_services:
+        asset_id = str(service.get("asset_id") or _stable_id("asset", service.get("target"), service.get("host"), service.get("port"), service.get("protocol")))
+        entry = assets.setdefault(
+            asset_id,
+            {
+                "asset_id": asset_id,
+                "label": service.get("host") or service.get("target"),
+                "target": service.get("target"),
+                "services": [],
+                "finding_count": 0,
+                "validated_count": 0,
+                "highest_severity": "info",
+            },
+        )
+        entry["services"].append(service)
+
+    async for finding in _as_async_iter(findings):
+        severity = str(finding.get("severity", "info")).lower()
+        severity_counts[severity] = severity_counts.get(severity, 0) + 1
+
+        group_id = str(finding.get("finding_group_id") or finding.get("id") or _stable_id("group", finding.get("title"), finding.get("target")))
+        current = groups.get(group_id)
+        if current is None:
+            groups[group_id] = {
+                "id": group_id,
+                "title": finding.get("title"),
+                "severity": _normalize_severity(finding.get("severity")),
+                "category": finding.get("category"),
+                "target": finding.get("target"),
+                "asset_id": finding.get("asset_id"),
+                "finding_kind": finding.get("finding_kind", "observation"),
+                "validated": bool(finding.get("validated")),
+                "cve": finding.get("cve"),
+                "cpe": finding.get("cpe"),
+                "confidence": finding.get("confidence"),
+                "confidence_reason": finding.get("confidence_reason"),
+                "first_seen_at": finding.get("first_seen_at") or finding.get("discovered_at"),
+                "last_seen_at": finding.get("last_seen_at") or finding.get("discovered_at"),
+                "occurrence_count": int(finding.get("occurrence_count") or 1),
+                "evidence_count": int(finding.get("evidence_count") or len(finding.get("evidence", []))),
+                "corroborating_sources": list(finding.get("corroborating_sources", [])),
+                "analyst_status": finding.get("analyst_status", "new"),
+                "retest_status": finding.get("retest_status", "not_requested"),
+                "latest_finding_id": finding.get("id"),
+                "findings": [finding],
+            }
+        else:
+            current["validated"] = bool(current.get("validated")) or bool(finding.get("validated"))
+            if _severity_rank(finding.get("severity", "info")) > _severity_rank(current.get("severity", "info")):
+                current["severity"] = _normalize_severity(finding.get("severity"))
+            current["last_seen_at"] = max(str(current.get("last_seen_at") or ""), str(finding.get("last_seen_at") or finding.get("discovered_at") or ""))
+            current["first_seen_at"] = min(str(current.get("first_seen_at") or ""), str(finding.get("first_seen_at") or finding.get("discovered_at") or ""))
+            current["occurrence_count"] = max(int(current.get("occurrence_count") or 1), int(finding.get("occurrence_count") or 1))
+            current["evidence_count"] = max(int(current.get("evidence_count") or 0), int(finding.get("evidence_count") or len(finding.get("evidence", []))))
+            current["corroborating_sources"] = _sort_sources([*current.get("corroborating_sources", []), *finding.get("corroborating_sources", [])])
+            current["confidence"] = max(float(current.get("confidence") or 0.0), float(finding.get("confidence") or 0.0))
+            current["findings"].append(finding)
+
+        asset_id = str(finding.get("asset_id") or _stable_id("asset", finding.get("target"), *(finding.get("asset_refs") or [])))
+        entry = assets.setdefault(
+            asset_id,
+            {
+                "asset_id": asset_id,
+                "label": finding.get("target"),
+                "target": finding.get("target"),
+                "services": [],
+                "finding_count": 0,
+                "validated_count": 0,
+                "highest_severity": "info",
+            },
+        )
+        entry["finding_count"] += 1
+        if finding.get("validated"):
+            entry["validated_count"] += 1
+        if _severity_rank(finding.get("severity", "info")) > _severity_rank(entry.get("highest_severity", "info")):
+            entry["highest_severity"] = _normalize_severity(finding.get("severity"))
+
+    grouped = list(groups.values())
+    grouped.sort(
+        key=lambda item: (
+            -_severity_rank(item.get("severity", "info")),
+            -(float(item.get("confidence") or 0.0)),
+            str(item.get("title") or "").lower(),
+        )
+    )
+
+    asset_summary = list(assets.values())
+    asset_summary.sort(key=lambda item: (-_severity_rank(item.get("highest_severity", "info")), -int(item.get("finding_count", 0)), str(item.get("label") or "")))
+
+    return severity_counts, grouped, asset_summary
+
+
 def build_scan_diff(current_findings: List[Dict[str, Any]], previous_findings: List[Dict[str, Any]]) -> Dict[str, Any]:
     current = {str(item.get("finding_group_id") or item.get("id")): item for item in current_findings}
     previous = {str(item.get("finding_group_id") or item.get("id")): item for item in previous_findings}

backend/secuscan/routes.py

@@ -26,6 +26,19 @@
 
 # Re-exported for backward compatibility with integration tests
 SSE_RAW_OUTPUT_CHUNK_SIZE = 64 * 1024
+
+# GET /task/{task_id}/result pagination (issue #1621): a wide-range scan can
+# produce tens of thousands of finding rows. Returning them all in one
+# response materialises the entire result set in memory before the first
+# byte is sent, which can OOM-crash the backend process. The `findings` list
+# in the response is paginated via page/per_page; aggregate views
+# (severity_counts, finding_groups, asset_summary) always reflect the whole
+# scan exactly, computed by streaming every finding row through in bounded-
+# size chunks (TASK_RESULT_AGGREGATION_CHUNK_SIZE at a time) rather than
+# loading the full scan into memory at once.
+TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE = 100
+TASK_RESULT_FINDINGS_MAX_PER_PAGE = 500
+TASK_RESULT_AGGREGATION_CHUNK_SIZE = 1000
 from .routes_report_helpers import (
     _slugify_filename_part,
     build_report_filename,
@@ -119,7 +132,7 @@ def _json_payload(value: Any, fallback: str) -> str:
 from .workflows import scheduler, _finalize_workflow_run
 from .auth import require_api_key, get_current_owner
 from .execution_context import is_offensive_validation, normalize_execution_context
-from .finding_intelligence import build_asset_summary, build_finding_groups
+from .finding_intelligence import aggregate_findings_streaming, build_finding_groups
 from .knowledgebase import KnowledgeBase
 from .platform_resources import (
     deserialize_resource_rows,
@@ -819,15 +832,57 @@ async def download_sarif_report(task_id: str, owner: str = Depends(get_current_o
     )
 
 
+async def _iter_all_findings_for_aggregation(db, owner: str, task_id: str, chunk_size: int = TASK_RESULT_AGGREGATION_CHUNK_SIZE):
+    """Yield every finding for a task as dicts, chunk_size rows at a time.
+
+    Used to compute exact whole-scan aggregates (severity_counts,
+    finding_groups, asset_summary) without ever holding more than
+    chunk_size raw finding rows in memory at once, regardless of how many
+    findings the scan produced in total.
+    """
+    offset = 0
+    while True:
+        rows = await db.fetchall(
+            "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? "
+            "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC "
+            "LIMIT ? OFFSET ?",
+            (owner, task_id, chunk_size, offset),
+        )
+        if not rows:
+            return
+        for finding in deserialize_finding_rows(rows):
+            yield finding
+        if len(rows) < chunk_size:
+            return
+        offset += chunk_size
+
+
 @router.get("/task/{task_id}/result")
-async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)):
-    """Get task execution result"""
+async def get_task_result(
+    task_id: str,
+    owner: str = Depends(get_current_owner),
+    page: int = Query(1, ge=1, description="1-indexed page number for the `findings` list."),
+    per_page: int = Query(
+        TASK_RESULT_FINDINGS_DEFAULT_PER_PAGE,
+        ge=1,
+        le=TASK_RESULT_FINDINGS_MAX_PER_PAGE,
+        description="Findings per page, capped at TASK_RESULT_FINDINGS_MAX_PER_PAGE.",
+    ),
+):
+    """Get task execution result.
+
+    `page`/`per_page` only paginate the `findings` list in the response;
+    `total_findings`, `page`, `per_page`, and `has_more_findings` describe
+    that pagination so a client can fetch subsequent pages. All aggregate
+    views -- `severity_counts`, `finding_groups`, `asset_summary` -- always
+    reflect the entire scan exactly, independent of the requested page.
+    """
     db = await get_db()
 
     # Enforce ownership and existence check first
     await require_owned_task(db, task_id, owner)
 
-    cache_key = f"tasks:result:{task_id}:{owner}"
+    cache_key = f"tasks:result:{task_id}:{owner}:page={page}:per_page={per_page}"
     cache = await get_cache()
     cached = await cache.get_json(cache_key)
     if cached is not None:
@@ -853,32 +908,53 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner))
         except json.JSONDecodeError:
             structured = {}
 
-    finding_rows = await db.fetchall(
-        "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC",
+    total_findings_row = await db.fetchone(
+        "SELECT COUNT(*) AS count FROM findings WHERE owner_id = ? AND task_id = ?",
         (owner, task_id),
     )
+    total_findings_count = total_findings_row["count"] if total_findings_row else 0
+
+    offset = (page - 1) * per_page
+    finding_rows = await db.fetchall(
+        "SELECT * FROM findings WHERE owner_id = ? AND task_id = ? "
+        "ORDER BY (risk_score IS NULL) ASC, risk_score DESC, discovered_at DESC "
+        "LIMIT ? OFFSET ?",
+        (owner, task_id, per_page, offset),
+    )
     findings = deserialize_finding_rows(finding_rows)
+
     asset_rows = await db.fetchall(
         "SELECT * FROM asset_services WHERE owner_id = ? AND task_id = ? ORDER BY created_at DESC",
         (owner, task_id),
     )
     asset_services = deserialize_asset_service_rows(asset_rows)
 
-    if not findings and isinstance(structured, dict):
-        findings = [item for item in structured.get("findings", []) if isinstance(item, dict)]
-
-    severity_counts: Dict[str, int] = {}
-    for finding in findings:
-        severity = str(finding.get("severity", "info")).lower()
-        severity_counts[severity] = severity_counts.get(severity, 0) + 1
+    # Aggregate views (severity breakdown, groups, asset summary) must reflect
+    # the whole scan exactly, not just the current page. Streaming every
+    # finding through in bounded-size chunks keeps this exact regardless of
+    # scan size, without reintroducing the OOM this endpoint was fixed for.
+    if total_findings_count > 0:
+        severity_counts, computed_finding_groups, computed_asset_summary = await aggregate_findings_streaming(
+            _iter_all_findings_for_aggregation(db, owner, task_id), asset_services,
+        )
+    else:
+        severity_counts, computed_finding_groups, computed_asset_summary = {}, [], []
+
+    if total_findings_count == 0 and isinstance(structured, dict):
+        structured_findings = [item for item in structured.get("findings", []) if isinstance(item, dict)]
+        total_findings_count = len(structured_findings)
+        findings = structured_findings[offset:offset + per_page]
+        severity_counts, computed_finding_groups, computed_asset_summary = await aggregate_findings_streaming(
+            structured_findings, asset_services,
+        )
 
     finding_groups = structured.get("finding_groups") if isinstance(structured, dict) else None
     if not isinstance(finding_groups, list) or not finding_groups:
-        finding_groups = build_finding_groups(findings)
+        finding_groups = computed_finding_groups
 
     asset_summary = structured.get("asset_summary") if isinstance(structured, dict) else None
     if not isinstance(asset_summary, list) or not asset_summary:
-        asset_summary = build_asset_summary(findings, asset_services)
+        asset_summary = computed_asset_summary
 
     scan_diff = structured.get("scan_diff") if isinstance(structured, dict) else None
     if not isinstance(scan_diff, dict):
@@ -897,7 +973,7 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner))
         str(item) for item in structured_summary
         if isinstance(item, (str, int, float)) and str(item).strip()
     ] if isinstance(structured_summary, list) else []
-    total_findings = len(findings)
+    total_findings = total_findings_count
     if not summary and total_findings > 0:
         critical_high = severity_counts.get("critical", 0) + severity_counts.get("high", 0)
         if critical_high > 0:
@@ -947,7 +1023,11 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner))
         "errors": [{"message": redact(task_row["error_message"])}] if task_row["error_message"] else [],
         "error_message": redact(task_row["error_message"]) if task_row["error_message"] else None,
         "exit_code": task_row["exit_code"],
-        "metadata": {}
+        "metadata": {},
+        "total_findings": total_findings_count,
+        "page": page,
+        "per_page": per_page,
+        "has_more_findings": offset + len(findings) < total_findings_count,
     }
 
     if task_row["status"] in ["completed", "failed", "cancelled"]:

frontend/src/api.ts

@@ -211,6 +211,10 @@ export interface TaskResultResponse {
   error_message?: string
   exit_code?: number
   metadata?: Record<string, unknown>
+  total_findings?: number
+  page?: number
+  per_page?: number
+  has_more_findings?: boolean
 }
 
 export interface NamedResourceList<T> {
@@ -557,8 +561,15 @@ export function getTaskStatus(taskId: string): Promise<TaskStatusResponse> {
   return request<TaskStatusResponse>(`/task/${taskId}/status`)
 }
 
-export function getTaskResult(taskId: string): Promise<TaskResultResponse | null> {
-  return request<TaskResultResponse | null>(`/task/${taskId}/result`)
+export function getTaskResult(
+  taskId: string,
+  options?: { page?: number; per_page?: number },
+): Promise<TaskResultResponse | null> {
+  const params = new URLSearchParams()
+  if (options?.page) params.set('page', String(options.page))
+  if (options?.per_page) params.set('per_page', String(options.per_page))
+  const suffix = params.toString() ? `?${params.toString()}` : ''
+  return request<TaskResultResponse | null>(`/task/${taskId}/result${suffix}`)
 }
 
 export function getTaskDiff(taskId: string): Promise<ScanDiff> {