feat: add page citation source API

apps/api/app/api/v2/api_v2.py

@@ -1,5 +1,6 @@
 """API v2 route registry."""
 
+from app.api.v1.routes import documents as v1_documents
 from app.api.v2.routes import documents, jobs, retrieval
 from fastapi import APIRouter
 
@@ -8,5 +9,6 @@
 api_router.include_router(jobs.router, prefix="/jobs", tags=["Jobs"])
 api_router.include_router(retrieval.router, prefix="/retrieval", tags=["Retrieval"])
 api_router.include_router(documents.router, prefix="/documents", tags=["Documents"])
+api_router.include_router(v1_documents.router, prefix="/documents", tags=["Documents"])
 
 __all__ = ["api_router"]

apps/api/app/api/v2/routes/documents.py

@@ -1,5 +1,40 @@
 """Documents API v2 routes."""
 
-from app.api.v1.routes.documents import router
+from __future__ import annotations
+
+from typing import Any
+
+from app.api.dependencies.current_user import with_current_user
+from app.services.documents.lifecycle_service import DocumentService
+from app.services.rate_limit.data_structures import CurrentUser
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from shared.core.database import get_db
+from shared.core.exceptions.domain_exceptions import NotFoundException
+
+router = APIRouter(tags=["Documents"])
+
+_document_service = DocumentService()
+
+
+@router.get("/{document_id}/files/page-citation-source")
+async def get_document_page_citation_source(
+    document_id: str,
+    current_user: CurrentUser = Depends(with_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> dict[str, Any]:
+    response = await _document_service.get_document_page_citation_source(
+        db,
+        user_id=current_user.user_id,
+        document_id=document_id,
+    )
+    if response is None:
+        raise NotFoundException(
+            resource="Document page citation source",
+            resource_id=document_id,
+            internal_message="Document page citation source not found",
+        )
+    return response
 
 __all__ = ["router"]

apps/api/app/repositories/document_repository.py

@@ -11,9 +11,11 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from shared.models.database.document import Document, DocumentChunk, DocumentSection
+from shared.models.database.job import Job
 from shared.models.database.job_result import JobResult
 
 DocumentChunkRow = tuple[DocumentChunk, DocumentSection | None, JobResult]
+DocumentJobRevisionRow = tuple[Document, JobResult, Job]
 
 
 class DocumentRepository:
@@ -66,6 +68,29 @@ async def get_document(
         )
         return result.scalar_one_or_none()
 
+    async def get_current_document_job_revision(
+        self,
+        db: AsyncSession,
+        *,
+        document_id: str,
+        user_id: str,
+    ) -> DocumentJobRevisionRow | None:
+        stmt = (
+            select(Document, JobResult, Job)
+            .join(JobResult, JobResult.id == Document.current_job_result_id)
+            .join(Job, Job.job_id == JobResult.job_id)
+            .where(Document.document_id == document_id)
+            .where(Document.user_id == user_id)
+            .where(JobResult.document_id == Document.document_id)
+            .where(Job.user_id == user_id)
+            .where(Document.status != "archived")
+            .limit(1)
+        )
+
+        result = await db.execute(stmt)
+        row = result.first()
+        return cast(DocumentJobRevisionRow | None, row)
+
     async def archive_document(
         self,
         db: AsyncSession,

apps/api/app/services/documents/lifecycle_service.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import math
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from typing import Any
 
 from app.repositories.document_repository import DocumentRepository
@@ -19,6 +19,10 @@
 
 _DOCUMENT_CHUNK_ASSET_URL_EXPIRES_SECONDS = 7 * 24 * 60 * 60
 _MEDIA_CHUNK_TYPES = frozenset({"image", "table"})
+_PAGE_CITATION_SOURCE_EXPIRES_SECONDS = 60 * 60
+_PAGE_CITATION_SOURCE_FILE_NAME = "source.pdf"
+_PAGE_CITATION_SOURCE_VARIANT = "normalized_pdf"
+_PAGE_MEMORY_PARSE_TRACK = "page_memory"
 
 
 def _datetime_payload(value: datetime | None) -> str | None:
@@ -53,6 +57,89 @@ def _document_chunk_asset_url(
         return None
 
 
+def _document_page_assets(
+    *,
+    metadata: dict[str, Any] | None,
+    job_id: str | None,
+    include_asset_urls: bool,
+    result_storage: ResultStorage | None,
+) -> list[dict[str, Any]]:
+    if not isinstance(metadata, dict):
+        return []
+    raw_assets = metadata.get("page_assets")
+    if not isinstance(raw_assets, list):
+        return []
+
+    page_assets: list[dict[str, Any]] = []
+    for raw_asset in raw_assets:
+        if not isinstance(raw_asset, dict):
+            continue
+        asset = _normalize_page_asset(raw_asset)
+        if asset is None:
+            continue
+        if include_asset_urls and job_id and result_storage is not None:
+            asset_url = _page_asset_url(
+                job_id=job_id,
+                artifact_ref=asset["artifact_ref"],
+                result_storage=result_storage,
+            )
+            if asset_url:
+                asset["asset_url"] = asset_url
+        page_assets.append(asset)
+    return page_assets
+
+
+def _normalize_page_asset(raw_asset: dict[str, Any]) -> dict[str, Any] | None:
+    page_num = _positive_int(raw_asset.get("page_num"))
+    artifact_ref = str(raw_asset.get("artifact_ref") or "").strip()
+    content_type = str(raw_asset.get("content_type") or "").strip()
+    source = str(raw_asset.get("source") or "").strip()
+    if page_num is None or not artifact_ref or not content_type or not source:
+        return None
+
+    asset: dict[str, Any] = {
+        "page_num": page_num,
+        "artifact_ref": artifact_ref,
+        "content_type": content_type,
+        "source": source,
+    }
+    if (asset_url := str(raw_asset.get("asset_url") or "").strip()):
+        asset["asset_url"] = asset_url
+    if (width := _positive_int(raw_asset.get("width"))) is not None:
+        asset["width"] = width
+    if (height := _positive_int(raw_asset.get("height"))) is not None:
+        asset["height"] = height
+    return asset
+
+
+def _page_asset_url(
+    *,
+    job_id: str,
+    artifact_ref: str,
+    result_storage: ResultStorage,
+) -> str | None:
+    normalized_ref = result_storage.normalize_artifact_ref(artifact_ref)
+    if not normalized_ref or not normalized_ref.startswith("page_citation_assets/"):
+        return None
+    try:
+        return result_storage.generate_artifact_url(
+            job_id=job_id,
+            artifact_ref=normalized_ref,
+            expires_in=_DOCUMENT_CHUNK_ASSET_URL_EXPIRES_SECONDS,
+        )
+    except Exception as exc:
+        logger.warning(f"Failed to generate page citation asset URL (ignored): {exc}")
+        return None
+
+
+def _positive_int(value: Any) -> int | None:
+    try:
+        number = int(value)
+    except (TypeError, ValueError):
+        return None
+    return number if number > 0 else None
+
+
 def document_payload(document) -> dict[str, Any]:
     return {
         "document_id": document.document_id,
@@ -75,9 +162,11 @@ def __init__(
         *,
         repository: DocumentRepository | None = None,
         graph_service: DocumentGraphService | None = None,
+        result_storage: ResultStorage | None = None,
     ) -> None:
         self._repository = repository or DocumentRepository()
         self._graph_service = graph_service or DocumentGraphService()
+        self._result_storage = result_storage
 
     async def list_documents(
         self,
@@ -246,6 +335,55 @@ async def get_document(
             return None
         return document_payload(document)
 
+    async def get_document_page_citation_source(
+        self,
+        db: AsyncSession,
+        *,
+        user_id: str,
+        document_id: str,
+    ) -> dict[str, Any] | None:
+        row = await self._repository.get_current_document_job_revision(
+            db,
+            user_id=user_id,
+            document_id=document_id,
+        )
+        if row is None:
+            return None
+
+        document, job_result, job = row
+        if document.parse_track != _PAGE_MEMORY_PARSE_TRACK:
+            return None
+
+        result_storage = self._result_storage or get_result_storage()
+        if not result_storage.verify_raw_exists(
+            job_id=job_result.job_id,
+            relative_path=_PAGE_CITATION_SOURCE_FILE_NAME,
+        ):
+            return None
+
+        source_url = result_storage.generate_raw_file_url(
+            job_id=job_result.job_id,
+            relative_path=_PAGE_CITATION_SOURCE_FILE_NAME,
+            expires_in=_PAGE_CITATION_SOURCE_EXPIRES_SECONDS,
+        )
+        if not source_url:
+            return None
+
+        expires_at = datetime.now(timezone.utc) + timedelta(
+            seconds=_PAGE_CITATION_SOURCE_EXPIRES_SECONDS,
+        )
+        return {
+            "document_id": document.document_id,
+            "namespace": document.namespace,
+            "job_id": job.job_id,
+            "job_result_id": job_result.id,
+            "variant": _PAGE_CITATION_SOURCE_VARIANT,
+            "file_name": _PAGE_CITATION_SOURCE_FILE_NAME,
+            "content_type": "application/pdf",
+            "url": source_url,
+            "expires_at": expires_at.isoformat(),
+        }
+
     def _chunk_payload(
         self,
         *,
@@ -257,7 +395,17 @@ def _chunk_payload(
     ) -> dict[str, Any]:
         chunk_type = _normalize_chunk_type(chunk.chunk_type)
         file_path = chunk.file_path
-        return {
+        raw_metadata = chunk.chunk_metadata or {}
+        page_assets = _document_page_assets(
+            metadata=raw_metadata,
+            job_id=job_id,
+            include_asset_urls=include_asset_urls,
+            result_storage=result_storage,
+        )
+        metadata = dict(raw_metadata)
+        if include_asset_urls and page_assets:
+            metadata["page_assets"] = page_assets
+        payload = {
             "id": chunk.id,
             "chunk_id": chunk.chunk_id,
             "chunk_type": chunk_type,
@@ -267,7 +415,7 @@ def _chunk_payload(
             "source_chunk_path": chunk.source_chunk_path,
             "file_path": file_path,
             "sort_order": chunk.sort_order,
-            "metadata": chunk.chunk_metadata,
+            "metadata": metadata,
             "asset_url": _document_chunk_asset_url(
                 chunk_type=chunk_type,
                 job_id=job_id,
@@ -277,6 +425,7 @@ def _chunk_payload(
             ),
             "created_at": _datetime_payload(chunk.created_at),
         }
+        return payload
 
     async def archive_document(
         self,