Auto email to Brevo

OPERATIONAL_AUXILIARIES.md

@@ -3,7 +3,7 @@
 ## Scope and status
 
 - Stage 1: implemented. Consent applies only to newly submitted private tokens; public submissions are unchanged.
-- Stage 2: implemented, disabled by default. Frontend-only Resend notifications.
+- Stage 2: implemented, disabled by default. Frontend-only Brevo notifications and an independently gated operator-shell test command.
 - Stage 3: **deferred at the size/complexity gate**. No production exporter, GitHub client, backup thread or GitHub configuration has been added. Do not treat the synthetic example as a backup.
 - No Worker changes, production reads/writes, real credentials, GPU execution, live emails, GitHub uploads, deployment or commits were used.
 
@@ -43,18 +43,24 @@ Set these in **Frontend**, not Worker:
 | Variable | Type | Value |
 | --- | --- | --- |
 | `EVALUATOR_ALERTS_ENABLED` | Variable | `true`; absent/other values disable |
+| `EVALUATOR_ALERT_TEST_ENABLED` | Variable | default `false`; `true` enables the operator-shell test only |
 | `EVALUATOR_SPACE_ID` | Existing Variable | the existing managed Worker `owner/space` |
-| `EVALUATOR_ALERT_RESEND_API_KEY` | Secret | Resend sending credential |
+| `EVALUATOR_ALERT_BREVO_API_KEY` | Secret | Brevo Transactional Email API key, not an SMTP password |
 | `EVALUATOR_ALERT_FROM` | Secret | one authorized sender email, without display-name syntax |
 | `EVALUATOR_ALERT_TO` | Secret | one recipient email |
 
