Prefer the original telemetry exception over its recorded-cause copy

.github/scripts/telemetry_error_issues/records.py

@@ -35,6 +35,14 @@
 FINGERPRINT_ALGORITHM = 2
 FINGERPRINT_LENGTH = 16
 MIN_FINGERPRINT_KEY_LENGTH = 32
+# Exception classes that only copy another exception of the same failure. When
+# a simulation worker or coordinator fails, the simulation service logs the
+# original exception, then ``record_failure_cause`` records a redacted copy
+# that carries the original's traceback: a handled ``exception.recorded``
+# event. A worker's copy often has the original's frames and timestamp. The
+# copy sets no event name or attribute of its own, so only its class tells it
+# apart from other handled exceptions.
+RECORDED_CAUSE_COPY_TYPES = frozenset({"Stage12FailureCause"})
 
 
 @dataclass(frozen=True)
@@ -89,17 +97,22 @@ def kind(self) -> str:
 
 
 def _primary_preference(record: ErrorRecord) -> tuple[int, int, datetime]:
-    """Sort key: exceptions before messages before request completions.
+    """Sort key: exceptions, recorded-cause copies, messages, request completions.
 
+    A recorded-cause copy (see ``RECORDED_CAUSE_COPY_TYPES``) ranks after the
+    exception it copies, so the original's class decides the failure's group
+    whatever their timestamps; a failure with only the copy still uses it.
     Among exceptions, an unhandled one (the cause of an error response) comes
     first; otherwise the earliest record does.
     """
     if record.event_name == REQUEST_COMPLETED_EVENT:
-        source_rank = 2
+        source_rank = 3
+    elif record.error_type in RECORDED_CAUSE_COPY_TYPES:
+        source_rank = 1
     elif record.error_type:
         source_rank = 0
     else:
-        source_rank = 1
+        source_rank = 2
     handled_rank = 0 if record.error_handled is False else 1
     return source_rank, handled_rank, record.timestamp
 

changelog.d/telemetry-rank-cause-copies.fixed.md

@@ -0,0 +1 @@
+The daily telemetry error report counts a failed simulation worker or coordinator in its original exception's group. The simulation service also records each such failure as a redacted copy of class `Stage12FailureCause`, which carries the original's traceback and often its timestamp, and the report could choose either record, splitting one cause across two groups and two issues. The copy now ranks after the original exception, and is used only when the failure has no other exception.

gcp/observability/README.md

@@ -151,10 +151,15 @@ through `github_client`. Each run:
    `request.completed` records per route in the report day, for error rates.
    Each query is limited to 5 GiB billed.
 2. Chooses one record per failed request and service: the exception record,
-   else the error message, else the failed `request.completed` record. Records
-   that tie, such as two exceptions logged at the same moment, are ordered by
-   their own values, so the same records give the same report whatever order
-   the query returns them in.
+   else the error message, else the failed `request.completed` record. An
+   exception of class `Stage12FailureCause` ranks after any other exception
+   and before messages: the simulation service records it as a redacted copy
+   of a worker's or coordinator's original exception, carrying the original's
+   traceback and often its timestamp, so the original's class decides the
+   group, and a failure with only the copy still uses it.
+   Records that tie, such as two exceptions logged at the same moment, are
+   ordered by their own values, so the same records give the same report
+   whatever order the query returns them in.
 3. Groups failures by a 16-character HMAC-SHA256 fingerprint of the
    fingerprint algorithm number, the service family and either the exception
    type and innermost application stack frames, the route and message

tests/unit/telemetry_error_issues/report_support.py

@@ -175,6 +175,49 @@ def failures_on_the_unmatched_route(failure_count: int = 15) -> list[ErrorRecord
     ]
 
 
+STAGE12_WORKER_SERVICE = "policyengine-stage12-uk-worker"
+STAGE12_WORKER_FRAMES = (
+    '/root/policyengine_simulation_executor/runtime.py", line 120, in run_simulation',
+    '/root/policyengine_simulation_executor/datasets.py", line 80, in _load_dataset',
+    '/venv/lib/python3.13/site-packages/policyengine_uk/data.py", line 40, in get_year',
+)
+
+
+def original_worker_failure(**overrides: object) -> ErrorRecord:
+    """Return the original exception the simulation service logs for a failure.
+
+    It is a log record with an error, but no event name or handled flag.
+    """
+    fields: dict[str, object] = {
+        "service_name": STAGE12_WORKER_SERVICE,
+        "event_name": None,
+        "error_type": "ValueError",
+        "error_handled": None,
+        "frames": STAGE12_WORKER_FRAMES,
+        "route": None,
+        "status_code": None,
+        "operation_name": "stage12_simulation_execution",
+    }
+    fields.update(overrides)
+    return exception_record(**fields)
+
+
+def recorded_cause_copy(**overrides: object) -> ErrorRecord:
+    """Return the copy of a failure that the simulation service records next.
+
+    ``record_failure_cause`` records the failure again as a handled
+    ``exception.recorded`` event of class ``Stage12FailureCause``, with the
+    original's stack and often its timestamp to the microsecond.
+    """
+    fields: dict[str, object] = {
+        "event_name": "exception.recorded",
+        "error_type": "Stage12FailureCause",
+        "error_handled": True,
+    }
+    fields.update(overrides)
+    return original_worker_failure(**fields)
+
+
 PROJECT_ID = "policyengine-observability"
 # The deployment's values of the repository variables that name the logs.
 LOG_SOURCE = LogSource(

tests/unit/telemetry_error_issues/test_groups.py

@@ -21,6 +21,8 @@
     hours_after_window_noon,
     input_orders,
     message_record,
+    original_worker_failure,
+    recorded_cause_copy,
     request_count,
 )
 
@@ -524,3 +526,24 @@ def test_groups_are_the_same_in_any_input_order() -> None:
 
     assert groups_by_order["reversed"] == groups_by_order["forward"]
     assert groups_by_order["shuffled"] == groups_by_order["forward"]
+
+
+def test_groups_count_each_worker_failure_in_its_original_exception_group() -> None:
+    # As on 2026-10-26, the copy of some failures has the original's
+    # timestamp, and the copy of others is logged a moment later.
+    records = []
+    for index in range(4):
+        failed_at = hours_after_window_noon(index)
+        copy_logged_at = failed_at + timedelta(milliseconds=100 * (index % 2))
+        records += [
+            original_worker_failure(request_key=f"worker-{index}", timestamp=failed_at),
+            recorded_cause_copy(
+                request_key=f"worker-{index}", timestamp=copy_logged_at
+            ),
+        ]
+
+    groups = build_groups(records, [], reporting_window(REPORT_DATE), FINGERPRINT_KEY)
+
+    assert [(group.error_type, group.production.occurrences) for group in groups] == [
+        ("ValueError", 4)
+    ]

tests/unit/telemetry_error_issues/test_records.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 import re
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
+from itertools import permutations
 
 import pytest
 from telemetry_error_issues.records import (
+    ErrorRecord,
     fingerprint,
     fingerprint_key,
     fingerprint_parts,
@@ -21,6 +23,8 @@
     extract_frames,
     failed_request_record,
     message_record,
+    original_worker_failure,
+    recorded_cause_copy,
 )
 
 
@@ -169,6 +173,62 @@ def test_primary_keeps_failures_of_different_services_separate() -> None:
     assert primary_records([simulation_failure, api_failure]) == expected
 
 
+def _worker_failure_records() -> dict[str, ErrorRecord]:
+    """Return the records the simulation service logs for one worker failure.
+
+    The original exception and its recorded-cause copy share a timestamp, and
+    the worker's operation then fails with a RuntimeError of its own.
+    """
+    original = original_worker_failure()
+    return {
+        "original": original,
+        "copy": recorded_cause_copy(),
+        "operation": original_worker_failure(
+            event_name="operation.completed",
+            error_type="RuntimeError",
+            frames=original.frames[:1],
+            timestamp=original.timestamp + timedelta(seconds=1),
+        ),
+    }
+
+
+@pytest.mark.parametrize(
+    "record_order",
+    list(permutations(["original", "copy", "operation"])),
+    ids="-then-".join,
+)
+def test_primary_prefers_the_original_exception_over_its_recorded_cause_copy(
+    record_order: tuple[str, ...],
+) -> None:
+    records = _worker_failure_records()
+
+    primary = primary_records([records[name] for name in record_order])
+
+    assert [record.error_type for record in primary] == ["ValueError"]
+    assert primary == [records["original"]]
+
+
+@pytest.mark.parametrize(
+    "record_order",
+    list(permutations(["copy", "message", "failed_request"])),
+    ids="-then-".join,
+)
+def test_primary_uses_a_recorded_cause_copy_without_its_original(
+    record_order: tuple[str, ...],
+) -> None:
+    copy = recorded_cause_copy()
+    records = {
+        "copy": copy,
+        "message": message_record(service_name=copy.service_name),
+        "failed_request": failed_request_record(service_name=copy.service_name),
+    }
+
+    primary = primary_records([records[name] for name in record_order])
+
+    assert [record.error_type for record in primary] == ["Stage12FailureCause"]
+    assert primary == [copy]
+
+
 def test_fingerprint_exception_ignores_lines_path_prefixes_and_routes() -> None:
     moved_frames = tuple(
         frame.replace("/app/", "/srv/checkout/").replace("line 4", "line 9")