Investigate Recurring Bugs and Prevent the Same Failure from Returning

Problem Statement

In telemetry and distributed control systems, recurring bugs often slip past hotfixes because the underlying state invariant was never formally captured or tested. When multiple production incidents are reported, teams must compare execution traces against system invariants to identify which fundamental rule broke, pinpoint the earliest step where state corrupted, and create regression tests.

Write a function identify_recurring_failures(incidents, invariants, k) that audits incident traces against a set of state invariants, identifies invariants that failed across at least k distinct incidents, and reports the recurring issues ordered by impact.

Invariant Definition

Each invariant is represented as a dictionary:

• id (str): Unique identifier of the invariant.
• field (str): The state variable to inspect.
• op (str): One of ">", ">=", "<", "<=", "==", "!=".
• value (int): Threshold value to compare against.

An invariant condition is satisfied at step t if state[field] op value evaluates to True. If field is missing from state, the invariant is violated at that step. An incident is considered to have violated an invariant if the condition is False at any step; only the first (earliest) step index of violation is recorded for that incident.

Function Signature

def identify_recurring_failures(
    incidents: list[dict],
    invariants: list[dict],
    k: int
) -> list[dict]:
    pass

Input

• incidents: A list of dictionaries, each with:
  • "id" (str): Incident identifier.
  • "states" (list[dict[str, int]]): Ordered sequence of system states from step 0 to S - 1.
• invariants: A list of invariant dictionaries (id, field, op, value).
• k: Integer recurrence threshold (k >= 1).

Output

Return a list of dictionaries for each invariant that failed in >= k incidents, sorted by:

1. Number of affected incidents in descending order (incident_count desc).
2. invariant_id in ascending alphabetical order.

Each dictionary in the output should have:

• "invariant_id" (str)
• "incident_count" (int)
• "earliest_violations" (dict[str, int]): Mapping of incident_id to the first step index where the invariant failed.

Constraints

• 1 <= len(incidents) <= 500
• 1 <= len(states) <= 1000 for each incident
• 1 <= len(invariants) <= 50
• All state values and thresholds are integers between -10^9 and 10^9

Example

incidents = [
    {
        "id": "INC-101",
        "states": [
            {"temp": 45, "voltage": 380},
            {"temp": 62, "voltage": 375},
            {"temp": 85, "voltage": 360}
        ]
    },
    {
        "id": "INC-102",
        "states": [
            {"temp": 40, "voltage": 390},
            {"temp": 78, "voltage": 370}
        ]
    }
]

invariants = [
    {"id": "INV_TEMP_MAX", "field": "temp", "op": "=", "value": 370}
]

# With k = 2:
# INV_TEMP_MAX violated in INC-101 at step 2 (temp=85), and in INC-102 at step 1 (temp=78).
# INV_VOLT_MIN violated in INC-101 at step 2 (voltage=360), but not in INC-102.
result = identify_recurring_failures(incidents, invariants, k=2)
# Returns:
# [
#     {
#         "invariant_id": "INV_TEMP_MAX",
#         "incident_count": 2,
#         "earliest_violations": {"INC-101": 2, "INC-102": 1}
#     }
# ]