SUBMISSION-196 PDF-Only Stamping

submit_ce/api/compile_service.py

@@ -169,3 +169,36 @@ def is_available(self) -> bool:
             `True` if service is configured and available.
         """
         ...
+
+    @abstractmethod
+    def stamp(self, pdf_bytes: bytes, watermark_text: str,
+              watermark_link: Optional[str] = None) -> bytes:
+        """
+        Apply the temporary submission watermark/stamp to a PDF.
+
+        Used for PDF-only submissions, which never run ``/convert`` (TeX
+        submissions are stamped by the compile service during conversion).
+        The stamped PDF is returned so the caller can install it; nothing is
+        written to the file store here.
+
+        Parameters
+        ----------
+        pdf_bytes : bytes
+            The unstamped PDF.
+        watermark_text : str
+            Stamp text, e.g. ``arXiv:submit/1234567  [cs.LG]  15 Jan 2026``.
+        watermark_link : Optional[str]
+            Optional link embedded in the stamp.
+
+        Returns
+        -------
+        bytes
+            The stamped PDF.
+
+        Raises
+        ------
+        Exception
+            If stamping fails; the caller is expected to fall back to the
+            unstamped PDF so the preview slot is never empty.
+        """
+        ...

submit_ce/api/file_store.py

@@ -167,6 +167,19 @@ def store_preview(self, submission_id: str, content: IO[bytes], chunk_size: int
         Returns checksum"""
         pass
 
