diff --git a/docs/modules/program-certificates.md b/docs/modules/program-certificates.md
index e9e8bd25..3471c4b9 100644
--- a/docs/modules/program-certificates.md
+++ b/docs/modules/program-certificates.md
@@ -19,7 +19,8 @@ JSON использует snake_case; штатный Angular API interceptor п
 - PUT /: `{revision,draft}`. Конфликт ревизии 409, невалидные поля 400.
 - POST assets/: multipart `file`, `kind=background|signature`; проверка реального
   содержимого PNG/JPEG, до 10 MB и 20 MP; подпись PNG. GET assets/{asset_id}/.
-- POST preview/: `{revision}` → PDF с именем текущего организатора, без выдачи.
+- POST preview/: `{revision,long_name?: boolean}` → PDF с тем же примером имени,
+  что в редакторе; `long_name` выбирает длинное ФИО. Без выдачи участнику.
 - POST test/: `{revision}` → письмо с пробным PDF только текущему организатору.
 - POST activate/: `{revision}` → неизменяемый снимок шаблона и рассылки;
   `timing=manual` запускает сейчас, `scheduled` требует дату не ранее окончания.
@@ -34,7 +35,8 @@ JSON использует snake_case; штатный Angular API interceptor п
   из письма (случайный UUID, без раскрытия списка участников).
 
 Celery beat раз в минуту подбирает запуски и сохраняемую очередь. Уникальный
-ключ program/user исключает повторную выдачу. Старые документы неизменяемы.
+ключ program/user исключает повторную выдачу. Сохранение настроек само по себе
+не меняет выданные документы. Их обновление требует явного действия менеджера.
 Отправка с неопределенным результатом после падения worker требует ручного
 повтора, чтобы автоматически не дублировать письма. `sent` означает принятие
 почтовым провайдером, а не доказательство попадания во входящие.
@@ -98,3 +100,26 @@ Pillow и PDF используют одни и те же локальные TTF,
 синтетического полужирного начертания. Шрифт изображения подписи не применяется.
 Миграция и обновление ранее выданных PDF не требуются. Источники и лицензии:
 `assets/certificates/fonts/README.md`.
+
+## Управление выданными сертификатами
+
+- Журнал поддерживает `unsent` и возвращает `outdated` у строки, счетчики
+  `unsent`/`outdated`. Сравнивается визуальный шаблон, без настроек письма и расписания.
+- `POST selection/`: `{revision,action,ids}` либо
+  `{revision,action,all_matching:true,status,search,excluded_ids}`. Выбор всех
+  охватывает весь фильтр, а не страницу. ID другой программы, конфликт ревизии
+  и строки в обработке отклоняются до изменений.
+- `action=send` отправляет выбранным с текущим шаблоном, пропуская `sent`.
+  `refresh` пересоздает PDF без писем; `retry` сохраняет прежний способ доставки
+  каждой строки, включая режим только на платформе. Ответ: `{count,skipped}`.
+- Ручные задания используют отдельные неизменяемые снимки с `manual_action` и
+  не подменяют управление автоматическим запуском. Обрабатываются штатным outbox.
+- `POST deliveries/{id}/pdf/` с `{revision}` применяет текущий шаблон при
+  необходимости, сохраняет PDF и возвращает его. `POST archive/` с ревизией
+  и тем же выбором делает это для ZIP. Секретная ссылка участника сохраняется.
+  Новая версия получает статус `ready`, письмо само не отправляется.
+- Старые GET download/archive и retry остаются совместимыми. Повторное скачивание
+  текущего PDF не пересоздает файл. Ошибка генерации откатывает изменения транзакции.
+
+Миграция БД не требуется. Сначала API, затем frontend. Отправление проверяется
+отдельно от готовности PDF: `ready` означает доступность на платформе, не письмо.
diff --git a/partner_programs/certificate_views.py b/partner_programs/certificate_views.py
index 7b78ee6a..f16b418f 100644
--- a/partner_programs/certificate_views.py
+++ b/partner_programs/certificate_views.py
@@ -20,7 +20,7 @@
     ProgramCertificateConfig,
 )
 from partner_programs.management_views import ManagementBase