-All values must be present and addresses valid for the helper to start. Incomplete configuration disables it. Use an authorized Resend sender/domain. Existing `httpx` suffices; no dependency changes. Provider billing/account limits are external; no live delivery is claimed. The provider is fixed to `https://api.resend.com/emails`, with redirects disabled. [Resend idempotency documentation](https://resend.com/changelog/idempotency-keys).
+All mail configuration values must be present and addresses valid for the helper to start; the two feature switches are independent and default off. Incomplete/invalid configuration disables sending. Use a sender registered and verified in Brevo; the operator must configure that verified address and Brevo enforces sender acceptance. No extra sender lookup API call is added. Existing `httpx` suffices; no dependency changes. The old `EVALUATOR_ALERT_RESEND_API_KEY` is ignored: there is no Resend fallback or dual send. Provider billing/account limits are external; no live delivery is claimed.
+
+Transport is fixed to `POST https://api.brevo.com/v3/smtp/email`, with `api-key` authentication, JSON `sender.email`, `to[].email`, `subject`, and `textContent`. Redirects are disabled. HTTP **201** is treated as API acceptance; a validated, non-secret `messageId` is retained where provided. Missing/malformed IDs are not logged or treated as a reason to resend accepted mail. API acceptance is **not** inbox delivery. See [Brevo send-email API](https://developers.brevo.com/reference/send-transac-email) and [plain-text request guide](https://developers.brevo.com/docs/send-a-transactional-email).
 
 ### Exact files/functions and boundaries
 
 `src/submission/lifecycle_alerts.py`:
 
+For the Brevo migration specifically, only `_config`, `Notifications.load`, `Notifications.consume` (receipt metadata only) and `Notifications.deliver` change. Added functions are `_mail_config`, `_message_id`, `_send_brevo`, `_save_test_record`, `send_test_email` and `main`. Lifecycle categorization/continuity, `_run`, auxiliary locking, `observe`, `start` and `stop` remain unchanged. The other two changed files are `lifecycle_alerts_regression.py` and this document.
+
 - `_config`, `start`, `stop`: opt-in lifecycle for a single daemon sender per Frontend process.
+- `_mail_config`, `_send_brevo`, `_message_id`: shared finite-timeout transport/configuration and filtered provider receipt, used by production alerts and the operator test.
 - `observe`: copies only Space ID, allowlisted runtime stage, active-work boolean, restart count/exhaustion flag and observation timestamp into a bounded in-memory queue. No record, token, error text or submission content is passed.
 - `Notifications.consume`: categorizes/deduplicates observations and persists a small outbox/incident latch.
 - `Notifications.deliver`: best-effort email via a separate HTTP client. Connect timeout 10s, read/write/pool 20s each; these are phase timeouts, not a universal wall-clock guarantee.
@@ -74,13 +80,44 @@ The helper uses only observations from existing controller calls, not an indepen
 
 ### Persistence, retry and rollback
 
-Only `/data/auxiliary/lifecycle-alerts/notifications.json` and `sender.lock` are added when enabled. No authoritative business file is changed. The JSON contains version, Space ID, three category latches and an outbox of allowlisted notification fields/receipts; no API key, sender/recipient address or submission data. Writes use temp file, fsync and replace. Invalid state fails closed without overwriting it.
+Production alerts use only `/data/auxiliary/lifecycle-alerts/notifications.json` and `sender.lock` when enabled. No authoritative business file is changed. The JSON contains version, Space ID, three category latches and an outbox of allowlisted notification fields/receipts, including provider and optional filtered message ID; no API key, sender/recipient address or submission data. Writes use temp file, fsync and replace. Invalid state fails closed without overwriting it. Operator tests use the separate directory described below.
+
+Delivery is persisted as `unknown` **before** calling Brevo. If a timeout, crash, **5xx**, or unexpected non-201 2xx occurs, it is not automatically resent: this trades possible missed email for duplicate avoidance. Definite **429** rate-limit rejection gets at most three total attempts (60s then 300s); incidents older than 23h are no longer sent/retried. This existing 23h bound is a notification-freshness rule, not a Brevo idempotency guarantee. Other failures are marked failed. Receipt-write failure after acceptance leaves the durable unknown receipt, also preventing resend.
+
+The verified Brevo documentation describes `headers.idempotencyKey` for batch emails, not Resend's HTTP `Idempotency-Key`. This implementation sends a single plain-text message and **does not assume batch idempotency applies**, invent an idempotency header, or automatically retry ambiguous 5xx. The rate-limit retry cap/delays are unchanged; the more conservative 5xx treatment is the intentional provider-migration difference. See [Brevo batch idempotency scope](https://developers.brevo.com/docs/heterogenous-versions-batch-emails).
 
-Delivery is persisted as `unknown` **before** calling Resend. If a timeout, crash or uncertain response occurs, it is not automatically resent: this trades possible missed email for duplicate avoidance. A definite 429/5xx response gets at most three total attempts using the same provider idempotency key (60s then 300s); older than 23h is no longer resent. Other non-success status codes are marked failed. HTTP 2xx means provider acceptance, not proven delivery to the inbox. Receipt-write failure after acceptance leaves the durable unknown receipt, also preventing resend.
+The reader accepts existing **v1 Resend state** and new **v2 state**. V2 adds `provider` and optional filtered `message_id` to each receipt. Migration preserves all IDs, latches, categories, delivery values, attempt counts and timestamps. Merely loading v1 does not rewrite it; conversion is persisted on a normal subsequent save. A v1 pending receipt with zero attempts may be sent once through Brevo. Any previously attempted or processed receipt is attributed to Resend; sent/unknown/failed receipts are not resent, and attempted pending Resend receipts are **parked**, not retried through Brevo. Parked pending receipts need separate operator review, not automatic replay. Normal pre-existing bounded-outbox pruning policy remains unchanged; migration itself does not delete/reset historical state.
 
 The sender processes at most one due delivery per poll. Pending receipts survive restart. An in-flight email may finish after shutdown; shutdown does not wait on it or delay existing controller shutdown. Receipts are capped at 1000 entries; completed/unknown/failed entries can be pruned on a new incident while category latches remain. If 1000 pending entries accumulate, notifications fail independently. Read size is capped at 2MB. Provider and FUSE failures remain possible; optional alerts are not a guaranteed external monitoring system.
 
-Disable with `EVALUATOR_ALERTS_ENABLED=false` (or remove configuration) and redeploy Frontend when convenient. Do not restart Worker. Keep the auxiliary state to preserve incident deduplication if re-enabled. Rollback to previous Frontend simply ignores the sidecar; never reset evaluator restart budget or delete Queue files as part of rollback.
+Disable production alerts with `EVALUATOR_ALERTS_ENABLED=false` (or remove configuration) and redeploy Frontend when convenient. Do not restart Worker. Keep auxiliary state to preserve incident deduplication if re-enabled. For rollback, disable both flags first and follow the v2 compatibility precautions below; never reset evaluator restart budget or delete Queue files as part of rollback.
+
+### Operator-only test email
+
+No established, server-authorized administrator UI was found. The publication subject allowlist is **not** an administrator role. Consequently no button, public HTTP endpoint, Gradio handler or new authentication system was added, and `app.py`/`evaluator_space.py` were not changed by this transport migration.
+
+The entry point is `python -m src.submission.lifecycle_alerts send-test`, executed **only through an already authorized Frontend container terminal/SSH/operator execution route**. Authorization is the platform/OS permission to execute code in that container, not a caller-supplied username or `--admin` flag. Ordinary/unauthenticated website researchers have no registered route to this command. Anyone with container execution and server Secret access is already a trusted operator; the CLI does not authenticate an HF session itself. If your deployment has no restricted operator terminal/SSH route, this test capability is unavailable until that platform access is approved—do not expose the function via an unauthenticated endpoint.
+
+Set `EVALUATOR_ALERT_TEST_ENABLED=true` in Frontend to enable the test; default is false. It uses the same Brevo key/from/to/Worker Space configuration, but production `EVALUATOR_ALERTS_ENABLED` can remain false while testing. There are no recipient/subject/key CLI overrides. Its subject is `[TEST] PhysInOne Leaderboard Email Alert`, and the body identifies a test, UTC timestamp and absence of any real GPU/Worker failure.
+
+The command sends at most one API request per invocation. A separate nonblocking lock and atomic receipt under `/data/auxiliary/lifecycle-alert-tests/{test-send.lock,test-send.json}` enforce **one test attempt per 60 seconds across all operators** (stronger than a per-administrator limit), surviving process restart. Failures and ambiguous outcomes consume the cooldown; test sends do not use the production outbox, incident latches or restart state and are never automatically retried. The receipt contains only version, last attempt time, outcome and optional safe message ID. Invalid/unwritable test state fails closed. A concurrent command may report unavailable rather than waiting; it cannot send a second request.
+
+Return values are sanitized: `api_accepted`, `unknown`, `failed`, `disabled`, `unconfigured`, `cooldown`, or `unavailable`. Exit code is zero only for API acceptance. A timeout/save failure after submission can leave outcome unknown; inspect Brevo/inbox before issuing another deliberate manual test. No receipt contains the API key or raw HTTP response.
+
+**After-deployment operator procedure (not executed during development):**
+
+1. Configure the Brevo API key as Frontend `EVALUATOR_ALERT_BREVO_API_KEY` Secret and set the existing sender/recipient Secrets. Use the already verified personal sender; keep keys out of commands and logs.
+2. Explicitly enable `EVALUATOR_ALERT_TEST_ENABLED=true`. Deploy Frontend only when ready; no Worker deployment/restart is required. Production alert enablement is independent.
+3. Enter the authorized **Frontend** container terminal, change to `/app`, and execute once:
+
+   ```bash
+   python -m src.submission.lifecycle_alerts send-test
+   ```
+
+4. If the output reports `api_accepted`, check Brevo transactional activity and the configured recipient inbox/spam folder. Do not claim delivery based on the API result alone. For uncertain results, check activity before retrying, and wait at least 60 seconds.
+5. Disable `EVALUATOR_ALERT_TEST_ENABLED` after testing. Changing the configured flag requires the normal Frontend configuration refresh/deployment; do not modify Bucket cooldown files to bypass it.
+
+**Minimal deployment/rollback:** deploy the three changed Frontend files (alert module, offline tests and this document) and replace the key Secret; no new dependencies or Worker changes. Keep both optional flags false until deliberately enabled. The old provider key may be retained securely for a future reviewed rollback but is never read by this version. For immediate rollback, disable both features before reverting Frontend. Old Resend-only code cannot read persisted v2 receipts and fails closed; do not delete state or blindly restore an older snapshot, which could replay already processed incidents. Re-enabling Resend after rollback requires separately reviewed state compatibility handling; evaluation remains unaffected with notifications disabled.
 
 ## Stage 3 — gate results and explicit deferral
 
@@ -175,7 +212,7 @@ Restore must remain separately authorized: validate offline, reconstruct per-sou
 ## Offline verification
 
 - `token_consent_regression.py`: 8/8 (strict consent, public/private paths, identity/Write role, encryption/recovery, all terminal token stripping, exception non-disclosure, UI input/output forwarding and reset).
-- `lifecycle_alerts_regression.py`: 14/14 (three events, normal-state suppression, continuity/restart, deduplication/new-budget episodes, bounded/ambiguous HTTP failure, invalid sidecar, persist-before-send, sanitized payload and controller isolation).
+- `lifecycle_alerts_regression.py`: 26/26 after the Brevo migration (three events, continuity/restart/deduplication, Brevo payload/message ID, configuration validation, conservative 429/5xx/timeouts, v1 compatibility, independent operator-test cooldown/configuration/persistence failure and no public test registration).
 - Existing `lifecycle_regression.py`: 49/49; `identity_history_regression.py`: 21/21; `stable_identity_regression.py`: 27/27; `publication_order_regression.py`: 25/25.
 - Existing `staging_regression.py`: 15/16; the previously known Future Prediction primary metric mismatch remains (`Worker manifest pmf` vs `Frontend pmf_novel`). It was not introduced or changed by this task.
 - Existing `traceability_regression.py`: 33/34 in the full suite. `test_multiple_entries_keep_individual_scores_and_metadata` passes alone (1/1), but the full suite leaves `Model B` presentation metadata from the preceding metadata test in its shared fixture; the subsequent test expects `Method 0123456789`. Its setup does not clear presentation metadata. The involved publication/metadata readers and tests are unchanged; this fixture-isolation issue was not fixed as part of this task.

lifecycle_alerts_regression.py

@@ -1,5 +1,7 @@
 """Offline email observations, durable deduplication and isolated delivery."""
 import json
+import ast
+import copy
 import os
 from pathlib import Path
 import socket
@@ -113,13 +115,20 @@ class AlertChecks(unittest.TestCase):
             requests.append(request)
             self.assertTrue(controller._mutex.acquire(blocking=False))
             controller._mutex.release()
-            return httpx.Response(200, json={"id": "synthetic"})
+            return httpx.Response(201, json={"messageId": "<synthetic@relay.example.org>"})
         self.deliver(respond)
         self.deliver(respond)
         self.assertEqual(len(requests), 1)
         payload = json.loads(requests[0].content)
         self.assertNotIn(self.config["key"], json.dumps(payload) + (self.directory / "notifications.json").read_text())
-        self.assertIn("process crash is not inferred", payload["text"])
+        self.assertIn("process crash is not inferred", payload["textContent"])
+        self.assertEqual(str(requests[0].url), "https://api.brevo.com/v3/smtp/email")
+        self.assertEqual(requests[0].headers["api-key"], self.config["key"])
+        self.assertNotIn("Authorization", requests[0].headers)
+        self.assertNotIn("Idempotency-Key", requests[0].headers)
+        self.assertEqual(payload["sender"], {"email": self.config["sender"]})
+        self.assertEqual(payload["to"], [{"email": self.config["recipient"]}])
+        self.assertEqual(self.notifications.state["outbox"][0]["message_id"], "<synthetic@relay.example.org>")
         self.assertEqual(requests[0].extensions["timeout"], dict(connect=10, read=20, write=20, pool=20))
 
     def test_ambiguous_timeout_not_resent_after_restart(self):
@@ -133,12 +142,12 @@ class AlertChecks(unittest.TestCase):
         self.deliver(lambda request: self.fail("Ambiguous delivery must not resend"))
         self.assertEqual(self.notifications.state["outbox"][0]["delivery"], "unknown")
 
-    def test_known_transient_failure_bounded_and_idempotent(self):
+    def test_definite_rate_limit_failure_bounded(self):
         self.sample("RUNTIME_ERROR")
         keys = []
         def fail(request):
-            keys.append(request.headers["Idempotency-Key"])
-            return httpx.Response(503)
+            keys.append(request.headers["api-key"])
+            return httpx.Response(429)
         self.deliver(fail)
         self.deliver(fail)
         self.assertEqual(len(keys), 1)
@@ -149,7 +158,7 @@ class AlertChecks(unittest.TestCase):
         self.assertEqual(len(set(keys)), 1)
         self.assertEqual(self.notifications.state["outbox"][0]["delivery"], "failed")
 
-    def test_expired_retry_not_sent_with_expired_provider_key(self):
+    def test_expired_notification_not_sent(self):
         self.sample("RUNTIME_ERROR")
         self.now += 24 * 3600
         self.deliver(lambda request: self.fail("Expired retry must not send"))
@@ -179,6 +188,174 @@ class AlertChecks(unittest.TestCase):
             api.return_value.restart_space.assert_not_called()
             api.return_value.set_space_sleep_time.assert_not_called()
 
+    def mail_environment(self, **changes):
+        values = {"EVALUATOR_SPACE_ID": "fixture/worker", "EVALUATOR_ALERT_BREVO_API_KEY": self.config["key"],
+                  "EVALUATOR_ALERT_FROM": self.config["sender"], "EVALUATOR_ALERT_TO": self.config["recipient"],
+                  "EVALUATOR_ALERTS_ENABLED": "true", "EVALUATOR_ALERT_TEST_ENABLED": "true"}
+        values.update(changes)
+        return patch.dict(os.environ, values, clear=True)
+
+    def test_brevo_configuration_validation_and_no_resend_fallback(self):
+        with self.mail_environment():
+            self.assertEqual(alerts._config(), self.config)
+            for variable, value in (("EVALUATOR_ALERT_BREVO_API_KEY", ""), ("EVALUATOR_ALERT_BREVO_API_KEY", "bad\nkey"),
+                                    ("EVALUATOR_ALERT_FROM", "bad@example"), ("EVALUATOR_ALERT_TO", "a@example.org,b@example.org"),
+                                    ("EVALUATOR_ALERT_TO", "a@example.org\nBcc:other@example.org"), ("EVALUATOR_ALERTS_ENABLED", "false")):
+                with patch.dict(os.environ, {variable: value}):
+                    self.assertIsNone(alerts._config())
+        with self.mail_environment(EVALUATOR_ALERT_BREVO_API_KEY="", EVALUATOR_ALERT_RESEND_API_KEY="synthetic-old-key"):
+            self.assertIsNone(alerts._config())
+
+    def test_5xx_is_ambiguous_not_automatically_retried(self):
+        self.sample("RUNTIME_ERROR")
+        self.deliver(lambda request: httpx.Response(503, json={"secret": self.config["key"]}))
+        self.assertEqual(self.notifications.state["outbox"][0]["delivery"], "unknown")
+        self.notifications = alerts.Notifications(self.directory, "fixture/worker")
+        self.now += 300
+        self.deliver(lambda request: self.fail("Ambiguous 5xx must not resend"))
+        self.assertNotIn(self.config["key"], (self.directory / "notifications.json").read_text())
+
+    def test_201_without_trustworthy_message_id_is_not_resent(self):
+        self.sample("RUNTIME_ERROR")
+        self.deliver(lambda request: httpx.Response(201, json={"messageId": f"<{self.config['key']}@relay.example.org>"}))
+        item = self.notifications.state["outbox"][0]
+        self.assertEqual(item["delivery"], "sent")
+        self.assertIsNone(item["message_id"])
+        self.deliver(lambda request: self.fail("Accepted mail must not resend"))
+        self.assertNotIn(self.config["key"], (self.directory / "notifications.json").read_text())
+
+    def test_legacy_resend_processed_receipts_preserved_and_parked(self):
+        self.sample("RUNTIME_ERROR")
+        legacy = copy.deepcopy(self.notifications.state)
+        legacy["version"] = 1
+        template = legacy["outbox"][0]
+        template.pop("provider")
+        template.pop("message_id")
+        legacy["outbox"] = [dict(template, id=f"{index:032x}", delivery=delivery, tries=1)
+                            for index, delivery in enumerate(("sent", "unknown", "failed", "pending"), 1)]
+        path = self.directory / "notifications.json"
+        path.write_text(json.dumps(legacy))
+        before = path.read_bytes()
+        self.notifications = alerts.Notifications(self.directory, "fixture/worker")
+        self.deliver(lambda request: self.fail("Migration must not replay processed incidents"))
+        self.assertEqual(path.read_bytes(), before)
+        self.assertEqual(self.notifications.state["latched"], legacy["latched"])
+        self.assertEqual([item["delivery"] for item in self.notifications.state["outbox"]], [item["delivery"] for item in legacy["outbox"]])
+        self.assertTrue(all(item["provider"] == "resend" for item in self.notifications.state["outbox"]))
+        self.sample("RUNTIME_ERROR", attempts=1)
+        self.assertEqual(len(self.notifications.state["outbox"]), 4)
+
+    def test_legacy_unattempted_pending_receipt_can_use_brevo_once(self):
+        self.sample("RUNTIME_ERROR")
+        legacy = copy.deepcopy(self.notifications.state)
+        legacy["version"] = 1
+        legacy["outbox"][0].pop("provider")
+        legacy["outbox"][0].pop("message_id")
+        (self.directory / "notifications.json").write_text(json.dumps(legacy))
+        self.notifications = alerts.Notifications(self.directory, "fixture/worker")
+        self.deliver(lambda request: httpx.Response(201, json={"messageId": "<synthetic@relay.example.org>"}))
+        item = self.notifications.state["outbox"][0]
+        self.assertEqual(item["id"], legacy["outbox"][0]["id"])
+        self.assertEqual(item["provider"], "brevo")
+        self.deliver(lambda request: self.fail("No duplicate after migration"))
+
+    def test_operator_test_disabled_or_unconfigured_does_not_write(self):
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(EVALUATOR_ALERT_TEST_ENABLED="false"):
+            with patch.object(alerts, "_send_brevo") as transport:
+                self.assertEqual(alerts.send_test_email()["status"], "disabled")
+                transport.assert_not_called()
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(EVALUATOR_ALERT_BREVO_API_KEY=""):
+            self.assertEqual(alerts.send_test_email()["status"], "unconfigured")
+        self.assertEqual(list(self.directory.iterdir()), [])
+
+    def test_operator_test_fixed_recipient_cooldown_and_incident_isolation(self):
+        self.sample("RUNTIME_ERROR")
+        incident = (self.directory / "notifications.json").read_bytes()
+        budget = self.directory / "queue/evaluator-restart.json"
+        budget.parent.mkdir()
+        budget.write_text('{"attempts":6,"blocked":"exhausted"}')
+        before = budget.read_bytes()
+        requests = []
+        client = httpx.Client
+        def respond(request):
+            requests.append(request)
+            payload = json.loads(request.content)
+            self.assertEqual(payload["subject"], "[TEST] PhysInOne Leaderboard Email Alert")
+            self.assertEqual(payload["to"], [{"email": self.config["recipient"]}])
+            self.assertIn("No real GPU or Worker failure occurred", payload["textContent"])
+            return httpx.Response(201, json={"messageId": "<test@relay.example.org>"})
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(EVALUATOR_ALERTS_ENABLED="false"), \
+                patch.object(alerts.time, "time", side_effect=lambda: self.now), \
+                patch.object(alerts.httpx, "Client", side_effect=lambda **options: client(transport=httpx.MockTransport(respond), **options)):
+            self.assertEqual(alerts.send_test_email()["status"], "api_accepted")
+            self.assertEqual(alerts.send_test_email()["status"], "cooldown")
+            self.now += 60
+            self.assertEqual(alerts.send_test_email()["status"], "api_accepted")
+        self.assertEqual(len(requests), 2)
+        self.assertEqual(budget.read_bytes(), before)
+        self.assertEqual((self.directory / "notifications.json").read_bytes(), incident)
+        receipt = json.loads((self.directory / "auxiliary/lifecycle-alert-tests/test-send.json").read_bytes())
+        self.assertEqual(receipt["message_id"], "<test@relay.example.org>")
+        self.assertNotIn(self.config["key"], json.dumps(receipt))
+
+    def test_failed_operator_send_single_attempt_and_consumes_cooldown(self):
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(), \
+                patch.object(alerts.time, "time", return_value=self.now), \
+                patch.object(alerts, "_send_brevo", return_value=("retryable", None)) as transport:
+            self.assertEqual(alerts.send_test_email()["status"], "failed")
+            self.assertEqual(alerts.send_test_email()["status"], "cooldown")
+            transport.assert_called_once()
+
+    def test_test_persistence_failure_prevents_network_and_is_sanitized(self):
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(), \
+                patch.object(alerts, "_save_test_record", side_effect=OSError(self.config["key"])), \
+                patch.object(alerts, "_send_brevo") as transport:
+            with self.assertLogs(alerts.logger, level="WARNING") as logs:
+                result = alerts.send_test_email()
+            self.assertEqual(result["status"], "unavailable")
+            transport.assert_not_called()
+            self.assertNotIn(self.config["key"], json.dumps(result) + "".join(logs.output))
+
+    def test_no_public_web_user_entry_or_arbitrary_recipient_cli_option(self):
+        app = (Path(__file__).parent / "app.py").read_text(encoding="utf-8")
+        tree = ast.parse(app)
+        names = {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)}
+        attributes = {node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute)}
+        self.assertNotIn("send_test_email", names | attributes)
+        self.assertNotIn("send-test", app)
+        with patch("sys.argv", ["operator-command", "send-test", "--recipient", "ordinary@example.org"]), \
+                patch.object(alerts, "send_test_email") as send, patch("sys.stderr"):
+            with self.assertRaises(SystemExit) as result:
+                alerts.main()
+            self.assertEqual(result.exception.code, 2)
+            send.assert_not_called()
+
+    def test_operator_timeout_receipt_and_cooldown_survive_new_invocation(self):
+        client = httpx.Client
+        def timeout(request):
+            raise httpx.ReadTimeout(self.config["key"], request=request)
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(), \
+                patch.object(alerts.time, "time", return_value=self.now), \
+                patch.object(alerts.httpx, "Client", side_effect=lambda **options: client(transport=httpx.MockTransport(timeout), **options)):
+            with self.assertLogs(alerts.logger, level="WARNING") as logs:
+                result = alerts.send_test_email()
+            self.assertEqual(result["status"], "unknown")
+            self.assertEqual(alerts.send_test_email()["status"], "cooldown")
+        raw = (self.directory / "auxiliary/lifecycle-alert-tests/test-send.json").read_text()
+        self.assertEqual(json.loads(raw)["delivery"], "unknown")
+        self.assertNotIn(self.config["key"], raw + json.dumps(result) + "".join(logs.output))
+
+    def test_malformed_operator_receipt_fails_closed(self):
+        path = self.directory / "auxiliary/lifecycle-alert-tests/test-send.json"
+        path.parent.mkdir(parents=True)
+        path.write_text('{"unexpected":"synthetic-secret"}')
+        before = path.read_bytes()
+        with patch.object(alerts, "DATA_ROOT", str(self.directory)), self.mail_environment(), \
+                patch.object(alerts, "_send_brevo") as transport:
+            self.assertEqual(alerts.send_test_email()["status"], "unavailable")
+            transport.assert_not_called()
+        self.assertEqual(path.read_bytes(), before)
+
 
 if __name__ == "__main__":
     with tempfile.TemporaryDirectory(prefix="physinone-alerts-import-") as root:

src/submission/lifecycle_alerts.py

@@ -29,22 +29,62 @@ _CATEGORIES = {"restart_budget_exhausted", "starting_30_minutes", "unexpected_st
 _MAX_GAP = 45
 
 
-def _config() -> dict | None:
-    if os.environ.get("EVALUATOR_ALERTS_ENABLED", "").lower() != "true":
-        return None
+def _mail_config() -> dict | None:
     config = {name: os.environ.get(variable, "").strip() for name, variable in (
-        ("space", "EVALUATOR_SPACE_ID"), ("key", "EVALUATOR_ALERT_RESEND_API_KEY"),
+        ("space", "EVALUATOR_SPACE_ID"), ("key", "EVALUATOR_ALERT_BREVO_API_KEY"),
         ("sender", "EVALUATOR_ALERT_FROM"), ("recipient", "EVALUATOR_ALERT_TO"))}
     if not all(config.values()):
         return None
     if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", config["space"]):
         return None
+    if not re.fullmatch(r"[\x21-\x7e]{1,4096}", config["key"]):
+        return None
     for name in ("sender", "recipient"):
-        if not re.fullmatch(r"[^\s<>@,;]+@[^\s<>@,;]+", config[name]) or len(config[name]) > 254:
+        address = config[name]
+        if (not re.fullmatch(r"[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?)+", address)
+                or len(address) > 254 or len(address.split("@")[0]) > 64
+                or address.startswith(".") or ".@" in address or ".." in address):
             return None
     return config
 
 
+def _config() -> dict | None:
+    if os.environ.get("EVALUATOR_ALERTS_ENABLED", "").lower() != "true":
+        return None
+    return _mail_config()
+
+
+def _message_id(value, key="") -> str | None:
+    if (not isinstance(value, str) or not re.fullmatch(r"<[A-Za-z0-9._+:-]{1,200}@[A-Za-z0-9.-]{1,200}>", value)
+            or key and key in value or re.search(r"(?i)hf_|ghp_|github_pat_|xkeysib-|xsmtpsib-|bearer", value)):
+        return None
+    return value
+
+
+def _send_brevo(config: dict, subject: str, text: str) -> tuple[str, str | None]:
+    try:
+        with httpx.Client(timeout=httpx.Timeout(20, connect=10), follow_redirects=False) as client:
+            response = client.post("https://api.brevo.com/v3/smtp/email",
+                headers={"api-key": config["key"], "Accept": "application/json"},
+                json={"sender": {"email": config["sender"]}, "to": [{"email": config["recipient"]}],
+                      "subject": subject, "textContent": text})
+        if response.status_code == 201:
+            try:
+                body = response.json()
+                identifier = _message_id(body.get("messageId"), config["key"]) if isinstance(body, dict) else None
+            except ValueError:
+                identifier = None
+            return "sent", identifier
+        if response.status_code == 429:
+            return "retryable", None
+        if response.status_code >= 500 or 200 <= response.status_code < 300:
+            return "unknown", None
+        return "failed", None
+    except Exception as error:
+        logger.warning("Auxiliary email outcome unknown (%s); not automatically resending", type(error).__name__)
+        return "unknown", None
+
+
 def observe(space_id: str, stage: str | None, active: bool | None, state: dict | None) -> None:
     try:
         config = _config()
@@ -107,14 +147,17 @@ class Notifications:
                 raise ValueError("Notification state limit")
             state = json.loads(raw)
         except FileNotFoundError:
-            state = {"version": 1, "space": self.space, "latched": [], "outbox": []}
+            state = {"version": 2, "space": self.space, "latched": [], "outbox": []}
         if (not isinstance(state, dict) or set(state) != {"version", "space", "latched", "outbox"}
-                or state["version"] != 1 or state["space"] != self.space
+                or type(state["version"]) is not int or state["version"] not in {1, 2} or state["space"] != self.space
                 or not isinstance(state["latched"], list) or any(item not in _CATEGORIES for item in state["latched"])
                 or not isinstance(state["outbox"], list) or len(state["outbox"]) > 1000):
             raise ValueError("Invalid notification state")
         for item in state["outbox"]:
-            if (not isinstance(item, dict) or set(item) != {"id", "category", "stage", "timestamp", "attempts", "delivery", "tries", "next_try"}
+            fields = {"id", "category", "stage", "timestamp", "attempts", "delivery", "tries", "next_try"}
+            if state["version"] == 2:
+                fields |= {"provider", "message_id"}
+            if (not isinstance(item, dict) or set(item) != fields
                     or not isinstance(item["id"], str) or not re.fullmatch(r"[0-9a-f]{32}", item["id"])
                     or item["category"] not in _CATEGORIES or item["stage"] not in _STAGES | {"UNKNOWN"}
                     or not isinstance(item["timestamp"], str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", item["timestamp"])
@@ -123,6 +166,15 @@ class Notifications:
                     or type(item["tries"]) is not int or not 0 <= item["tries"] <= 3
                     or type(item["next_try"]) not in {int, float} or not 0 <= item["next_try"] < float("inf")):
                 raise ValueError("Invalid notification receipt")
+            if state["version"] == 2 and (item["provider"] not in {"brevo", "resend", "unassigned"}
+                    or item["message_id"] is not None and _message_id(item["message_id"]) != item["message_id"]):
+                raise ValueError("Invalid notification provider receipt")
+            if state["version"] == 2 and item["provider"] == "unassigned" and (item["tries"] != 0 or item["delivery"] != "pending" or item["message_id"] is not None):
+                raise ValueError("Invalid unattempted notification receipt")
+        if state["version"] == 1:
+            for item in state["outbox"]:
+                item.update(provider="resend" if item["tries"] or item["delivery"] != "pending" else "unassigned", message_id=None)
+            state["version"] = 2
         self.state, self.loaded = state, True
 
     def save(self):
@@ -179,7 +231,8 @@ class Notifications:
                     raise ValueError("Notification outbox full")
                 self.state["outbox"].append({"id": secrets.token_hex(16), "category": category,
                     "stage": stage or "UNKNOWN", "timestamp": datetime.fromtimestamp(observed, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
-                    "attempts": sample["attempts"], "delivery": "pending", "tries": 0, "next_try": now})
+                    "attempts": sample["attempts"], "delivery": "pending", "tries": 0, "next_try": now,
+                    "provider": "brevo", "message_id": None})
                 latches.add(category)
         self.state["latched"] = sorted(latches)
         if json.dumps(self.state, sort_keys=True) != before:
@@ -187,7 +240,8 @@ class Notifications:
 
     def deliver(self, config: dict, now: float):
         self.load()
-        item = next((item for item in self.state["outbox"] if item["delivery"] == "pending" and item["next_try"] <= now), None)
+        item = next((item for item in self.state["outbox"] if item["delivery"] == "pending"
+                     and item["provider"] in {"brevo", "unassigned"} and item["next_try"] <= now), None)
         if item is None:
             return
         created = datetime.fromisoformat(item["timestamp"].replace("Z", "+00:00")).timestamp()
@@ -197,26 +251,18 @@ class Notifications:
             return
         item["tries"] += 1
         item["delivery"] = "unknown"
+        item["provider"] = "brevo"
         self.save()
         diagnostic = {"restart_budget_exhausted": "Automatic restart budget exhausted; operator review required.",
                       "starting_30_minutes": "Startup continuously observed for 30 minutes; no automated action taken.",
                       "unexpected_stop_or_runtime_error": "Stop or runtime error observed with active work; process crash is not inferred."}
         text = (f"Worker Space: {self.space}\nCategory: {item['category']}\nUTC: {item['timestamp']}\n"
                 f"Runtime: {item['stage']}\nRestart attempts: {item['attempts']}\n{diagnostic[item['category']]}")
-        try:
-            with httpx.Client(timeout=httpx.Timeout(20, connect=10), follow_redirects=False) as client:
-                response = client.post("https://api.resend.com/emails",
-                    headers={"Authorization": f"Bearer {config['key']}", "Idempotency-Key": f"physinone-{item['id']}"},
-                    json={"from": config["sender"], "to": [config["recipient"]], "subject": f"PhysInOne: {item['category']}", "text": text})
-            if 200 <= response.status_code < 300:
-                item["delivery"] = "sent"
-            elif response.status_code == 429 or response.status_code >= 500:
-                item["delivery"] = "pending" if item["tries"] < 3 else "failed"
-                item["next_try"] = now + (60 if item["tries"] == 1 else 300)
-            else:
-                item["delivery"] = "failed"
-        except Exception as error:
-            logger.warning("Auxiliary email outcome unknown (%s); not automatically resending", type(error).__name__)
+        outcome, item["message_id"] = _send_brevo(config, f"PhysInOne: {item['category']}", text)
+        item["delivery"] = outcome
+        if outcome == "retryable":
+            item["delivery"] = "pending" if item["tries"] < 3 else "failed"
+            item["next_try"] = now + (60 if item["tries"] == 1 else 300)
         self.save()
 
 
@@ -266,3 +312,76 @@ def start():
 
 def stop():
     _stop.set()
+
+
+def _save_test_record(directory: Path, record: dict):
+    temporary = directory / f".test-send.{secrets.token_hex(8)}.tmp"
+    try:
+        with temporary.open("x", encoding="utf-8") as handle:
+            json.dump(record, handle, allow_nan=False)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, directory / "test-send.json")
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+def send_test_email() -> dict:
+    """Operator-shell entry only; deliberately not registered with HTTP/Gradio."""
+    if os.environ.get("EVALUATOR_ALERT_TEST_ENABLED", "").lower() != "true":
+        return {"status": "disabled", "message": "Operator test email is disabled."}
+    config = _mail_config()
+    if config is None:
+        return {"status": "unconfigured", "message": "Brevo email configuration is missing or invalid."}
+    directory = Path(DATA_ROOT) / "auxiliary" / "lifecycle-alert-tests"
+    try:
+        directory.mkdir(parents=True, exist_ok=True)
+        with _auxiliary_lock(directory / "test-send.lock"):
+            path = directory / "test-send.json"
+            try:
+                with path.open("rb") as handle:
+                    raw = handle.read(4097)
+                if len(raw) > 4096:
+                    raise ValueError("Invalid test receipt")
+                previous = json.loads(raw)
+                if (not isinstance(previous, dict) or set(previous) != {"version", "last_attempt", "delivery", "message_id"}
+                        or previous["version"] != 1 or type(previous["last_attempt"]) not in {int, float}
+                        or not 0 <= previous["last_attempt"] < float("inf")
+                        or previous["delivery"] not in {"sent", "unknown", "failed"}
+                        or previous["message_id"] is not None and _message_id(previous["message_id"]) != previous["message_id"]):
+                    raise ValueError("Invalid test receipt")
+            except FileNotFoundError:
+                previous = None
+            now = time.time()
+            if previous and now - previous["last_attempt"] < 60:
+                return {"status": "cooldown", "message": "Test email cooldown is active; wait at least 60 seconds between attempts."}
+            record = {"version": 1, "last_attempt": now, "delivery": "unknown", "message_id": None}
+            _save_test_record(directory, record)
+            text = ("This is a test email from the PhysInOne Leaderboard monitoring system. "
+                    "No real GPU or Worker failure occurred. API acceptance does not confirm inbox delivery.\n"
+                    f"UTC: {datetime.fromtimestamp(now, timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}")
+            outcome, record["message_id"] = _send_brevo(config, "[TEST] PhysInOne Leaderboard Email Alert", text)
+            record["delivery"] = "failed" if outcome == "retryable" else outcome
+            _save_test_record(directory, record)
+            if outcome == "sent":
+                return {"status": "api_accepted", "message": "Brevo API accepted the test email. Confirm delivery separately in Brevo or the configured recipient inbox."}
+            if outcome == "unknown":
+                return {"status": "unknown", "message": "Test email outcome is uncertain; check Brevo or the inbox before another manual attempt."}
+            return {"status": "failed", "message": "Brevo did not accept the test email; review the configured credentials, verified sender and provider limits."}
+    except Exception as error:
+        logger.warning("Operator email test unavailable (%s)", type(error).__name__)
+        return {"status": "unavailable", "message": "Operator email test is unavailable; no evaluation or incident state was changed."}
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Operator-only email test; use an authorized Frontend container terminal.")
+    parser.add_argument("command", choices=["send-test"])
+    parser.parse_args()
+    result = send_test_email()
+    print(json.dumps(result))
+    return 0 if result["status"] == "api_accepted" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())