Incident Drill: Debug a Failing ML Service from Logs and Metrics

"""Your incident analysis. Perch's listing-guard started holding almost every new listing at 14:00, and the
review queue is drowning. Work from the request log and the live service to find why, prove it, fix it, and
add the check that would have paged you sooner."""
import json

import pandas as pd

import serving

DEPLOY_TS = "2026-03-18T14:00:00"   # when the review queue started filling


def load_log(path="requests.jsonl"):
    """The production request log: one row per listing with its raw fields, the score and decision the
    service emitted, and (added later, from review outcomes) the ground-truth is_bad."""
    return pd.DataFrame(json.loads(line) for line in open(path))


def raw_records(logs):
    """The log rows as plain dicts, the way they arrived on the queue: only the fields each row actually
    carried (the log unions columns across rows, so absent fields read as NaN and are dropped here)."""
    return [{k: v for k, v in row.items() if pd.notna(v)} for row in logs.to_dict("records")]


# ---------- Step 1: triage from the log ----------

def timeline(logs, freq="15min"):
    """Bucket the log by time and report volume and hold-rate per bucket. Returns a DataFrame indexed by
    bucket start with columns n and hold_rate."""
    # TODO (Step 1): to_datetime the ts, group by pd.Grouper(freq=freq), and per bucket
    # return n = len and hold_rate = mean of decision == 'hold'; drop empty buckets.
    raise NotImplementedError("Step 1: write timeline()")


def change_point(tl):
    """The start time of the bucket where hold_rate jumps up most from the bucket before it."""
    # TODO (Step 1): the index where hold_rate rises most from the previous bucket:
    # diff the hold_rate values, take argmax, add one for the shift.
    raise NotImplementedError("Step 1: write change_point()")


# ---------- Step 2: reproduce it ----------

def reproduce(records):
    """Run the current service over these raw records and return the list of decisions. If the current
    code reproduces the log, the fault is in the code path you can see, not a fluke of production."""
    raise NotImplementedError("reproduce() arrives in Step 2")


# ---------- Step 3: localize the feature ----------

def feature_report(before, after):
    """Featurize the before and after records through the current service and, per model feature, report
    its median before, its median after, and the share of after rows that collapsed onto one constant
    value. A feature that used to vary and is now a single value is the break. Returns a DataFrame indexed
    by feature."""
    raise NotImplementedError("feature_report() arrives in Step 3")


def culprit(report):
    """The feature that broke: among features now pinned to one constant (constant_share > 0.9), the one
    whose median moved most. Return its name."""
    raise NotImplementedError("culprit() arrives in Step 3")


# ---------- Step 5: verify and guard ----------

def outcome(records):
    """Run the current service over records that carry ground-truth is_bad, and report hold_rate and
    false_hold_rate: the share of legit (is_bad == 0) listings that were held. The incident's harm is a
    flood of false holds burying the real ones."""
    raise NotImplementedError("outcome() arrives in Step 5")


def silent_default_guard(records, feature, limit=0.5):
    """A release check to run before a deploy: featurize the batch and return False if `feature` collapses
    onto one constant for more than `limit` of it, the fingerprint of an input field that stopped
    arriving. True means the batch looks safe."""
    raise NotImplementedError("silent_default_guard() arrives in Step 5")