+    @abstractmethod
+    def store_nostamp_preview(self, submission_id: str, content: IO[bytes],
+                              chunk_size: int = 4096) -> str:
+        """Store the *unstamped* PDF at ``<submission_id>-nostamp.pdf``.
+
+        Mirrors how TeX2PDF keeps both a stamped and an unstamped copy for
+        TeX submissions. For PDF-only submissions Submit 2.0 writes the
+        unstamped upload here and the stamped copy to the preview slot
+        (``store_preview`` / ``<submission_id>.pdf``).
+
+        Returns checksum"""
+        pass
+
     @abstractmethod
     def store_directives(self, submission_id: str, content: dict) -> str:
         """Store directives.json for a submission.

submit_ce/domain/event/process.py

@@ -1,5 +1,7 @@
 """Events related to external or long-running processes."""
-from datetime import datetime
+import io
+import logging
+from datetime import datetime, timezone
 import json
 from typing import Optional
 
@@ -9,13 +11,17 @@
 from pydantic import BaseModel
 
 from ..exceptions import InvalidEvent
+from ..preview import Preview
 from ..submission import Submission
 from ..process import ProcessStatus
+from ..uploads import SourceFormat
 from .base import Event, EventWithSideEffect
 
 
 from submit_ce.api import SubmitApi
 
+logger = logging.getLogger(__name__)
+
 
 class ProcessInfo(BaseModel):
     process_id: str
@@ -155,6 +161,125 @@ def project(self, submission: Submission) -> Submission:
         return submission
 
 
+def _stamp_text_and_link(submission: Submission) -> tuple[str, Optional[str]]:
+    """Build the temporary submission stamp text and link.
+
+    Mirrors ``arXiv::Submit::Util::stamp_and_link`` (arxiv-lib):
+    ``arXiv:submit/<id>  [<primary category>]  <D Mon YYYY>`` -- two spaces
+    between segments, category in brackets, day not zero-padded, three-letter
+    month. A working submission has no submit_time yet, so the date is "now".
+    """
+    sid = submission.submission_id
+    text = f"arXiv:submit/{sid}"
+    try:
+        category = submission.primary_category
+    except Exception:
+        category = None
+    if category:
+        text += f"  [{category}]"
+    now = datetime.now(timezone.utc)
+    text += f"  {now.day} {now.strftime('%b %Y')}"
+    link = f"https://arxiv.org/submit/{sid}/pdf"
+    return text, link
+
+
+class InstallPdfPreview(EventWithSideEffect):
+    """Install a PDF-only submission's PDF as its (stamped) preview, under lock.
+
+    PDF-only submissions never run ``/convert``, so the stamped + unstamped
+    PDFs that TeX2PDF produces for TeX submissions must be produced on the
+    Submit 2.0 side. Running as an :class:`.EventWithSideEffect`,
+    ``SubmitApi.save`` holds the submission row lock for the whole operation
+    (see the "critical section" note in ``CLAUDE.md``), so a concurrent
+    upload/delete cannot change the file set mid-install. Under the lock this:
+
+    1. copies the single uploaded PDF to the unstamped slot
+       ``<id>-nostamp.pdf``;
+    2. calls the stamp service to watermark it with the temporary submission
+       stamp;
+    3. writes the stamped PDF to the preview slot ``<id>.pdf`` -- or, if
+       stamping fails, writes the unstamped bytes there so the preview slot is
+       never empty.
+
+    :meth:`project` marks the source processed and records the preview, the
+    same as the pre-stamping ``ConfirmSourceProcessed`` install did.
+    """
+
+    NAME = "install pdf preview"
+    NAMED = "installed pdf preview"
+
+    source_checksum: str = field(default='')
+    preview_checksum: str = field(default='')
+    size_bytes: int = field(default=-1)
+    stamped: bool = field(default=False)
+    added: Optional[datetime] = field(default=None)
+
+    def validate_pre_lock(self, submission: Submission) -> None:
+        """Only applies to PDF-only submissions with an id."""
+        if not submission.submission_id:
+            raise InvalidEvent(
+                self, "Cannot install PDF preview: submission has no id.")
+        if submission.source_format != SourceFormat.PDF:
+            raise InvalidEvent(
+                self, "InstallPdfPreview only applies to PDF-only submissions.")
+
+    def execute(self, api: 'SubmitApi', submission: Submission) -> None:
+        """Install unstamped + stamped PDFs from the single uploaded PDF."""
+        file_store = api.get_file_store()
+        sid = submission.submission_id
+        workspace = file_store.get_workspace(submission_id=sid)
+        pdfs = [f for f in (workspace.files if workspace else [])
+                if f.name.lower().endswith('.pdf')]
+        if len(pdfs) != 1:
+            logger.warning(
+                "InstallPdfPreview: expected exactly one PDF for %s, found "
+                "%d; skipping install", sid, len(pdfs))
+            return
+
+        pdf = pdfs[0]
+        source = file_store.get_source_file(sid, pdf.path)
+        with source.open('rb') as stream:
+            data = stream.read()
+
+        # Always keep the unstamped copy (for the -nostamp slot and as the
+        # fallback preview if stamping fails).
+        file_store.store_nostamp_preview(sid, io.BytesIO(data))
+
+        text, link = _stamp_text_and_link(submission)
+        stamped_bytes: Optional[bytes] = None
+        try:
+            stamped_bytes = api.get_compiler().stamp(data, text, link)
+        except Exception as exc:
+            logger.error(
+                "InstallPdfPreview: stamping failed for %s: %s; installing "
+                "unstamped preview instead", sid, exc)
+
+        if stamped_bytes:
+            self.preview_checksum = file_store.store_preview(
+                sid, io.BytesIO(stamped_bytes))
+            self.stamped = True
+        else:
+            self.preview_checksum = file_store.store_preview(
+                sid, io.BytesIO(data))
+            self.stamped = False
+
+        self.source_checksum = pdf.crc32c
+        self.size_bytes = pdf.bytes
+        self.added = datetime.now(timezone.utc)
+
+    def project(self, submission: Submission) -> Submission:
+        """Mark source processed and record the preview (as ConfirmSourceProcessed did)."""
+        submission.is_source_processed = True
+        submission.preview = Preview(
+            source_id=-1,
+            source_checksum=self.source_checksum,
+            preview_checksum=self.preview_checksum,
+            size_bytes=self.size_bytes,
+            added=self.added,
+        )
+        return submission
+
+
 class StartDirectives(EventWithSideEffect):
     """Start directives generation for a submission."""
 

submit_ce/implementations/__init__.py

@@ -131,6 +131,9 @@ def get_full_outcome_path(self, submission_id: str) -> str:
     def store_preview(self, submission_id: str, content: IO[bytes], chunk_size: int) -> str:
         return "not really stored, NullFileStore"
 
+    def store_nostamp_preview(self, submission_id: str, content: IO[bytes], chunk_size: int = 4096) -> str:
+        return "not really stored, NullFileStore"
+
     def store_compile_log(self, submission_id: str, content: IO[bytes], chunk_size: int = 4096) -> str:
         return "not really stored, NullFileStore"
 

submit_ce/implementations/compile/compile_api_service.py

@@ -248,6 +248,32 @@ def is_available(self) -> bool:
             logger.error(f"Compile service at '{settings.COMPILE_API_URL}' is not available: {exc}")
             return False
 
+    @override
+    def stamp(self, pdf_bytes: bytes, watermark_text: str,
+              watermark_link: Optional[str] = None) -> bytes:
+        """Stamp a PDF via the tex2pdf ``/stamp/`` endpoint.
+
+        The endpoint takes the PDF as a multipart upload plus the watermark
+        text/link as query params, and returns the stamped PDF bytes in the
+        response body (it does not write to the bucket). Non-2xx raises, so
+        the caller falls back to the unstamped PDF.
+        """
+        query_params = {'watermark_text': watermark_text}
+        if watermark_link:
+            query_params['watermark_link'] = watermark_link
+        # No trailing slash: the deployed service routes ``/stamp`` and 307-
+        # redirects ``/stamp/`` -> ``/stamp`` (also downgrading to http, which
+        # we must not follow with the auth token). Matches how /convert,
+        # /preflight and /directives are called.
+        url = f'{settings.COMPILE_API_URL}/stamp?{urllib.parse.urlencode(query_params)}'
+        files = {'incoming': ('submission.pdf', pdf_bytes, 'application/pdf')}
+        headers = _auth_headers()
+
+        with httpx.Client(timeout=settings.COMPILE_API_CONVERT_TIMEOUT) as client:
+            response = client.post(url, files=files, headers=headers)
+        response.raise_for_status()
+        return response.content
+
     @override
     def start_directives(self,
             submission: Submission,

submit_ce/implementations/compile/mock_compile_mimesis_pdf.py

@@ -149,3 +149,14 @@ def check(self, process_id: str, user: User, client: Client) -> ProcessStatus:
     @override
     def is_available(self) -> bool:
         return True
+
+    @override
+    def stamp(self, pdf_bytes: bytes, watermark_text: str,
+              watermark_link: Optional[str] = None) -> bytes:
+        """Pretend to stamp by prepending a marker.
+
+        The marker lets tests distinguish the stamped preview from the
+        unstamped fallback without inspecting PDF pixels, and keeps stamping
+        offline/deterministic.
+        """
+        return b"STAMPED:" + pdf_bytes

submit_ce/implementations/file_store/gs_file_store.py

@@ -218,6 +218,16 @@ def store_preview(self, submission_id: str,
         blob.reload()
         return blob.crc32c
 
+    @override
+    def store_nostamp_preview(self, submission_id: str,
+                              content: IO[bytes],
+                              chunk_size: int = 4096) -> str:
+        """Store the unstamped PDF at ``<submission_id>-nostamp.pdf``."""
+        blob = self.bucket.blob(self._nostamp_preview_path(submission_id))
+        blob.upload_from_file(content)
+        blob.reload()
+        return blob.crc32c
+
     @override
     def store_directives(self, submission_id: str, content: dict) -> str:
         """Store directives.json for a submission."""
@@ -608,6 +618,9 @@ def _source_package_path(self, submission_id: str) -> str:
     def _preview_path(self, submission_id: str) -> str:
         return posixpath.join(self._submission_path(submission_id), f'{submission_id}.pdf')
 
+    def _nostamp_preview_path(self, submission_id: str) -> str:
+        return posixpath.join(self._submission_path(submission_id), f'{submission_id}-nostamp.pdf')
+
     def _preflight_path(self, submission_id: str) -> str:
         return posixpath.join(self._submission_path(submission_id), 'gcp_preflight.json')
 

submit_ce/implementations/file_store/mock_file_store.py

@@ -92,6 +92,7 @@ def __init__(self) -> None:
         self._source: Dict[str, Dict[str, bytes]] = {}
         # single-blob slots
         self._preview: Dict[str, bytes] = {}
+        self._nostamp_preview: Dict[str, bytes] = {}
         self._preflight: Dict[str, bytes] = {}
         self._directives: Dict[str, bytes] = {}
         self._user_decisions: Dict[str, bytes] = {}
@@ -199,6 +200,11 @@ def store_preview(self, submission_id: str, content: IO[bytes],
         self._preview[submission_id] = content.read()
         return ""
 
+    def store_nostamp_preview(self, submission_id: str, content: IO[bytes],
+                              chunk_size: int = 4096) -> str:
+        self._nostamp_preview[submission_id] = content.read()
+        return ""
+
     def get_preview(self, submission_id: str) -> FileObj:
         data = self._preview.get(submission_id)
         return _InMemoryFileObj("preview.pdf", data) if data is not None else FileDoesNotExist(submission_id)