From e17d84b88a75ea60c6b5e501717fdbe6c2154dee Mon Sep 17 00:00:00 2001 From: Ahmed Ghazouani Date: Mon, 9 Nov 2026 19:46:56 +0100 Subject: [PATCH] Eval: ingest email items at their report date, not the wall clock (#346) The runner pushed each email item through ingest_email, which stamped it with datetime.now(UTC). That time became the alert's occurred_at, and triage's rule score adds 5 to an alert outside 06:00-18:00 UTC (agents/triage/base.py _score). The phishing_v1 cassettes were recorded at 08:23 UTC, so after 18:00 UTC phish-003/-004/-006/-008 built a triage prompt with rule confidence 100 instead of 95 and missed. (The other four were already capped at 100.) ingest_email takes an optional received_at; the runner passes the report's top-level Date header, or a fixed noon when there is none. Production still uses the arrival time. The committed cassettes replay unchanged. A new integration test replays them with the ingestion clock frozen at 03:30 and 21:00 UTC; it fails without the fix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JmdYVXkgpNHoygVJCrUeZc --- changelog.d/fix-phishing-cassette-clock.md | 10 ++++ src/lg_sotf/eval/runner.py | 31 ++++++++++++- src/lg_sotf/ingestion/service.py | 8 +++- tests/integration/eval/test_cassettes.py | 34 ++++++++++++++ tests/unit/eval/test_email_received_at.py | 54 ++++++++++++++++++++++ 5 files changed, 134 insertions(+), 3 deletions(-) create mode 100644 changelog.d/fix-phishing-cassette-clock.md create mode 100644 tests/unit/eval/test_email_received_at.py diff --git a/changelog.d/fix-phishing-cassette-clock.md b/changelog.d/fix-phishing-cassette-clock.md new file mode 100644 index 00000000..7ec150fa --- /dev/null +++ b/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. diff --git a/src/lg_sotf/eval/runner.py b/src/lg_sotf/eval/runner.py index 0ff70ce4..c1efee60 100644 --- a/src/lg_sotf/eval/runner.py +++ b/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)]) diff --git a/src/lg_sotf/ingestion/service.py b/src/lg_sotf/ingestion/service.py index 6d653e72..57af5dbc 100644 --- a/src/lg_sotf/ingestion/service.py +++ b/src/lg_sotf/ingestion/service.py @@ -282,6 +282,7 @@ 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``. @@ -289,11 +290,14 @@ async def ingest_email( 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:'") - received_at = datetime.now(UTC) + received_at = received_at or datetime.now(UTC) events: List[Dict[str, Any]] = [] for data in messages: try: diff --git a/tests/integration/eval/test_cassettes.py b/tests/integration/eval/test_cassettes.py index dbdf7133..f8ba1847 100644 --- a/tests/integration/eval/test_cassettes.py +++ b/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 == [] diff --git a/tests/unit/eval/test_email_received_at.py b/tests/unit/eval/test_email_received_at.py new file mode 100644 index 00000000..7dccb661 --- /dev/null +++ b/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 \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)]