-from partner_programs.services.management import managed_program, person_name
+from partner_programs.services.management import managed_program
 from partner_programs.services import certificates as service
 
 
@@ -83,10 +83,13 @@ def post(self, request, pk):
         program = managed_program(pk, request.user)
         config = service.config_for(program)
         revision_check(config, request.data)
+        long_name = serializers.BooleanField(default=False).run_validation(
+            request.data.get("long_name", False)
+        )
         content = service.render_pdf(
             config.draft,
             service.assets_for(program, config.draft),
-            person_name(request.user),
+            "Александра Константиновна Добровольская" if long_name else "Имя участника",
             program.name,
             "ПРОБНЫЙ",
             timezone.localtime(),
@@ -127,7 +130,11 @@ def post(self, request, pk):
         config = service.config_for(program)
         revision_check(config, request.data)
         draft = service.validate_draft(program, config.draft)
-        if ProgramCertificateBatch.objects.filter(program=program, active=True).exists():
+        if (
+            ProgramCertificateBatch.objects.filter(program=program, active=True)
+            .exclude(snapshot__has_key="manual_action")
+            .exists()
+        ):
             raise ValidationError("Сначала приостановите текущую выдачу.")
         if ProgramCertificate.objects.filter(
             program=program, status__in=["pending", "generating", "sending", "failed"]
@@ -171,6 +178,7 @@ def post(self, request, pk, action):
         batch = (
             ProgramCertificateBatch.objects.select_for_update()
             .filter(program=program)
+            .exclude(snapshot__has_key="manual_action")
             .order_by("-pk")
             .first()
         )
@@ -182,7 +190,9 @@ def post(self, request, pk, action):
 
 
 class JournalQuery(serializers.Serializer):
-    status = serializers.ChoiceField(choices=["all", "sent", "failed"], default="all")
+    status = serializers.ChoiceField(
+        choices=["all", "sent", "failed", "unsent"], default="all"
+    )
     search = serializers.CharField(max_length=200, allow_blank=True, default="")
     limit = serializers.IntegerField(min_value=1, max_value=100, default=20)
     offset = serializers.IntegerField(min_value=0, default=0)
@@ -193,7 +203,9 @@ def journal_rows(program, params):
     query.is_valid(raise_exception=True)
     data = query.validated_data
     rows = ProgramCertificate.objects.filter(program=program).defer("pdf")
-    if data["status"] != "all":
+    if data["status"] == "unsent":
+        rows = rows.exclude(status="sent")
+    elif data["status"] != "all":
         rows = rows.filter(status=data["status"])
     if data["search"]:
         rows = rows.filter(
@@ -206,11 +218,20 @@ class CertificateJournalView(ManagementBase):
     def get(self, request, pk):
         program = managed_program(pk, request.user)
         rows, query = journal_rows(program, request.query_params)
+        draft = service.config_for(program).draft
+        stale_batches = [
+            batch.pk
+            for batch in ProgramCertificateBatch.objects.filter(program=program)
+            if service.template_values(batch.snapshot, program.name)
+            != service.template_values(draft, program.name)
+        ]
         summary = ProgramCertificate.objects.filter(program=program).aggregate(
             total=Count("pk"),
             ready=Count("pk", filter=Q(pdf__isnull=False)),
             sent=Count("pk", filter=Q(status="sent")),
             failed=Count("pk", filter=Q(status="failed")),
+            unsent=Count("pk", filter=~Q(status="sent")),
+            outdated=Count("pk", filter=Q(batch_id__in=stale_batches)),
         )
         offset = query["offset"]
         end = offset + query["limit"]
@@ -223,12 +244,97 @@ def get(self, request, pk):
                 "pdf_ready": row.pdf_ready,
                 "error": row.error,
                 "sent_at": row.sent_at,
+                "outdated": row.batch_id in stale_batches,
             }
             for row in rows.annotate(pdf_ready=~Q(pdf__isnull=True))[offset:end]
         ]
         return Response({"count": rows.count(), "summary": summary, "results": result})
 
 
+def selected_rows(program, data):
+    class Selection(serializers.Serializer):
+        all_matching = serializers.BooleanField(default=False)
+        ids = serializers.ListField(
+            child=serializers.UUIDField(), required=False, min_length=1, max_length=1000
+        )
+        excluded_ids = serializers.ListField(
+            child=serializers.UUIDField(), default=list, max_length=1000
+        )
+
+    selection = Selection(data=data)
+    selection.is_valid(raise_exception=True)
+    values = selection.validated_data
+    if bool(values["all_matching"]) == bool(values.get("ids")):
+        raise ValidationError("Выберите участников.")
+    if values["all_matching"]:
+        rows, _ = journal_rows(program, data)
+        return rows.exclude(pk__in=values["excluded_ids"])
+    rows = ProgramCertificate.objects.filter(program=program, pk__in=values["ids"])
+    if rows.count() != len(set(values["ids"])):
+        raise ValidationError("Участник не найден в этой программе.")
+    return rows
+
+
+class CertificateSelectionView(ManagementBase):
+    @transaction.atomic
+    def post(self, request, pk):
+        program = managed_program(pk, request.user, lock=True)
+        config = service.config_for(program)
+        revision_check(config, request.data)
+        action = serializers.ChoiceField(
+            choices=["send", "refresh", "retry"]
+        ).run_validation(request.data.get("action"))
+        rows = list(
+            selected_rows(program, request.data)
+            .select_for_update(of=("self",))
+            .select_related("batch", "program")
+            .defer("pdf")
+            .order_by("pk")
+        )
+        if any(row.status in ("pending", "generating", "sending") for row in rows):
+            raise ValidationError("Дождитесь завершения текущей обработки сертификатов.")
+        selected = [row for row in rows if action == "refresh" or row.status != "sent"]
+        if not selected:
+            return Response({"count": 0, "skipped": len(rows)}, status=202)
+        draft = service.validate_draft(program, config.draft)
+        regenerate = [
+            row.pk
+            for row in selected
+            if action == "refresh" or service.outdated_certificate(row, draft)
+        ]
+        for send_email in (False, True):
+            group = [
+                row.pk
+                for row in selected
+                if (
+                    action == "send"
+                    or (action == "retry" and row.batch.snapshot["send_email"])
+                )
+                == send_email
+            ]
+            if not group:
+                continue
+            batch = service.manual_batch(
+                program,
+                draft,
+                request.build_absolute_uri("/").rstrip("/"),
+                "send" if send_email else "refresh",
+            )
+            targets = ProgramCertificate.objects.filter(pk__in=group)
+            targets.update(
+                batch=batch,
+                status="pending",
+                error="",
+                queued_at=None,
+                claimed_at=None,
+                sent_at=None,
+            )
+            targets.filter(pk__in=regenerate).update(pdf=None)
+        return Response(
+            {"count": len(selected), "skipped": len(rows) - len(selected)}, status=202
+        )
+
+
 class CertificateRetryView(ManagementBase):
     @transaction.atomic
     def post(self, request, pk):
@@ -247,7 +353,46 @@ def post(self, request, pk):
         return Response({"count": count}, status=202)
 
 
+def current_certificate_pdf(program, config, row, public_base_url):
+    if row.status in ("pending", "generating", "sending"):
+        raise ValidationError("Сертификат еще готовится. Дождитесь завершения обработки.")
+    if not row.pdf or service.outdated_certificate(row, config.draft):
+        draft = service.validate_draft(program, config.draft)
+        content = service.render_pdf(
+            draft,
+            service.assets_for(program, draft),
+            row.name,
+            program.name,
+            str(row.pk).upper(),
+            timezone.localtime(row.created_at),
+        )
+        row.batch = service.manual_batch(program, draft, public_base_url, "refresh")
+        row.pdf = content
+        row.status = "ready"
+        row.error = ""
+        row.sent_at = None
+        row.save(update_fields=["batch", "pdf", "status", "error", "sent_at"])
+    return bytes(row.pdf)
+
+
 class CertificateDownloadView(ManagementBase):
+    @transaction.atomic
+    def post(self, request, pk, certificate_id):
+        program = managed_program(pk, request.user, lock=True)
+        config = service.config_for(program)
+        revision_check(config, request.data)
+        row = get_object_or_404(
+            ProgramCertificate.objects.select_for_update(of=("self",)).select_related(
+                "batch", "program"
+            ),
+            program=program,
+            pk=certificate_id,
+        )
+        content = current_certificate_pdf(
+            program, config, row, request.build_absolute_uri("/").rstrip("/")
+        )
+        return pdf_response(content, "certificate-" + str(row.pk)[:8])
+
     def get(self, request, pk, certificate_id):
         program = managed_program(pk, request.user)
         row = get_object_or_404(
@@ -257,6 +402,39 @@ def get(self, request, pk, certificate_id):
 
 
 class CertificateArchiveView(ManagementBase):
+    @transaction.atomic
+    def post(self, request, pk):
+        import tempfile
+        from django.http import FileResponse
+
+        program = managed_program(pk, request.user, lock=True)
+        config = service.config_for(program)
+        revision_check(config, request.data)
+        rows = (
+            selected_rows(program, request.data)
+            .select_for_update(of=("self",))
+            .select_related("batch", "program")
+        )
+        if rows.filter(status__in=["pending", "generating", "sending"]).exists():
+            raise ValidationError(
+                "Дождитесь завершения обработки выбранных сертификатов."
+            )
+        output = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024)
+        try:
+            with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
+                for row in rows.iterator(chunk_size=20):
+                    content = current_certificate_pdf(
+                        program, config, row, request.build_absolute_uri("/").rstrip("/")
+                    )
+                    archive.writestr(f"certificate-{row.pk}.pdf", content)
+            output.seek(0)
+        except Exception:
+            output.close()
+            raise
+        response = FileResponse(output, as_attachment=True, filename="certificates.zip")
+        response["Cache-Control"] = "private, no-store"
+        return response
+
     def get(self, request, pk):
         program = managed_program(pk, request.user)
         rows, _ = journal_rows(program, request.query_params)
diff --git a/partner_programs/services/certificates.py b/partner_programs/services/certificates.py
index f93ea52d..4a31bf87 100644
--- a/partner_programs/services/certificates.py
+++ b/partner_programs/services/certificates.py
@@ -166,7 +166,10 @@ def assets_for(program, draft):
 def state(program):
     config = config_for(program)
     batch = (
-        ProgramCertificateBatch.objects.filter(program=program).order_by("-pk").first()
+        ProgramCertificateBatch.objects.filter(program=program)
+        .exclude(snapshot__has_key="manual_action")
+        .order_by("-pk")
+        .first()
     )
     return {
         "revision": config.revision,
@@ -189,6 +192,47 @@ def state(program):
     }
 
 
+def template_values(draft, program_name):
+    normalized = normalized_draft(draft)
+    return {
+        key: normalized[key]
+        for key in (
+            "orientation",
+            "background_id",
+            "signature_id",
+            "director",
+            "position",
+            "hidden_elements",
+            "layout",
+        )
+    } | {"program_name": draft.get("program_name", program_name)}
+
+
+def outdated_certificate(row, draft):
+    return template_values(row.batch.snapshot, row.program.name) != template_values(
+        draft, row.program.name
+    )
+
+
+def manual_batch(program, draft, public_base_url, action):
+    snapshot = {
+        **draft,
+        "program_name": program.name,
+        "manual_action": action,
+        "send_email": action == "send",
+        "timing": "manual",
+        "scheduled_at": None,
+    }
+    return ProgramCertificateBatch.objects.create(
+        program=program,
+        snapshot=snapshot,
+        **assets_for(program, draft),
+        issue_at=timezone.now(),
+        initialized_at=timezone.now(),
+        public_base_url=public_base_url,
+    )
+
+
 def upload_asset(program, kind, upload):
     if kind not in ("background", "signature") or not upload:
         raise ValidationError("Выберите фон или подпись и файл.")
diff --git a/partner_programs/tests/test_certificates.py b/partner_programs/tests/test_certificates.py
index e6387c16..07d38177 100644
--- a/partner_programs/tests/test_certificates.py
+++ b/partner_programs/tests/test_certificates.py
@@ -93,6 +93,224 @@ def image(self, kind="PNG"):
             "signature.png", buffer.getvalue(), content_type="image/png"
         )
 
