Eval: ingest email items at their report date, not the wall clock (#346)

changelog.d/fix-phishing-cassette-clock.md

@@ -0,0 +1,10 @@
+### Fixed
+
+- An evaluation no longer depends on the hour it runs. An email item used to
+  be stamped with the time it was ingested, which became the alert's
+  `occurred_at`, and triage adds rule confidence to alerts outside
+  06:00-18:00 UTC. A run in the evening built different triage prompts than
+  a run in the morning, so the committed `phishing_v1` cassettes stopped
+  replaying after 18:00 UTC (#346). The evaluation runner now ingests each
+  report at its own `Date` header, or at a fixed noon when it has none.
+  Production still stamps a pushed report with the time it arrives.

src/lg_sotf/eval/runner.py

@@ -43,6 +43,9 @@
 from __future__ import annotations
 
 import asyncio
+import email.parser
+import email.policy
+import email.utils
 import hashlib
 import json
 import re
@@ -86,6 +89,28 @@
 MAX_TENANT_ID_CHARS = 63
 EMAIL_EVAL_SOURCE = email_source_id("eval")
 ORACLE_FAILED = "oracle_failed"
+EMAIL_FALLBACK_RECEIVED_AT = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
+"""When an email item whose report has no usable ``Date`` header was received (a fixed weekday noon)."""
+
+
+def email_received_at(mime: str) -> datetime:
+    """When an email item's report was received: its top-level ``Date`` header, never the wall clock.
+
+    Production stamps a pushed report with the time it arrived, and that time
+    becomes the alert's ``occurred_at``, which triage scores (an after-hours
+    alert gains rule confidence). An evaluation that did the same would build a
+    different triage prompt depending on the hour it ran, so a recorded
+    cassette would stop replaying. The report's own date is fixed by the
+    dataset; a report without one gets :data:`EMAIL_FALLBACK_RECEIVED_AT`.
+    """
+    header = email.parser.HeaderParser(policy=email.policy.compat32).parsestr(mime).get("Date")
+    try:
+        parsed = email.utils.parsedate_to_datetime(str(header)) if header else None
+    except (TypeError, ValueError, IndexError, OverflowError):
+        parsed = None
+    if parsed is None:
+        return EMAIL_FALLBACK_RECEIVED_AT
+    return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC)
 
 
 class EvalSetupError(RuntimeError):
@@ -291,7 +316,11 @@ async def _ingest(self, item: EvalItem) -> IngestResult:
         if item.source_type == "email":
             message = item.email
             return await self.ingestion.ingest_email(
-                self.tenant_id, source_id, [message.mime.encode("utf-8")], reporter=message.reporter
+                self.tenant_id,
+                source_id,
+                [message.mime.encode("utf-8")],
+                reporter=message.reporter,
+                received_at=email_received_at(message.mime),
             )
         return await self.ingestion.ingest(self.tenant_id, source_id, [dict(item.alert)])
 

src/lg_sotf/ingestion/service.py

@@ -282,18 +282,22 @@ async def ingest_email(
         messages: Sequence[bytes],
         *,
         reporter: Optional[str] = None,
+        received_at: Optional[datetime] = None,
     ) -> IngestResult:
         """Parse raw RFC 5322 ``messages`` (user reports) and store them for the reserved ``source_id``.
 
         Parsing runs in a worker thread under :attr:`email_limits`; a message
         that cannot be parsed (too large, too deeply nested, malformed) is
         rejected with the reason. ``reporter`` overrides the reporter read from
         a forwarded report's envelope. Deduplication is on the message's
-        Message-ID plus the reporter.
+        Message-ID plus the reporter. ``received_at`` (default: now) is when
+        the reports arrived and becomes each alert's ``occurred_at``; the
+        evaluation runner passes the report's own date so a run does not
+        depend on the hour it starts.
         """
         if not SOURCE_ID_PATTERN.match(source_id or "") or not is_email_source_id(source_id):
             raise ValueError("email pushes use the reserved source ids 'email' and 'email:<suffix>'")
-        received_at = datetime.now(UTC)
+        received_at = received_at or datetime.now(UTC)
         events: List[Dict[str, Any]] = []
         for data in messages:
             try:

tests/integration/eval/test_cassettes.py

@@ -16,6 +16,7 @@
 import json
 import os
 import shutil
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Dict, List, Tuple
 
@@ -29,6 +30,7 @@
 from lg_sotf.eval.dataset import EvalItem, load_dataset, parse_dataset
 from lg_sotf.eval.results import ItemResult
 from lg_sotf.eval.runner import EvalRunner, open_engine, run_evaluation
+from lg_sotf.ingestion import service as ingestion_service
 from lg_sotf.ingestion.service import IngestionService
 from lg_sotf.llm.cassette import CassetteSet, cassette_name, recording_model_factory
 from lg_sotf.llm.fake import ScriptedChatModel
@@ -183,3 +185,35 @@ async def test_the_committed_cassettes_replay_offline_like_a_live_run(
     assert guard.attempts == []
     assert outcomes(replay.results) == outcomes(live.results)
     assert replay.info["llm_provider"] == "cassette" and replay.info["cassettes"]["mode"] == "replay"
+
+
+@pytest.mark.parametrize("wall_clock", ["2026-11-09T03:30:00+00:00", "2026-11-10T21:00:00+00:00"])
+async def test_the_phishing_cassettes_replay_at_any_hour_of_the_wall_clock(
+    wall_clock: str, eval_settings: Settings, repos: Repositories, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Issue #346: the hour an evaluation runs never reaches a prompt.
+
+    The cassettes were recorded in the morning (UTC). A pushed report used to be
+    stamped with the wall clock, which became the alert's ``occurred_at``, and
+    triage adds rule confidence to an alert outside 06:00-18:00: a replay in the
+    evening or at night built another prompt and missed. Both clocks here are
+    after hours.
+    """
+    frozen = datetime.fromisoformat(wall_clock)
+
+    class FrozenClock(datetime):
+        @classmethod
+        def now(cls, tz: Any = None) -> datetime:  # type: ignore[override]
+            return frozen.astimezone(tz) if tz is not None else frozen.replace(tzinfo=None)
+
+    monkeypatch.setattr(ingestion_service, "datetime", FrozenClock)
+    assert ingestion_service.datetime.now(UTC) == frozen
+    name = "phishing_v1"
+    dataset = load_dataset(FIXTURES / f"{name}.jsonl").select(limit=COMMITTED[name])
+    guard = block_network_and_model(eval_settings, monkeypatch)
+    replay = await run_evaluation(
+        eval_settings, dataset, tenant_id=f"clock-{frozen:%d%H}",
+        cassettes=CassetteSet(FIXTURES / "cassettes" / name, "replay"),
+    )
+    assert [(r.item_id, r.error) for r in replay.results if r.status == "error"] == []
+    assert guard.attempts == []

tests/unit/eval/test_email_received_at.py

@@ -0,0 +1,54 @@
+"""An email item's report time comes from the message, never the wall clock (issue #346)."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any, List
+
+import pytest
+
+from lg_sotf.eval import runner
+from lg_sotf.eval.dataset import parse_dataset
+from lg_sotf.eval.runner import EMAIL_FALLBACK_RECEIVED_AT, email_received_at
+
+REPORT = (
+    "From: Reporter <alice@example.com>\nTo: phish-reports@example.com\nSubject: Reported: invoice\n"
+    "Date: Mon, 28 Sep 2026 09:30:00 +0200\n\nPlease check.\n"
+)
+
+
+def test_the_report_time_is_its_date_header_in_utc() -> None:
+    assert email_received_at(REPORT) == datetime(2026, 9, 28, 7, 30, tzinfo=UTC)
+
+
+@pytest.mark.parametrize(
+    "mime",
+    [
+        "This is a pasted fragment without headers.",
+        "From: a@example.com\nDate: not a date\n\nbody\n",
+        "From: a@example.com\nDate: \n\nbody\n",
+    ],
+)
+def test_a_report_without_a_usable_date_gets_the_fixed_fallback(mime: str) -> None:
+    assert email_received_at(mime) == EMAIL_FALLBACK_RECEIVED_AT
+
+
+def test_a_date_without_a_zone_is_read_as_utc() -> None:
+    assert email_received_at("Date: Mon, 28 Sep 2026 21:00:00 -0000\n\nx\n") == datetime(2026, 9, 28, 21, tzinfo=UTC)
+
+
+async def test_the_runner_ingests_an_email_item_at_its_report_time() -> None:
+    item = parse_dataset(
+        '{"id": "e-1", "source_type": "email", "alert": {"mime": ' + json.dumps(REPORT) + "},"
+        ' "expected": {"disposition": "malicious"}}'
+    ).items[0]
+    seen: List[Any] = []
+
+    class Ingestion:
+        async def ingest_email(self, *args: Any, received_at: Any = None, **kwargs: Any) -> None:
+            seen.append(received_at)
+
+    r = runner.EvalRunner(tenant_id="t", repos=None, ingestion=Ingestion(), engine=None)  # type: ignore[arg-type]
+    await r._ingest(item)
+    assert seen == [datetime(2026, 9, 28, 7, 30, tzinfo=UTC)]