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,21 @@ def store_preview(self, submission_id: str, content: IO[bytes], chunk_size: int
         Returns checksum"""
         pass
 
+    @abstractmethod
+    def does_nostamp_preview_exist(self, submission_id: str) -> bool:
+        """Whether an unstamped PDF exists at ``<submission_id>-nostamp.pdf``.
+
+        For TeX submissions TeX2PDF writes this alongside the stamped preview.
+        PDF-only does not store one (the original PDF lives in ``src/``); this
+        lets the PDF-only install detect and clean up a stale copy left by a
+        prior TeX compile."""
+        pass
+
+    @abstractmethod
+    def delete_nostamp_preview(self, submission_id: str) -> None:
+        """Delete ``<submission_id>-nostamp.pdf`` if it exists."""
+        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,168 @@ 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 preview 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. calls the stamp service to watermark the uploaded PDF with the
+       temporary submission stamp;
+    2. 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;
+    3. removes any stale ``<id>-nostamp.pdf`` left by a prior TeX compile
+       (the submitter switched from TeX to PDF-only). PDF-only keeps its
+       original PDF in ``src/``, so no dedicated unstamped copy is stored.
+       [SUBMISSION-196]
+
+    :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 validate_under_lock(self, api: 'SubmitApi', submission: Submission) -> None:
+        """Require exactly one PDF in the source workspace before installing.
+
+        Runs inside the submission row lock, so the file set cannot change
+        between this check and :meth:`execute`. For a PDF-only submission the
+        workspace is a single lone PDF (see ``_infer_source_format``); if that
+        invariant is violated we reject the event here rather than install a
+        partial or absent preview -- the caller degrades gracefully on the
+        resulting :class:`.InvalidEvent`. [SUBMISSION-196]
+        """
+        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:
+            raise InvalidEvent(
+                self,
+                f"Expected exactly one PDF for PDF-only submission {sid}, "
+                f"found {len(pdfs)}.")
+
+    def execute(self, api: 'SubmitApi', submission: Submission) -> None:
+        """Stamp the uploaded PDF, install it as the preview, and drop any
+        stale unstamped copy left by a prior TeX compile.
+
+        :meth:`validate_under_lock` has already guaranteed exactly one PDF in
+        the workspace under the same lock, so we take it directly.
+        """
+        file_store = api.get_file_store()
+        sid = submission.submission_id
+        workspace = file_store.get_workspace(submission_id=sid)
+        pdf = [f for f in workspace.files
+               if f.name.lower().endswith('.pdf')][0]
+        source = file_store.get_source_file(sid, pdf.path)
+        with source.open('rb') as stream:
+            data = stream.read()
+
+        # PDF-only keeps its original PDF in src/, so we don't persist a
+        # redundant <id>-nostamp.pdf. If a prior TeX compile left an unstamped
+        # PDF at the top-level slot (the submitter switched from TeX to
+        # PDF-only), remove it so a stale copy can't linger. The stamping
+        # fallback below uses the in-memory `data`, not this slot. [SUBMISSION-196]
+        if file_store.does_nostamp_preview_exist(sid):
+            file_store.delete_nostamp_preview(sid)
+            logger.info(
+                "InstallPdfPreview: removed stale unstamped PDF for %s", sid)
+
+        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 the source processed and record the preview *artifact*.
+
+        Records ``submission.preview`` (the preview PDF's checksums and size)
+        and sets ``is_source_processed`` -- the same as
+        ``ConfirmSourceProcessed`` did. This does NOT mark the preview as
+        viewed: the "submitter reviewed the preview" bit
+        (``submitter_confirmed_preview``, which gates Submit on the Confirm
+        page) is set only by ``ConfirmPreview`` when the submitter actually
+        opens ``/preview.pdf``. Recording ``submission.preview`` here is in
+        fact the precondition for that step -- ``ConfirmPreview.validate``
+        requires it to be present and checksum-matches it -- not a claim that
+        the submitter previewed anything. [SUBMISSION-196]
+
+        The one-PDF precondition is enforced in :meth:`validate_under_lock`, so
+        an event that reaches ``project`` has installed a real preview via
+        ``execute`` (``added`` is set).
+        """
+        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/domain/event/tests/test_install_pdf_preview.py

@@ -0,0 +1,75 @@
+"""Tests for InstallPdfPreview.validate_under_lock. [SUBMISSION-196]
+
+The event requires exactly one PDF in the source workspace. That precondition
+lives in ``validate_under_lock`` (run under the submission row lock), so a
+violated invariant rejects the event cleanly -- raising ``InvalidEvent`` --
+instead of installing a partial or absent preview.
+
+Unreachable in normal flow (``source_format == PDF`` implies a single lone PDF;
+see ``_infer_source_format``); these tests guard the invariant.
+"""
+
+from datetime import datetime
+from types import SimpleNamespace
+
+import pytest
+from pytz import UTC
+
+from submit_ce.domain import submission as submod, agent
+from submit_ce.domain.uploads import SourceFormat
+from submit_ce.domain.exceptions import InvalidEvent
+from submit_ce.domain.event.process import InstallPdfPreview
+
+
+class _FakeStore:
+    def __init__(self, files):
+        self._files = files
+
+    def get_workspace(self, submission_id):
+        return SimpleNamespace(files=self._files)
+
+
+class _FakeApi:
+    def __init__(self, files):
+        self._store = _FakeStore(files)
+
+    def get_file_store(self):
+        return self._store
+
+
+def _user(uid="u1"):
+    return agent.PublicUser(name="Test User", user_id=uid,
+                            email=f"{uid}@example.org", endorsements=[])
+
+
+def _pdf_submission(sid="1234567"):
+    u = _user()
+    s = submod.Submission(creator=u, owner=u, created=datetime.now(UTC),
+                          source_format=SourceFormat.PDF)
+    s.submission_id = sid
+    return s
+
+
+def _pdf_file(name):
+    return SimpleNamespace(name=name, path=name, crc32c="c", bytes=10)
+
+
+def _validate(files):
+    s = _pdf_submission()
+    api = _FakeApi(files)
+    InstallPdfPreview(creator=s.creator).validate_under_lock(api, s)
+
+
+def test_zero_pdfs_rejected():
+    with pytest.raises(InvalidEvent):
+        _validate([])
+
+
+def test_multiple_pdfs_rejected():
+    with pytest.raises(InvalidEvent):
+        _validate([_pdf_file("a.pdf"), _pdf_file("b.pdf")])
+
+
+def test_exactly_one_pdf_passes():
+    # Should not raise.
+    _validate([_pdf_file("paper.pdf")])

submit_ce/implementations/__init__.py

@@ -132,6 +132,12 @@ 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 does_nostamp_preview_exist(self, submission_id: str) -> bool:
+        return False
+
+    def delete_nostamp_preview(self, submission_id: str) -> None:
+        pass
+
     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