+    @patch(
+        "partner_programs.services.certificates.render_pdf", return_value=b"%PDF-preview"
+    )
+    def test_preview_matches_the_editor_name_and_long_name_switch(self, render):
+        self.assertEqual(
+            self.client.post(
+                self.root + "preview/", {"revision": 1, "long_name": True}
+            ).status_code,
+            200,
+        )
+        self.assertEqual(
+            render.call_args.args[2], "Александра Константиновна Добровольская"
+        )
+        self.client.post(self.root + "preview/", {"revision": 1, "long_name": False})
+        self.assertEqual(render.call_args.args[2], "Имя участника")
+        self.assertEqual(
+            self.client.post(
+                self.root + "preview/", {"revision": 1, "long_name": "bad"}
+            ).status_code,
+            400,
+        )
+
+    @patch(
+        "partner_programs.services.certificates.render_pdf", return_value=b"%PDF-current"
+    )
+    def test_current_download_replaces_outdated_template_and_preserves_participant_link(
+        self, render
+    ):
+        row = self.row(status="sent", pdf=b"%PDF-old", sent_at=timezone.now())
+        token = row.download_token
+        self.config.draft = deepcopy(self.config.draft)
+        self.config.draft["director"] = "Новый директор"
+        self.config.revision += 1
+        self.config.save()
+        response = self.client.post(
+            self.root + f"deliveries/{row.pk}/pdf/", {"revision": 2}
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.content, b"%PDF-current")
+        self.assertEqual(render.call_args.args[0]["director"], "Новый директор")
+        row.refresh_from_db()
+        self.assertEqual(bytes(row.pdf), b"%PDF-current")
+        self.assertEqual(row.download_token, token)
+        self.assertEqual(row.status, "ready")
+        self.assertIsNone(row.sent_at)
+        self.assertEqual(len(mail.outbox), 0)
+        self.assertFalse(service.outdated_certificate(row, self.config.draft))
+        # Downloading the same current revision again does not regenerate or send.
+        self.client.post(self.root + f"deliveries/{row.pk}/pdf/", {"revision": 2})
+        self.assertEqual(render.call_count, 1)
+
+    @patch(
+        "partner_programs.services.certificates.render_pdf", return_value=b"%PDF-current"
+    )
+    def test_selected_send_is_scoped_and_does_not_duplicate_sent_mail(self, render):
+        ready = self.row(status="ready", pdf=b"%PDF-ready")
+        sent = self.row(
+            user=self.member, status="sent", pdf=b"%PDF-sent", sent_at=timezone.now()
+        )
+        foreign = self.row(
+            program=self.foreign,
+            user=self.only_registered,
+            status="ready",
+            pdf=b"%PDF-foreign",
+        )
+        payload = {"revision": 1, "action": "send", "ids": [str(ready.pk), str(sent.pk)]}
+        response = self.client.post(self.root + "selection/", payload, format="json")
+        self.assertEqual(response.status_code, 202, response.data)
+        self.assertEqual(response.data, {"count": 1, "skipped": 1})
+        service.process(ready.pk)
+        ready.refresh_from_db()
+        self.assertEqual(ready.status, "sent")
+        self.assertEqual(len(mail.outbox), 1)
+        self.assertEqual(mail.outbox[0].to, [ready.email])
+        self.assertEqual(
+            self.client.post(self.root + "selection/", payload, format="json").data,
+            {"count": 0, "skipped": 2},
+        )
+        service.process(ready.pk)
+        self.assertEqual(len(mail.outbox), 1)
+        foreign.refresh_from_db()
+        self.assertEqual(foreign.status, "ready")
+        self.assertEqual(bytes(foreign.pdf), b"%PDF-foreign")
+
+    @patch(
+        "partner_programs.services.certificates.render_pdf",
+        side_effect=RuntimeError("renderer unavailable"),
+    )
+    def test_failed_current_download_preserves_previously_issued_file(self, render):
+        row = self.row(status="sent", pdf=b"%PDF-retained", sent_at=timezone.now())
+        original_batch = row.batch_id
+        self.config.draft = deepcopy(self.config.draft)
+        self.config.draft["director"] = "Новый директор"
+        self.config.save()
+        with self.assertRaises(RuntimeError):
+            self.client.post(self.root + f"deliveries/{row.pk}/pdf/", {"revision": 1})
+        row.refresh_from_db()
+        self.assertEqual(bytes(row.pdf), b"%PDF-retained")
+        self.assertEqual(row.batch_id, original_batch)
+        self.assertEqual(row.status, "sent")
+        self.assertIsNotNone(row.sent_at)
+
+    def test_select_all_uses_the_filter_and_exclusions_not_pagination(self):
+        first = self.row(status="failed", pdf=b"%PDF-first")
+        second = self.row(user=self.member, status="failed", pdf=b"%PDF-second")
+        third = self.row(user=self.only_registered, status="sent", pdf=b"%PDF-third")
+        response = self.client.post(
+            self.root + "selection/",
+            {
+                "revision": 1,
+                "action": "refresh",
+                "all_matching": True,
+                "status": "failed",
+                "limit": 1,
+                "excluded_ids": [str(first.pk)],
+            },
+            format="json",
+        )
+        self.assertEqual(response.status_code, 202, response.data)
+        self.assertEqual(response.data["count"], 1)
+        first.refresh_from_db()
+        second.refresh_from_db()
+        third.refresh_from_db()
+        self.assertEqual(first.status, "failed")
+        self.assertEqual(second.status, "pending")
+        self.assertIsNone(second.pdf)
+        self.assertFalse(second.batch.snapshot["send_email"])
+        self.assertEqual(third.status, "sent")
+        # Manual jobs must not replace automatic issuance controls.
+        self.assertNotEqual(service.state(self.program)["batch"]["id"], second.batch_id)
+
+    @patch(
+        "partner_programs.services.certificates.render_pdf", return_value=b"%PDF-current"
+    )
+    def test_selected_retry_preserves_platform_only_and_email_delivery_modes(
+        self, render
+    ):
+        platform_draft = deepcopy(self.config.draft)
+        platform_draft["send_email"] = False
+        platform = self.row(status="failed", batch=self.batch(snapshot=platform_draft))
+        email = self.row(user=self.member, status="failed")
+        response = self.client.post(
+            self.root + "selection/",
+            {"revision": 1, "action": "retry", "all_matching": True, "status": "failed"},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 202, response.data)
+        for row in (platform, email):
+            service.process(row.pk)
+            row.refresh_from_db()
+        self.assertEqual(platform.status, "ready")
+        self.assertEqual(email.status, "sent")
+        self.assertEqual(len(mail.outbox), 1)
+        self.assertEqual(mail.outbox[0].to, [email.email])
+
+    def test_selection_rejects_foreign_ids_stale_revision_and_in_flight_rows(self):
+        row = self.row(status="sending", pdf=b"%PDF-inflight")
+        foreign = self.row(
+            program=self.foreign, user=self.only_registered, status="ready"
+        )
+        for payload, status in [
+            ({"revision": 1, "action": "send", "ids": [str(foreign.pk)]}, 400),
+            ({"revision": 0, "action": "send", "ids": [str(row.pk)]}, 400),
+            ({"revision": 2, "action": "send", "ids": [str(row.pk)]}, 409),
+            ({"revision": 1, "action": "refresh", "ids": [str(row.pk)]}, 400),
+            ({"revision": 1, "action": "refresh", "ids": []}, 400),
+            (
+                {
+                    "revision": 1,
+                    "action": "refresh",
+                    "ids": [str(row.pk)],
+                    "all_matching": True,
+                },
+                400,
+            ),
+        ]:
+            self.assertEqual(
+                self.client.post(
+                    self.root + "selection/", payload, format="json"
+                ).status_code,
+                status,
+            )
+        row.refresh_from_db()
+        self.assertEqual(row.status, "sending")
+        self.assertEqual(bytes(row.pdf), b"%PDF-inflight")
+        self.client.force_authenticate(self.other)
+        self.assertEqual(
+            self.client.post(
+                self.root + "selection/",
+                {"revision": 1, "action": "send", "all_matching": True},
+                format="json",
+            ).status_code,
+            403,
+        )
+
+    @patch(
+        "partner_programs.services.certificates.render_pdf", return_value=b"%PDF-current"
+    )
+    def test_current_archive_applies_template_and_respects_selected_rows(self, render):
+        import zipfile
+
+        first = self.row(status="ready", pdf=b"%PDF-old")
+        second = self.row(user=self.member, status="ready", pdf=b"%PDF-untouched")
+        self.config.draft = deepcopy(self.config.draft)
+        self.config.draft["director"] = "Новый директор"
+        self.config.save()
+        response = self.client.post(
+            self.root + "archive/", {"revision": 1, "ids": [str(first.pk)]}, format="json"
+        )
+        self.assertEqual(response.status_code, 200)
+        with zipfile.ZipFile(io.BytesIO(b"".join(response.streaming_content))) as archive:
+            self.assertEqual(archive.namelist(), [f"certificate-{first.pk}.pdf"])
+            self.assertEqual(archive.read(archive.namelist()[0]), b"%PDF-current")
+        first.refresh_from_db()
+        second.refresh_from_db()
+        self.assertEqual(bytes(first.pdf), b"%PDF-current")
+        self.assertEqual(bytes(second.pdf), b"%PDF-untouched")
+
     def test_access_requires_program_manager_and_does_not_leak_email(self):
         for user, status in [
             (None, 401),
@@ -627,7 +845,8 @@ def test_journal_filters_search_pagination_and_summary(self):
         self.assertEqual(str(response.data["results"][0]["id"]), str(row.pk))
         self.assertTrue(response.data["results"][0]["pdf_ready"])
         self.assertEqual(
-            response.data["summary"], {"total": 2, "ready": 2, "sent": 1, "failed": 1}
+            response.data["summary"],
+            {"total": 2, "ready": 2, "sent": 1, "failed": 1, "unsent": 1, "outdated": 0},
         )
         self.assertEqual(
             len(self.client.get(self.root + "deliveries/?status=sent").data["results"]), 1
@@ -776,6 +995,39 @@ def run():
             for future in futures:
                 future.result(timeout=20)
 
+    @skipUnlessDBFeature("has_select_for_update")
+    @patch("partner_programs.services.certificates.send_certificate")
+    def test_selected_send_is_claimed_once_with_concurrent_manager_requests(self, send):
+        self.program.managers.add(self.user)
+        config = service.config_for(self.program)
+        row = ProgramCertificate.objects.create(
+            program=self.program,
+            batch=self.batch,
+            user=self.user,
+            name="Тестовый участник",
+            email=self.user.email,
+            status="ready",
+            pdf=b"%PDF",
+        )
+        responses = []
+
+        def select():
+            client = APIClient()
+            client.force_authenticate(self.user)
+            response = client.post(
+                f"/programs/{self.program.pk}/management/certificates/selection/",
+                {"revision": config.revision, "action": "send", "ids": [str(row.pk)]},
+                format="json",
+            )
+            responses.append(response.status_code)
+
+        self.parallel(select)
+        self.assertEqual(sorted(responses), [202, 400])
+        self.parallel(lambda: service.process(row.pk))
+        send.assert_called_once()
+        row.refresh_from_db()
+        self.assertEqual(row.status, "sent")
+
     @skipUnlessDBFeature("has_select_for_update")
     def test_concurrent_initialization_has_one_certificate_per_participant(self):
         self.parallel(lambda: service.initialize_batch(self.batch.pk))
diff --git a/partner_programs/urls.py b/partner_programs/urls.py
index 8fad64e2..1eee6b67 100644
--- a/partner_programs/urls.py
+++ b/partner_programs/urls.py
@@ -98,6 +98,7 @@
     CertificateQueueView,
     CertificateJournalView,
     CertificateRetryView,
+    CertificateSelectionView,
     CertificateDownloadView,
     CertificateArchiveView,
     MyCertificatesView,
@@ -127,6 +128,9 @@
         "<int:pk>/management/certificates/deliveries/", CertificateJournalView.as_view()
     ),
     path("<int:pk>/management/certificates/retry/", CertificateRetryView.as_view()),
+    path(
+        "<int:pk>/management/certificates/selection/", CertificateSelectionView.as_view()
+    ),
     path(
         "<int:pk>/management/certificates/deliveries/<uuid:certificate_id>/pdf/",
         CertificateDownloadView.as_view(),