SRE

---
name: cron-schedule-explainer
description: Explains, validates, and writes cron schedules - translates a cron expression into plain English, lists the next run times, and flags pitfalls such as the day-of-month OR day-of-week rule, dates that never occur, daylight saving gaps, time zone confusion, and overlapping or too-frequent jobs. Use when a user pastes a crontab line, Kubernetes CronJob, GitHub Actions schedule, or asks "when will this run?" or "write a cron for every second Tuesday".
---

# Cron Schedule Explainer

You make scheduled jobs predictable. For every schedule you give a plain-English meaning, concrete next run times, and the risks that would surprise someone at 2 a.m.

## Files in this skill

- `scripts/cron_explain.py` - parser, explainer, next-run calculator, and pitfall checker (Python 3 standard library only)
- `references/cron-syntax.md` - field ranges, special characters, macros, and platform differences
- `references/scheduling-pitfalls.md` - common mistakes and how to avoid them
- `templates/schedule-review.md` - review format
- `examples/example-backup-review.md` - a worked review of three crontab lines

## Workflow

### 1. Identify the platform
Standard 5-field cron (Vixie cron, cronie, Kubernetes CronJob, GitHub Actions) is the default. Ask or check if the user means Quartz (6 or 7 fields with seconds and `?`), AWS EventBridge (6 fields with year), or systemd timers; see `references/cron-syntax.md`. Note the time zone: GitHub Actions always uses UTC; Kubernetes uses the controller's time zone unless `timeZone` is set.

### 2. Run the checker
```bash
python3 scripts/cron_explain.py "30 2 * * 1-5"
python3 scripts/cron_explain.py "0 9 1 * MON" --count 8 --from "2026-10-08 10:00"
python3 scripts/cron_explain.py --file crontab.txt
```
It prints a plain-English explanation, the next N run times (naive local time of the server), and warnings. Exit code is 1 when an expression is invalid or never runs.

If you cannot run the script, apply the same rules by hand and say so.

### 3. Explain
For each schedule give:
1. One-sentence plain-English meaning.
2. The next 3 to 5 runs with the time zone stated.
3. Warnings from the script and from `references/scheduling-pitfalls.md` that apply (overlap with long jobs, DST, UTC versus local, missed runs while the machine is off).

### 4. Write or fix schedules
When the user describes a schedule in words, write the expression, then run it through the checker to confirm the next runs match their intent. For things cron cannot express directly (every second Tuesday, the last weekday of the month), give a cron expression plus a guard in the command, for example `[ "$(date +\%d)" -le 07 ] && run-job`, and explain why.

### 5. Report
Use `templates/schedule-review.md`, as in `examples/example-backup-review.md`.

## Rules
- Always state the time zone you are assuming.
- Remember that `%` must be escaped as `\%` inside crontab command fields.
- Never edit a live crontab for the user; show the line to add and the `crontab -e` step.
- Recommend a lock (for example `flock -n /tmp/job.lock cmd`) whenever a job could run longer than its interval.
FILE:references/cron-syntax.md
# Cron Syntax Reference (5-field standard)

```
+------------- minute (0-59)
| +----------- hour (0-23)
| | +--------- day of month (1-31)
| | | +------- month (1-12 or JAN-DEC)
| | | | +----- day of week (0-7 or SUN-SAT; 0 and 7 are both Sunday)
| | | | |
* * * * *  command
```

## Special characters
| Symbol | Meaning | Example |
|---|---|---|
| `*` | every value | `* * * * *` every minute |
| `,` | list | `0 8,12,18 * * *` at 08:00, 12:00, 18:00 |
| `-` | range | `0 9 * * 1-5` 09:00 Monday to Friday |
| `/` | step | `*/15 * * * *` every 15 minutes; `10-50/20` = 10, 30, 50 |

Names are case-insensitive. Ranges of names (`MON-FRI`) work in most implementations, lists of names work everywhere.

## Macros
| Macro | Equivalent |
|---|---|
| `@yearly` / `@annually` | `0 0 1 1 *` |
| `@monthly` | `0 0 1 * *` |
| `@weekly` | `0 0 * * 0` |
| `@daily` / `@midnight` | `0 0 * * *` |
| `@hourly` | `0 * * * *` |
| `@reboot` | once at startup (not time based) |

## The day rule
If both day of month and day of week are restricted (neither is `*`), the job runs when EITHER matches. `0 9 1 * MON` runs on the 1st of every month AND every Monday.

## Platform differences
| Platform | Fields | Time zone | Notes |
|---|---|---|---|
| Linux cron (cronie, Vixie) | 5 | system local time | `CRON_TZ=` supported by cronie |
| Kubernetes CronJob | 5 | controller time zone, or `spec.timeZone` | use `concurrencyPolicy: Forbid` to prevent overlaps |
| GitHub Actions `schedule` | 5 | always UTC | runs can be delayed under load; minimum interval 5 minutes |
| Quartz (Java) | 6-7 (seconds first, optional year) | configurable | `?` for "no specific value", `L`, `W`, `#` supported |
| AWS EventBridge | 6 (with year) | UTC unless a scheduler time zone is set | either day-of-month or day-of-week must be `?` |

The script in this skill supports the 5-field standard plus the macros above (except `@reboot`, which it reports as not time based).
FILE:references/scheduling-pitfalls.md
# Scheduling Pitfalls

## 1. Day of month OR day of week
`0 0 13 * 5` is NOT "Friday the 13th". It runs on every 13th and every Friday. Use `0 0 13 * *` plus a guard: `[ "$(date +\%u)" = 5 ] && cmd`.

## 2. Dates that never or rarely occur
- `0 0 30 2 *` never runs (February has no 30th).
- `0 0 31 * *` runs only in 7 months of the year.
- `0 0 29 2 *` runs only in leap years.
For "last day of the month" use `0 0 28-31 * *` with a guard: `[ "$(date -d tomorrow +\%d)" = 01 ] && cmd`.

## 3. Daylight saving time
In local time zones with DST, times between about 01:00 and 03:00 can be skipped (spring forward) or run twice (fall back), depending on the cron implementation. Schedule critical jobs outside that window, or run cron in UTC.

## 4. UTC versus local time
GitHub Actions and many cloud schedulers use UTC. "Every day at 09:00" for a team in Istanbul (UTC+3) is `0 6 * * *` in UTC. Always write the time zone next to the expression in docs and code comments.

## 5. Too frequent or overlapping runs
- `* * * * *` runs 1440 times a day. Make sure that is intended.
- A minute field of `*` with a fixed hour (`* 3 * * *`) runs 60 times between 03:00 and 03:59; usually `0 3 * * *` was meant.
- If a job can take longer than its interval, use a lock (`flock -n`) or `concurrencyPolicy: Forbid`.

## 6. Step values do not wrap evenly
`*/7` in the minute field runs at 0, 7, ..., 56, then again at 0 (a 4-minute gap). `*/25` runs at 0, 25, 50. Steps restart every hour, day, or month.

## 7. Thundering herd
Many teams pick `0 0 * * *` or `0 * * * *`. Shift jobs to an odd minute (for example `17 2 * * *`) to avoid load spikes on shared systems and rate-limited APIs.

## 8. Environment and output
Cron runs with a minimal PATH and no login shell. Use absolute paths, set needed variables in the crontab, and redirect output (`>> /var/log/job.log 2>&1`) so failures are visible. Escape `%` as `\%`.

## 9. Missed runs
Plain cron does not catch up on runs missed while the machine was off. Use anacron, systemd timers with `Persistent=true`, or Kubernetes `startingDeadlineSeconds` when a missed run matters.
FILE:templates/schedule-review.md
# Schedule Review: {{system_or_repo}}

**Platform:** {{Linux cron | Kubernetes CronJob | GitHub Actions | other}}
**Time zone assumed:** {{time_zone}}
**Reviewed on:** {{date}}

## Summary
{{One or two sentences: are the schedules doing what the team expects, and what must change.}}

## Schedules

### {{n}}. `{{expression}}` - {{job name}}
- **Meaning:** {{plain-English explanation}}
- **Next runs:** {{run 1}}, {{run 2}}, {{run 3}}
- **Verdict:** {{OK | FIX | CLARIFY}}
- **Warnings:**
  - {{warning}}
- **Suggested line:**
  ```
  {{corrected crontab line}}
  ```

## Questions
- {{question for the team}}
FILE:examples/example-backup-review.md
# Schedule Review: ops server crontab

**Platform:** Linux cron (cronie)
**Time zone assumed:** Europe/Berlin (server local time, has DST)
**Reviewed on:** 2026-10-08

## Summary
Two of the three lines do not do what the comments say. The backup runs inside the DST window, and the "Friday the 13th" report actually runs every Friday and every 13th.

## Schedules

### 1. `30 2 * * *` - nightly database backup
- **Meaning:** At 02:30 every day.
- **Next runs:** 2026-10-09 02:30, 2026-10-10 02:30, 2026-10-11 02:30
- **Verdict:** FIX
- **Warnings:**
  - 02:30 is inside the DST change window; on the spring-forward night it may be skipped and in autumn it may run twice.
  - The backup can take over an hour on month-end; no lock.
- **Suggested line:**
  ```
  17 4 * * * flock -n /tmp/db-backup.lock /opt/scripts/db-backup.sh >> /var/log/db-backup.log 2>&1
  ```

### 2. `0 9 13 * FRI` - "Friday the 13th" fun report
- **Meaning:** At 09:00 on day 13 of the month OR on every Friday (cron's day rule).
- **Next runs:** 2026-10-09 09:00 (Fri), 2026-10-13 09:00 (Tue, the 13th), 2026-10-16 09:00 (Fri)
- **Verdict:** FIX
- **Warnings:**
  - Both day fields are restricted, so cron uses OR, not AND.
- **Suggested line:**
  ```
  0 9 13 * * [ "$(date +\%u)" = 5 ] && /opt/scripts/fun-report.sh
  ```

### 3. `*/20 8-18 * * 1-5` - sync tickets from the help desk
- **Meaning:** Every 20 minutes (at :00, :20, :40) from 08:00 to 18:59, Monday to Friday.
- **Next runs:** 2026-10-08 10:20, 2026-10-08 10:40, 2026-10-08 11:00
- **Verdict:** CLARIFY
- **Warnings:**
  - Last run of the day is 18:40, not 18:00. Use `8-17` plus a separate `0 18 * * 1-5` if the sync should stop at 18:00.

## Questions
- Should the server run cron in UTC to avoid DST issues entirely?
FILE:scripts/cron_explain.py
#!/usr/bin/env python3
"""Explain, validate, and preview standard 5-field cron expressions (stdlib only).

Usage:
  python3 cron_explain.py "EXPR" [--count N] [--from "YYYY-MM-DD HH:MM"]
  python3 cron_explain.py --file crontab.txt [--count N] [--from ...]

For each expression: a plain-English explanation, the next N run times
(naive server-local time), and pitfall warnings. In --file mode, crontab
lines are read; comments, blank lines and VAR=value lines are skipped and the
first five fields (or a leading @macro) are taken as the schedule.
Exit code: 0 = all valid, 1 = an expression is invalid or never runs, 2 = usage.
"""
import argparse
import calendar
import datetime as dt
import sys

MONTHS = {m.lower(): i for i, m in enumerate(calendar.month_abbr) if m}
DAYS = {"sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6}
MACROS = {
    "@yearly": "0 0 1 1 *", "@annually": "0 0 1 1 *", "@monthly": "0 0 1 * *",
    "@weekly": "0 0 * * 0", "@daily": "0 0 * * *", "@midnight": "0 0 * * *",
    "@hourly": "0 * * * *",
}
FIELDS = [("minute", 0, 59, {}), ("hour", 0, 23, {}), ("day of month", 1, 31, {}),
          ("month", 1, 12, MONTHS), ("day of week", 0, 7, DAYS)]
DAY_NAMES = ["Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"]


def compress(values, fmt=str):
    """[1,2,3,5] -> '1-3, 5' using fmt for each number."""
    vals, out, i = sorted(values), [], 0
    while i < len(vals):
        j = i
        while j + 1 < len(vals) and vals[j + 1] == vals[j] + 1:
            j += 1
        out.append(fmt(vals[i]) if j - i < 2 else f"{fmt(vals[i])}-{fmt(vals[j])}")
        if 0 < j - i < 2:
            out.append(fmt(vals[j]))
        i = j + 1
    return ", ".join(out)


class CronError(ValueError):
    pass


def _num(token, lo, hi, names, field):
    t = token.lower()
    if t in names:
        return names[t]
    if not t.isdigit():
        raise CronError(f"{field}: '{token}' is not a number or known name")
    v = int(t)
    if not lo <= v <= hi:
        raise CronError(f"{field}: {v} is outside {lo}-{hi}")
    return v


def parse_field(text, lo, hi, names, field):
    values = set()
    for part in text.split(","):
        if not part:
            raise CronError(f"{field}: empty list item in '{text}'")
        step = 1
        if "/" in part:
            part, step_s = part.split("/", 1)
            if not step_s.isdigit() or int(step_s) == 0:
                raise CronError(f"{field}: bad step '/{step_s}'")
            step = int(step_s)
        if part == "*":
            start, end = lo, hi
        elif "-" in part:
            a, b = part.split("-", 1)
            start, end = _num(a, lo, hi, names, field), _num(b, lo, hi, names, field)
            if start > end:
                raise CronError(f"{field}: range {a}-{b} is reversed")
        else:
            start = _num(part, lo, hi, names, field)
            end = hi if step > 1 else start
        values.update(range(start, end + 1, step))
    return values


def parse(expr):
    expr = expr.strip()
    if expr.lower() == "@reboot":
        raise CronError("@reboot runs once at startup and is not time based")
    expr = MACROS.get(expr.lower(), expr)
    parts = expr.split()
    if len(parts) != 5:
        hint = " (6-7 fields look like Quartz or EventBridge; see references/cron-syntax.md)" if len(parts) in (6, 7) else ""
        raise CronError(f"expected 5 fields, got {len(parts)}{hint}")
    for p in parts:
        bare = p.lower()
        for n in list(MONTHS) + list(DAYS):
            bare = bare.replace(n, "")
        if any(c in bare for c in "?lw#"):
            raise CronError(f"'{p}': '?', 'L', 'W' and '#' are Quartz extensions, not standard cron")
    sets = [parse_field(p, lo, hi, names, name) for p, (name, lo, hi, names) in zip(parts, FIELDS)]
    if 7 in sets[4]:
        sets[4].discard(7)
        sets[4].add(0)
    return parts, sets


def describe_set(values, lo, hi, field, raw):
    vals = sorted(values)
    if raw == "*":
        return None
    if field == "day of week":
        names = [DAY_NAMES[v] for v in vals]
        if vals == [1, 2, 3, 4, 5]:
            return "Monday to Friday"
        if vals == [0, 6]:
            return "on weekends"
        return ", ".join(names)
    if field == "month":
        return ", ".join(calendar.month_name[v] for v in vals)
    return compress(vals)


def explain(parts, sets):
    minute, hour, dom, month, dow = sets
    rm, rh, rdom, rmon, rdow = parts
    if rm == "*" and rh == "*":
        time_txt = "every minute"
    elif rm.startswith("*/") and rh == "*":
        time_txt = f"every {rm[2:]} minutes"
    elif rm.startswith("*/"):
        time_txt = (f"every {rm[2:]} minutes (at minute " + ", ".join(str(m) for m in sorted(minute)) +
                    ") during hour(s) " + compress(hour, lambda h: f"{h:02d}"))
    elif rh == "*":
        time_txt = "at minute " + ", ".join(str(m) for m in sorted(minute)) + " of every hour"
    elif rm == "*":
        time_txt = "every minute during hour(s) " + compress(hour, lambda h: f"{h:02d}")
    elif len(minute) * len(hour) <= 6:
        time_txt = "at " + ", ".join(f"{h:02d}:{m:02d}" for h in sorted(hour) for m in sorted(minute))
    else:
        time_txt = ("at minute(s) " + ", ".join(str(m) for m in sorted(minute)) +
                    " past hour(s) " + compress(hour, lambda h: f"{h:02d}"))
    day_txt = []
    d_dom = describe_set(dom, 1, 31, "day of month", rdom)
    d_dow = describe_set(dow, 0, 6, "day of week", rdow)
    if d_dom and d_dow:
        day_txt.append(f"on day(s) {d_dom} of the month OR on {d_dow}")
    elif d_dom:
        day_txt.append(f"on day(s) {d_dom} of the month")
    elif d_dow:
        day_txt.append(d_dow if d_dow.startswith("on ") else f"on {d_dow}")
    else:
        day_txt.append("every day")
    d_mon = describe_set(month, 1, 12, "month", rmon)
    if d_mon:
        day_txt.append(f"in {d_mon}")
    text = f"{time_txt}, {' '.join(day_txt)}"
    return text[0].upper() + text[1:] + "."


def day_matches(d, parts, sets):
    _, _, dom, month, dow = sets
    if d.month not in month:
        return False
    cron_dow = (d.weekday() + 1) % 7
    dom_r, dow_r = parts[2] != "*", parts[4] != "*"
    if dom_r and dow_r:
        return d.day in dom or cron_dow in dow
    if dom_r:
        return d.day in dom
    if dow_r:
        return cron_dow in dow
    return True


def next_runs(parts, sets, start, count, max_days=366 * 8):
    minute, hour = sorted(sets[0]), sorted(sets[1])
    runs = []
    day = start.date()
    for _ in range(max_days):
        if day_matches(day, parts, sets):
            for h in hour:
                for m in minute:
                    t = dt.datetime(day.year, day.month, day.day, h, m)
                    if t > start:
                        runs.append(t)
                        if len(runs) >= count:
                            return runs
        day += dt.timedelta(days=1)
    return runs


def warnings(parts, sets, runs):
    minute, hour, dom, month, dow = sets
    out = []
    if parts[2] != "*" and parts[4] != "*":
        out.append("Day of month AND day of week are both set: cron runs when EITHER matches (OR, not AND).")
    if parts[2] != "*" and parts[4] == "*":
        max_days = {m: (29 if m == 2 else calendar.monthrange(2026, m)[1]) for m in month}
        if not any(d <= max_days[m] for m in month for d in dom):
            out.append("Never runs: the chosen day(s) of month do not exist in the chosen month(s).")
        elif any(d > 28 for d in dom):
            out.append("Some chosen days (29-31) do not exist in every month, so some months are skipped.")
    if parts[0] == "*" and parts[1] != "*":
        out.append("Minute is '*': runs every minute of the chosen hour(s); did you mean minute 0?")
    runs_per_day = len(minute) * len(hour)
    if runs_per_day >= 288:
        out.append(f"Runs {runs_per_day} times a day; make sure that is intended and add a lock against overlap.")
    if any(1 <= h <= 2 for h in hour) and parts[1] != "*":
        out.append("Runs between 01:00 and 02:59: in local time zones with DST this can be skipped or run twice.")
    for i, raw in ((0, parts[0]), (1, parts[1])):
        if "/" in raw:
            step = int(raw.split("/")[1])
            span = 60 if i == 0 else 24
            if span % step:
                out.append(f"Step /{step} does not divide {span}: the gap is uneven where the {'hour' if i == 0 else 'day'} wraps.")
    if parts[0] == "0" and parts[1] in ("*", "0"):
        out.append("Minute 0 at the top of the hour is a popular slot; consider an odd minute to avoid load spikes.")
    return out


def check(expr, start, count):
    print(f"Expression: {expr}")
    try:
        parts, sets = parse(expr)
    except CronError as e:
        print(f"  INVALID: {e}\n")
        return False
    print(f"  Meaning: {explain(parts, sets)}")
    runs = next_runs(parts, sets, start, count)
    ok = True
    if runs:
        print(f"  Next {len(runs)} run(s) after {start:%Y-%m-%d %H:%M} (server local time):")
        for r in runs:
            print(f"    {r:%Y-%m-%d %H:%M} {r:%a}")
    else:
        print("  Next runs: none found in the next 8 years")
        ok = False
    for w in warnings(parts, sets, runs):
        print(f"  WARNING: {w}")
    print()
    return ok


def crontab_schedules(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            s = line.strip()
            if not s or s.startswith("#"):
                continue
            first = s.split()[0]
            if "=" in first and not first.startswith("@"):
                continue
            yield first if first.startswith("@") else " ".join(s.split()[:5])


def main(argv=None):
    ap = argparse.ArgumentParser(description="Explain and validate cron expressions.")
    ap.add_argument("expr", nargs="?", help='cron expression in quotes, e.g. "*/15 9-17 * * 1-5"')
    ap.add_argument("--file", help="read schedules from a crontab file")
    ap.add_argument("--count", type=int, default=5, help="number of next runs to show (default 5)")
    ap.add_argument("--from", dest="start", help='start time "YYYY-MM-DD HH:MM" (default: now)')
    a = ap.parse_args(argv)
    if bool(a.expr) == bool(a.file):
        ap.print_usage(sys.stderr)
        print("error: give exactly one of EXPR or --file", file=sys.stderr)
        return 2
    try:
        start = dt.datetime.strptime(a.start, "%Y-%m-%d %H:%M") if a.start else dt.datetime.now().replace(second=0, microsecond=0)
    except ValueError:
        print("error: --from must look like 2026-10-08 10:00", file=sys.stderr)
        return 2
    exprs = list(crontab_schedules(a.file)) if a.file else [a.expr]
    results = [check(e, start, max(1, a.count)) for e in exprs]
    print(f"{sum(results)} of {len(results)} schedule(s) valid and runnable.")
    return 0 if all(results) else 1


if __name__ == "__main__":
    sys.exit(main())

role: >
  You are a senior site reliability engineer who has written and maintained
  on-call runbooks for web services, APIs, data pipelines, and background
  workers. You write runbooks for a tired engineer paged at 3 a.m.: short,
  ordered steps, exact places to look, safe mitigations first, and clear
  rules for when to escalate. You never invent dashboard names, commands, or
  contacts that the user did not give you; you leave clearly marked
  placeholders instead.

task: >
  Build an on-call runbook scaffold for the service described below:
  severity levels, detection signals and alerts, a triage flow, mitigation
  playbooks for each common incident, a rollback procedure, communication
  templates, an escalation path, and a post-incident checklist, ready to be
  pasted into the team's docs and filled in.

inputs:
  service_name: "checkout-api"
  what_it_does: "handles cart checkout and payment authorization for the web shop and mobile app"
  architecture: "stateless Node.js service on Kubernetes, 6 pods, PostgreSQL primary with one replica, Redis cache, calls an external payment provider"
  traffic_and_slo: "about 40 requests per second at peak, 99.9 percent availability and p95 latency under 400 ms"
  common_incidents: "payment provider timeouts, database connection pool exhaustion, bad deploys, Redis eviction storms, certificate expiry"
  dashboards_and_tools: "Grafana dashboard 'Checkout Overview', Loki logs, PagerDuty, Argo CD for deploys, status page on Statuspage"
  deploy_and_rollback: "Argo CD sync from main, roughly 10 deploys a week, feature flags in LaunchDarkly"
  escalation_contacts: "primary on-call (payments team), secondary on-call, payments team lead, database on-call from platform team, payment provider support line"
  comms_channels: "#inc-checkout Slack channel, customer support lead, public status page"
  business_hours_and_timezone: "Europe/Berlin, 24x7 on-call, weekly rotation"
  known_gaps: "no alert on certificate expiry yet, replica lag is not graphed"

instructions:
  - State assumptions where inputs are missing instead of asking questions.
  - Define 4 severity levels (SEV1 to SEV4) with customer impact, examples for this service, response time, who is paged, and whether the status page is updated.
  - For detection, list each alert or signal with a metric or log query placeholder, a threshold idea tied to the SLO, the likely causes, and the first dashboard panel to open.
  - Write a generic triage flow of at most 10 ordered steps for the first 15 minutes, starting with acknowledging the page and confirming impact.
  - Write one mitigation playbook per common incident with symptoms, quick checks, safe mitigations ordered from least to most risky, when to escalate, and how to confirm recovery.
  - Write a rollback procedure for a bad deploy, including how to find the last good version, how to roll back, how to disable a feature flag, and what to verify after.
  - Provide short communication templates for the incident channel kickoff, a 30-minute update, the status page (investigating, identified, monitoring, resolved), and a note to customer support.
  - Give an escalation path with conditions and time limits, using the roles I gave.
  - Add a post-incident checklist (timeline, impact numbers, follow-up tickets, alert tuning, runbook updates) and a blameless review agenda.
  - Turn known gaps into follow-up tasks with an owner role.
  - Use <PLACEHOLDER> markers for every link, query, command, or name that the team must fill in; never make them up.

output_format: YAML only, no prose outside the YAML, using exactly this structure

output_schema:
  assumptions: [string]
  service:
    name: string
    purpose: string
    dependencies: [string]
    slo: string
    owners: string
  severity_levels:
    - level: SEV1 | SEV2 | SEV3 | SEV4
      customer_impact: string
      examples: [string]
      response_time: string
      page: [string]
      status_page_update: boolean
  detection_signals:
    - name: string
      signal: string
      query_placeholder: string
      threshold_idea: string
      likely_causes: [string]
      first_panel_to_open: string
  triage_steps:
    - step: number
      action: string
      look_at: string
      decision: string
  playbooks:
    - incident: string
      symptoms: [string]
      quick_checks: [string]
      mitigations:
        - order: number
          action: string
          risk: low | medium | high
          command_placeholder: string
      escalate_when: string
      confirm_recovery: [string]
  rollback:
    find_last_good_version: [string]
    steps: [string]
    feature_flag_kill_switch: [string]
    verify_after: [string]
  comms_templates:
    incident_channel_kickoff: string
    update_every_30_min: string
    status_page:
      investigating: string
      identified: string
      monitoring: string
      resolved: string
    customer_support_note: string
  escalation_path:
    - role: string
      when: string
      after_minutes: number
      how_to_reach: string
  post_incident_checklist: [string]
  blameless_review_agenda: [string]
  follow_up_tasks:
    - task: string
      owner_role: string
      priority: high | medium | low

---
name: time-series-gap-anomaly-checker
description: Checks CSV time series (metrics, sensor readings, sales by hour) for data problems before you chart or alert on them: missing intervals, duplicate or out-of-order timestamps, empty values, flatlines from stuck exporters, robust median/MAD spikes, and counter resets. Use when a dashboard looks wrong, before training a forecast, or when asking "is this data complete?". Includes a stdlib Python checker with --self-test.
---

# Time Series Gap and Anomaly Checker

You help analysts, SREs, and data engineers decide whether a time series can be trusted. You find where data is missing, repeated, stuck, or suspicious, explain the likely cause of each problem, and separate collection problems (the pipeline) from real events (the system being measured). You never silently fill gaps or delete outliers; you propose a treatment and say what it changes.

## Files in this skill

- `scripts/ts_check.py` - reads a CSV, infers the sampling interval, and reports gaps, duplicates, out-of-order rows, missing values, flatlines, spikes, and counter resets as markdown or JSON (Python 3 standard library only)
- `references/anomaly-methods.md` - how the checks work, how to tune thresholds, and when each method fails
- `references/gap-handling-playbook.md` - what to do with each kind of problem: fill, interpolate, drop, or annotate
- `templates/data-quality-report.md` - a report for a dashboard fix, incident review, or dataset hand-off
- `examples/api-metrics.csv` - three hours of 5-minute API metrics with a scrape outage, a duplicate, a stuck value, a latency spike, and a counter reset

## Workflow

### 1. Understand the series
Ask what each column measures, its unit, the expected sampling interval, and whether any column is a monotonic counter (requests_total, bytes_sent). Ask which time zone naive timestamps use; the script treats them as UTC.

### 2. Run the checker
```
python3 scripts/ts_check.py examples/api-metrics.csv --counter requests_total
python3 scripts/ts_check.py data.csv --time-col ts --cols temp_c,humidity --interval 10m
python3 scripts/ts_check.py data.csv --z 5 --window 12 --flatline-points 12 --json > report.json
python3 scripts/ts_check.py --self-test
```
Timestamps can be ISO 8601 or Unix epoch seconds or milliseconds. Exit code 1 when any issue is found, 2 on bad input, so it can run as a pipeline gate.

### 3. Read the results
- **Completeness** = rows present / rows expected over the range. Below 95% usually means aggregates and averages are biased.
- **gap**: spacing over 1.5x the interval. Long gaps at regular times often mean deploys or cron jobs; one long gap usually means an outage of the collector.
- **duplicate / out_of_order**: double scrapes, retries, or merged files.
- **flatline**: the same value repeated; real for on/off signals, suspicious for noisy measurements.
- **spike**: robust z-score against a rolling median. Check whether neighbouring columns spike at the same time (a real event) or not (a glitch).
- **reset**: a counter that went down, usually a process restart; rate calculations must handle it.

### 4. Recommend treatment
Use `references/gap-handling-playbook.md`. Tune thresholds with `references/anomaly-methods.md` if the data is seasonal, bursty, or very sparse.

### 5. Report
Fill `templates/data-quality-report.md` with the summary numbers, the top issues with times in the user's time zone, the suspected cause, and the agreed treatment.

## Rules

- Report problems; do not alter the source data. Write cleaned data to a new file and document every change.
- Do not call a spike an incident without corroboration (other metrics, logs, or a human).
- State the interval and thresholds you used so results are reproducible.
- Convert UTC times to the user's time zone when presenting them.
FILE:references/anomaly-methods.md
# Anomaly Methods and Tuning

## Interval inference
The expected interval is the median spacing between sorted unique timestamps. This survives a few gaps and jitter. Pass `--interval` when the series is mostly gaps (sparse events) or when you know the scrape interval.

A gap is any spacing above 1.5x the interval, so 5-minute data with a 7-minute spacing is not a gap but a 10-minute spacing is (1 missing point).

## Robust spikes (rolling median + MAD)
For each point, take up to `--window` neighbours on each side (excluding the point), compute their median and the median absolute deviation (MAD). The robust z-score is:

    z = |value - median| / (1.4826 * MAD)

1.4826 makes MAD comparable to a standard deviation for normal data. Default threshold 4 is roughly a 1-in-15,000 event for normal noise. When more than half the neighbours are identical (MAD = 0), the mean absolute deviation is used, and for a perfectly flat neighbourhood 1% of the level.

Why not mean and standard deviation? One large spike inflates the standard deviation and hides itself. The median and MAD ignore it.

## Tuning
| Data | Suggestion |
|---|---|
| Noisy metrics (latency, CPU) | `--z 5`, `--window 12` |
| Daily or weekly seasonality | Check one season at a time, or compare to the same hour last week before trusting spikes |
| Bursty business data (orders) | Higher `--z` (6-8); spikes are often real |
| Integer or low-resolution values | Expect MAD = 0; raise `--z` or aggregate first |
| On/off or status signals | Ignore flatline; check only gaps |
| Counters | Always pass `--counter`; check rates after handling resets |

## Known limits
- Gradual drifts and level shifts are not spikes; plot the series or compare week-over-week medians.
- The first and last few points have fewer neighbours, so their scores are less reliable.
- Irregular event streams (logins, purchases) have no fixed interval; aggregate into buckets first.
FILE:references/gap-handling-playbook.md
# Gap and Anomaly Handling Playbook

| Problem | Likely causes | Default treatment | Avoid |
|---|---|---|---|
| Short gap (1-2 points) | scrape timeout, network blip | Linear interpolation for smooth signals; leave null for counts | Filling counts with zero |
| Long gap | collector down, deploy, host off | Leave null and annotate the chart; exclude from SLO math or count as unknown | Interpolating across hours |
| Duplicate timestamps | double scrape, retried batch, merged files | Keep one row (first, or the one with most fields); check whether values differ | Summing duplicates |
| Out-of-order rows | late-arriving data, file concatenation | Sort by time before any windowed calculation | Assuming file order is time order |
| Missing values | sensor dropout, parse errors, "N/A" strings | Fix the parser first; then treat as a gap | Treating blank as 0 |
| Flatline | stuck exporter, cached value, sensor freeze | Confirm with a second source; mark as suspect | Trusting it for alerts |
| Spike (single point) | glitch, unit error, real burst | Check related columns and logs; cap or remove only with a documented reason | Deleting without a note |
| Counter reset | process restart, overflow | Compute rates per segment (rate after reset = new value / interval) | Negative rates |

## Before modeling or forecasting
1. Sort, de-duplicate, and regularize to the expected interval (insert nulls for missing slots).
2. Decide gap policy per column and record it.
3. Keep a mask column (`is_imputed`) so models and readers know what was filled.
4. Re-run the checker on the cleaned file and attach both reports.

## When to escalate
- Completeness under 90% for a metric used in an SLO or invoice.
- Gaps that repeat at the same time every day (a job is killing the collector).
- Spikes that appear in all hosts at once (a real incident, not a data issue).
FILE:templates/data-quality-report.md
# Time Series Data Quality - <dataset or dashboard>

**Source:** <file / query>  **Range:** <start> to <end> (<time zone>)  **Interval:** <expected>
**Rows:** <n>  **Completeness:** <percent>  **Thresholds:** z=<z>, window=<w>, flatline=<n>

## Summary
<two sentences: can this data be trusted for the intended use, and what must change first>

## Issues
| Kind | When (<time zone>) | Column | Detail | Suspected cause | Treatment |
|---|---|---|---|---|---|
| | | | | | |

## Decisions
- Gap policy: <interpolate short gaps / leave null / exclude>
- Outlier policy: <keep / cap / remove with note>
- Counter handling: <segment rates at resets>

## Follow-ups
- [ ] <fix collector / parser / alert> - owner <name> - due <date>

Generated with scripts/ts_check.py. Source data was not modified.
FILE:examples/api-metrics.csv
timestamp,p95_latency_ms,error_rate,requests_total
2026-10-18T22:00:00Z,182,0.4,1200
2026-10-18T22:05:00Z,176,0.3,2435
2026-10-18T22:10:00Z,190,0.5,3705
2026-10-18T22:15:00Z,185,0.4,5010
2026-10-18T22:20:00Z,179,0.3,6210
2026-10-18T22:25:00Z,188,0.4,7445
2026-10-18T22:30:00Z,181,0.5,8715
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:40:00Z,177,0.3,11220
2026-10-18T22:45:00Z,184,0.4,12455
2026-10-18T22:50:00Z,186,0.4,13725
2026-10-18T22:55:00Z,180,0.4,15030
2026-10-18T23:00:00Z,191,0.4,16230
2026-10-18T23:05:00Z,,0.4,17465
2026-10-18T23:10:00Z,178,0.4,18735
2026-10-18T23:15:00Z,187,0.4,20040
2026-10-18T23:20:00Z,182,0.4,21240
2026-10-18T23:25:00Z,189,0.4,22475
2026-10-18T23:30:00Z,612,2.1,23745
2026-10-18T23:35:00Z,185,0.5,25050
2026-10-18T23:40:00Z,180,0.3,26250
2026-10-18T23:45:00Z,184,0.4,27485
2026-10-19T00:10:00Z,188,0.5,28755
2026-10-19T00:15:00Z,183,0.4,30060
2026-10-19T00:20:00Z,177,0.3,31260
2026-10-19T00:25:00Z,185,0.4,32495
2026-10-19T00:30:00Z,182,0.5,1180
2026-10-19T00:35:00Z,189,0.4,2485
2026-10-19T00:40:00Z,184,0.3,3685
2026-10-19T00:45:00Z,180,0.4,4920
2026-10-19T00:50:00Z,186,0.5,6190
2026-10-19T00:55:00Z,183,0.4,7495
FILE:scripts/ts_check.py
#!/usr/bin/env python3
"""Time series gap and anomaly checker for CSV metrics (Python 3 standard library only).

Reads a CSV with a timestamp column and one or more numeric columns, infers the sampling
interval (median spacing) unless --interval is given, and reports:
  gap          missing samples (spacing > 1.5 x interval), with how many points are missing
  duplicate    repeated timestamps
  out_of_order timestamps that go backwards
  missing      empty, NaN, or non-numeric values
  flatline     the same value repeated for >= --flatline-points samples (stuck sensor/exporter)
  spike        robust outliers: |value - rolling median| / (1.4826 * MAD) >= --z (default 4);
               if MAD is 0 the mean absolute deviation is used instead
  reset        a counter column that decreases (--counter col1,col2)

Timestamps: ISO 8601 (with Z or offset; naive = UTC) or Unix epoch seconds/milliseconds.

Usage:
  python3 ts_check.py metrics.csv
  python3 ts_check.py metrics.csv --time-col ts --cols cpu,latency_ms --interval 5m --z 5
  python3 ts_check.py metrics.csv --counter requests_total --json
  python3 ts_check.py --self-test
Exit code: 0 clean, 1 issues found, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import re
import statistics
import sys


class InputError(ValueError):
    pass


def parse_ts(raw):
    s = (raw or "").strip()
    if not s:
        raise InputError("empty timestamp")
    if re.fullmatch(r"-?\d+(\.\d+)?", s):
        v = float(s)
        return v / 1000.0 if v > 1e11 else v
    try:
        t = dt.datetime.fromisoformat(s.replace("Z", "+00:00").replace(" ", "T", 1))
    except ValueError:
        raise InputError(f"unparseable timestamp {s!r}")
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.timestamp()


def parse_interval(s):
    m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)?", s.strip())
    if not m:
        raise InputError(f"bad interval {s!r}; use e.g. 30s, 5m, 1h")
    mult = {"ms": 0.001, "s": 1, None: 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]
    return float(m.group(1)) * mult


def fmt_ts(ts):
    return dt.datetime.fromtimestamp(ts, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def fmt_dur(sec):
    for unit, size in (("d", 86400), ("h", 3600), ("m", 60)):
        if sec >= size:
            v = sec / size
            return f"{v:.0f}{unit}" if v == int(v) else f"{v:.1f}{unit}"
    return f"{sec:g}s"


def to_float(raw):
    s = (raw or "").strip()
    if s == "" or s.lower() in ("nan", "null", "none", "na", "n/a", "-"):
        return None
    try:
        v = float(s)
    except ValueError:
        return None
    return None if math.isnan(v) or math.isinf(v) else v


def load(text, time_col=None, cols=None):
    reader = csv.DictReader(io.StringIO(text))
    if not reader.fieldnames:
        raise InputError("CSV has no header row")
    fields = [f.strip() for f in reader.fieldnames]
    reader.fieldnames = fields
    if time_col is None:
        cands = [f for f in fields if f.lower() in ("timestamp", "time", "ts", "date", "datetime")]
        time_col = cands[0] if cands else fields[0]
    if time_col not in fields:
        raise InputError(f"time column {time_col!r} not in header {fields}")
    value_cols = cols or [f for f in fields if f != time_col]
    for c in value_cols:
        if c not in fields:
            raise InputError(f"column {c!r} not in header {fields}")
    rows = []
    for i, r in enumerate(reader, start=2):
        rows.append({"line": i, "ts": parse_ts(r.get(time_col)),
                     "raw": {c: r.get(c) for c in value_cols}})
    if len(rows) < 3:
        raise InputError("need at least 3 data rows")
    return time_col, value_cols, rows


def rolling_robust(values, idx, window):
    lo, hi = max(0, idx - window), min(len(values), idx + window + 1)
    neigh = [values[j] for j in range(lo, hi) if j != idx and values[j] is not None]
    if len(neigh) < 4:
        return None, None
    med = statistics.median(neigh)
    mad = statistics.median(abs(v - med) for v in neigh)
    scale = 1.4826 * mad
    if scale == 0:  # more than half the neighbours equal the median: fall back to mean absolute deviation
        scale = 1.2533 * statistics.fmean(abs(v - med) for v in neigh)
    if scale == 0:  # perfectly flat neighbourhood: 1% of the level
        scale = max(abs(med) * 0.01, 1e-9)
    return med, scale


def check(text, time_col=None, cols=None, interval=None, z=4.0, window=6, flatline_points=6, counters=()):
    time_col, value_cols, rows = load(text, time_col, cols)
    issues = []

    def add(kind, line, ts, col, detail):
        issues.append({"kind": kind, "line": line, "time": fmt_ts(ts), "column": col, "detail": detail})

    # order and duplicates (in file order)
    seen = {}
    for prev, cur in zip(rows, rows[1:]):
        if cur["ts"] < prev["ts"]:
            add("out_of_order", cur["line"], cur["ts"], time_col, f"goes back {fmt_dur(prev['ts'] - cur['ts'])} from line {prev['line']}")
    for r in rows:
        if r["ts"] in seen:
            add("duplicate", r["line"], r["ts"], time_col, f"same timestamp as line {seen[r['ts']]}")
        else:
            seen[r["ts"]] = r["line"]
    srt = sorted({r["ts"]: r for r in reversed(rows)}.values(), key=lambda r: r["ts"])  # first occurrence wins
    diffs = [b["ts"] - a["ts"] for a, b in zip(srt, srt[1:]) if b["ts"] > a["ts"]]
    if interval is None:
        if not diffs:
            raise InputError("cannot infer interval: all timestamps identical")
        interval = statistics.median(diffs)
    expected = int(round((srt[-1]["ts"] - srt[0]["ts"]) / interval)) + 1
    missing_points = 0
    for a, b in zip(srt, srt[1:]):
        d = b["ts"] - a["ts"]
        if d > 1.5 * interval:
            n = int(round(d / interval)) - 1
            missing_points += n
            add("gap", b["line"], a["ts"], time_col, f"{fmt_dur(d)} without data after {fmt_ts(a['ts'])} (~{n} missing)")

    stats = {}
    for c in value_cols:
        vals = [to_float(r["raw"][c]) for r in srt]
        for r, v in zip(srt, vals):
            if v is None:
                add("missing", r["line"], r["ts"], c, f"value {r['raw'][c]!r} is empty or not numeric")
        # flatline runs
        run_start = 0
        for i in range(1, len(vals) + 1):
            if i < len(vals) and vals[i] is not None and vals[i] == vals[run_start]:
                continue
            length = i - run_start
            if vals[run_start] is not None and length >= flatline_points:
                r0, r1 = srt[run_start], srt[i - 1]
                add("flatline", r0["line"], r0["ts"], c, f"value {vals[run_start]:g} repeated {length}x until {fmt_ts(r1['ts'])}")
            run_start = i
        if c in counters:
            for i in range(1, len(vals)):
                if vals[i] is not None and vals[i - 1] is not None and vals[i] < vals[i - 1]:
                    add("reset", srt[i]["line"], srt[i]["ts"], c, f"counter dropped {vals[i - 1]:g} -> {vals[i]:g} (restart or wrap)")
        else:
            for i, v in enumerate(vals):
                if v is None:
                    continue
                med, scale = rolling_robust(vals, i, window)
                if med is None:
                    continue
                score = abs(v - med) / scale
                if score >= z:
                    add("spike", srt[i]["line"], srt[i]["ts"], c, f"{v:g} vs local median {med:g} (robust z {score:.1f})")
        good = [v for v in vals if v is not None]
        stats[c] = {"count": len(good), "min": min(good) if good else None, "max": max(good) if good else None,
                    "median": statistics.median(good) if good else None}

    order = {"out_of_order": 0, "duplicate": 1, "gap": 2, "reset": 3, "missing": 4, "flatline": 5, "spike": 6}
    issues.sort(key=lambda x: (order[x["kind"]], x["time"], x["column"]))
    counts = {}
    for it in issues:
        counts[it["kind"]] = counts.get(it["kind"], 0) + 1
    return {
        "rows": len(rows), "time_column": time_col, "columns": value_cols,
        "start": fmt_ts(srt[0]["ts"]), "end": fmt_ts(srt[-1]["ts"]),
        "interval": fmt_dur(interval), "interval_seconds": interval,
        "expected_points": expected, "missing_points": missing_points,
        "completeness": round(100.0 * len(srt) / expected, 1) if expected else 100.0,
        "stats": stats, "counts": counts, "issues": issues,
    }


def to_markdown(r):
    lines = ["# Time series check", "",
             f"Rows: {r['rows']}  Range: {r['start']} to {r['end']}  Interval: {r['interval']}",
             f"Completeness: {r['completeness']}% ({r['missing_points']} of {r['expected_points']} expected points missing)", "",
             "| Column | Count | Min | Median | Max |", "|---|---|---|---|---|"]
    for c, s in r["stats"].items():
        f = lambda v: "-" if v is None else f"{v:g}"
        lines.append(f"| {c} | {s['count']} | {f(s['min'])} | {f(s['median'])} | {f(s['max'])} |")
    lines += ["", "## Issues"]
    if not r["issues"]:
        lines.append("No issues found.")
    else:
        lines.append("Totals: " + ", ".join(f"{k} {v}" for k, v in r["counts"].items()))
        lines += ["", "| Kind | Time (UTC) | Column | Line | Detail |", "|---|---|---|---|---|"]
        lines += [f"| {i['kind']} | {i['time']} | {i['column']} | {i['line']} | {i['detail']} |" for i in r["issues"]]
    return "\n".join(lines)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    base = 1760860800  # 2025-10-19T08:00:00Z
    def csv_of(points, header="timestamp,value"):
        return header + "\n" + "\n".join(f"{t},{v}" for t, v in points) + "\n"
    wave = [10 + (i % 3) for i in range(30)]
    clean = csv_of([(base + 60 * i, v) for i, v in enumerate(wave)])
    r = check(clean)
    ok("clean series has no issues", r["issues"] == [])
    ok("interval inferred as 1m", r["interval"] == "1m")
    ok("completeness 100%", r["completeness"] == 100.0)
    gap_pts = [(base + 60 * i, v) for i, v in enumerate(wave) if not 10 <= i < 15]
    r = check(csv_of(gap_pts))
    ok("gap detected", r["counts"].get("gap") == 1)
    ok("gap counts 5 missing points", r["missing_points"] == 5)
    spiky = list(wave); spiky[20] = 95
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(spiky)]))
    ok("spike detected", [i["line"] for i in r["issues"] if i["kind"] == "spike"] == [22])
    flat = list(wave); flat[5:13] = [7] * 8
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(flat)]), z=100)
    ok("flatline detected", r["counts"].get("flatline") == 1)
    miss = [(base + 60 * i, ("" if i == 4 else "NaN" if i == 9 else v)) for i, v in enumerate(wave)]
    ok("missing values detected", check(csv_of(miss)).get("counts").get("missing") == 2)
    dup = [(base + 60 * i, v) for i, v in enumerate(wave)]
    dup.insert(6, dup[5])
    ok("duplicate detected", check(csv_of(dup))["counts"].get("duplicate") == 1)
    ooo = [(base + 60 * i, v) for i, v in enumerate(wave)]
    ooo[7], ooo[8] = ooo[8], ooo[7]
    ok("out of order detected", check(csv_of(ooo))["counts"].get("out_of_order") == 1)
    counter = [(base + 60 * i, 100 * i if i < 20 else 100 * (i - 20)) for i in range(30)]
    r = check(csv_of(counter, "ts,requests_total"), counters=("requests_total",))
    ok("counter reset detected", r["counts"].get("reset") == 1 and "spike" not in r["counts"])
    iso_rows = "time,cpu\n" + "\n".join(f"2026-10-19T08:{i:02d}:00Z,{40 + (i % 2)}" for i in range(0, 60, 5)) + "\n"
    r = check(iso_rows)
    ok("ISO timestamps and 5m interval", r["interval"] == "5m" and r["issues"] == [])
    ms_rows = csv_of([((base + 30 * i) * 1000, v) for i, v in enumerate(wave)])
    ok("epoch milliseconds accepted", check(ms_rows)["interval"] == "30s")
    r = check(clean, interval=30.0)
    ok("explicit 30s interval flags every 1m step as a gap", r["counts"].get("gap") == 29 and r["completeness"] < 60)
    jitter = csv_of([(base + 60 * i + (3 if i % 2 else -2), v) for i, v in enumerate(wave)])
    ok("small timestamp jitter is not a gap", "gap" not in check(jitter)["counts"])
    multi = "timestamp,a,b\n" + "\n".join(f"{base + 60 * i},{wave[i]},{wave[i] * 2 if i != 12 else 500}" for i in range(30)) + "\n"
    r = check(multi, cols=["a", "b"])
    ok("multi-column spike only in b", {i["column"] for i in r["issues"]} == {"b"})
    for bad, label in (("timestamp,value\n1,2\n", "too few rows rejected"),
                       ("timestamp,value\nyesterday,1\ntoday,2\nnow,3\n", "bad timestamp rejected")):
        try:
            check(bad)
            ok(label, False)
        except InputError:
            ok(label, True)
    ok("markdown renders", "## Issues" in to_markdown(check(clean)))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Find gaps, duplicates, flatlines, spikes, and counter resets in CSV metrics.")
    ap.add_argument("csv", nargs="?", help="CSV file, or '-' for stdin")
    ap.add_argument("--time-col")
    ap.add_argument("--cols", help="comma-separated value columns (default: all except the time column)")
    ap.add_argument("--interval", help="expected spacing, e.g. 30s, 5m, 1h (default: median spacing)")
    ap.add_argument("--z", type=float, default=4.0, help="robust z-score threshold for spikes (default 4)")
    ap.add_argument("--window", type=int, default=6, help="neighbours on each side for the rolling median (default 6)")
    ap.add_argument("--flatline-points", type=int, default=6)
    ap.add_argument("--counter", default="", help="comma-separated monotonic counter columns (checked for resets, not spikes)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.csv:
        ap.error("csv path required (or --self-test)")
    try:
        text = sys.stdin.read() if a.csv == "-" else open(a.csv, encoding="utf-8-sig").read()
        r = check(text, a.time_col, [c.strip() for c in a.cols.split(",")] if a.cols else None,
                  parse_interval(a.interval) if a.interval else None, a.z, a.window, a.flatline_points,
                  tuple(c.strip() for c in a.counter.split(",") if c.strip()))
    except (OSError, InputError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    return 1 if r["issues"] else 0


if __name__ == "__main__":
    sys.exit(main())

---
name: slo-error-budget-burn-rate-calculator
description: Turns request counts into SLO answers: SLI, error budget for a 28d or 30d window, budget consumed and remaining, burn rates over 5m to 3d, multiwindow burn-rate alerts (page at 14.4x and 6x, ticket at 3x and 1x, scaled to the window), and time until the budget runs out. Use when setting an SLO, tuning alerts, deciding on a release freeze, or writing an incident summary. Includes a stdlib Python calculator with --self-test.
---

# SLO Error Budget and Burn Rate Calculator

You help engineers and on-call teams reason about reliability with numbers instead of gut feeling. You turn good/total request counts into an SLI, translate the SLO into an error budget, show how fast the budget is burning in several windows, say which alerts would page or open a ticket, and estimate when the budget runs out at the current pace. You explain every number in plain language so product owners can follow the decision.

## Files in this skill

- `scripts/slo_budget.py` - reads a CSV of request counts per interval and prints SLI, budget consumed and remaining, burn rates, alert states, exhaustion times and the worst interval; `--allowance` prints allowed downtime and alert thresholds for any SLO (Python 3 standard library only)
- `references/slo-math.md` - formulas for SLI, error budget, burn rate, projected consumption, and worked numbers for common targets
- `references/burn-rate-alerting.md` - the multiwindow, multi-burn-rate alert policy, how thresholds scale with the window, and tuning advice
- `templates/slo-status-report.md` - a weekly SLO status or incident budget summary
- `examples/checkout-api-requests.csv` - two days of 15-minute request counts for a checkout API with a small blip and a 2-hour incident at the end

## Workflow

### 1. Define the SLI and SLO
Ask what a "good" event is (status < 500, latency under 300 ms, successful job run), where the counts come from, the target (99%, 99.5%, 99.9%, 99.95%), and the window (rolling 28 or 30 days). Push back on targets the system has never met: show the historical SLI first.

### 2. Get the data
Export good and total (or bad and total) counts per interval, for example 1-minute, 5-minute or 15-minute buckets, as CSV with `timestamp,total,good`. The timestamp marks the start of each interval. Use UTC.

### 3. Run the calculator
```
python3 scripts/slo_budget.py examples/checkout-api-requests.csv --slo 99.9 --window 30d
python3 scripts/slo_budget.py requests.csv --slo 99.5 --window 28d --json > slo.json
python3 scripts/slo_budget.py requests.csv --now 2026-10-19T22:00:00Z
python3 scripts/slo_budget.py --allowance --slo 99.95 --window 30d
python3 scripts/slo_budget.py --self-test
```
Exit code 1 means a page alert is firing or the budget is exhausted; 2 means bad input (missing columns, bad > total, unparseable timestamps).

### 4. Interpret
- Burn rate 1.0 spends exactly the budget over the window; 14.4 for an hour spends 2% of a 30-day budget.
- Budget consumed is projected against the traffic seen so far; with less than a full window of data, say so.
- An alert fires only when both its long and short window exceed the threshold; the short window makes it stop quickly after recovery. See `references/burn-rate-alerting.md`.
- Windows shorter than the data step are rounded up to one step; windows longer than the data are "not evaluable".

### 5. Recommend and report
Tie the numbers to a decision: page now, open a ticket, slow down risky releases while remaining budget is low, or loosen an SLO that is always violated by design. Fill `templates/slo-status-report.md`.

## Example result
For `examples/checkout-api-requests.csv` at 99.9% over 30d: SLI 99.8768%, 8.2% of the budget consumed, 1h burn 26.46 and 5m burn 28.06 so the fast page fires, 6h burn 5.83 just under 6.0, the 3d ticket is not evaluable with two days of data, and at the last hour's pace the budget is gone in about 1d 16h. Exit code 1.

## Rules

- State the SLO, window, data range and any missing intervals with every result.
- Do not mix good-event definitions in one SLI; compute separate SLOs for availability and latency.
- An SLO is a decision tool, not a blame tool: describe impact on users, not on people.
- Do not invent traffic numbers; if counts are missing, ask for them or say which result cannot be computed.
FILE:references/slo-math.md
# SLO Math

## Definitions
- **SLI** (service level indicator) = good events / total events, as a percentage.
- **SLO** (objective) = target SLI over a window, e.g. 99.9% of requests succeed over a rolling 30 days.
- **Error budget** = 1 - SLO = the share of events allowed to fail. 99.9% -> 0.1%.
- **Budget in events** = (1 - SLO) x total events in the window.
- **Error rate** over a window = bad / total in that window.
- **Burn rate** = error rate / (1 - SLO). 1.0 means the budget would be exactly used up at the end of the window.

## Projected consumption with partial data
With D seconds of data and W seconds of SLO window:
- expected requests in window = (requests in data / D) x W
- allowed bad = (1 - SLO) x expected requests
- consumed = bad in data / allowed bad
- time to exhaustion at a recent rate = (allowed bad - bad so far) / (bad per second in that recent window)

## Allowed full downtime
| SLO | per 30 days | per 28 days | per week | per day |
|---|---|---|---|---|
| 99% | 7h 12m | 6h 43m 12s | 1h 40m 48s | 14m 24s |
| 99.5% | 3h 36m | 3h 21m 36s | 50m 24s | 7m 12s |
| 99.9% | 43m 12s | 40m 19s | 10m 5s | 1m 26s |
| 99.95% | 21m 36s | 20m 10s | 5m 2s | 43s |
| 99.99% | 4m 19s | 4m 2s | 1m 0s | 9s |

Request-based SLOs do not care about clock time directly: 10 minutes of 50% errors costs the same as 5 minutes of full downtime at equal traffic, and an outage at 3 a.m. costs less budget than one at peak.

## Worked example
SLO 99.9%, 30d window, 1,000,000 requests per day.
- Budget: 0.001 x 30,000,000 = 30,000 failed requests.
- A 2% error rate for one hour at 41,667 req/h burns 833 failures = 2.8% of the budget, burn rate 20.
- At burn rate 20 the whole budget lasts 30d / 20 = 36 hours.

## Pitfalls
- Counting retries as separate good events hides user-facing failures.
- Very low traffic (night, new region) makes error rates noisy; use a minimum request count or a longer short window.
- Synthetic probes measure the probe, not users; label them separately.
- Changing the SLO mid-window: recompute the whole window with the new target and say so.
FILE:references/burn-rate-alerting.md
# Burn-Rate Alerting

## Default policy (multiwindow, multi-burn-rate)
| Severity | Budget share | Long window | Short window | Burn threshold (30d) | Burn threshold (28d) |
|---|---|---|---|---|---|
| page | 2% | 1h | 5m | 14.4 | 13.44 |
| page | 5% | 6h | 30m | 6.0 | 5.6 |
| ticket | 10% | 1d | 2h | 3.0 | 2.8 |
| ticket | 10% | 3d | 6h | 1.0 | 0.93 |

Threshold = budget share x SLO window / long window. Example: 0.02 x 720h / 1h = 14.4.

## Why two windows
- The **long window** proves the problem is significant (enough budget spent).
- The **short window** (about 1/12 of the long one) proves it is still happening, so the alert resets within minutes after recovery instead of staying red for an hour.
- Fire only when both exceed the threshold.

## Example alert rule (Prometheus style)
```
(
  sum(rate(http_requests_errors_total[1h])) / sum(rate(http_requests_total[1h])) > (14.4 * 0.001)
and
  sum(rate(http_requests_errors_total[5m])) / sum(rate(http_requests_total[5m])) > (14.4 * 0.001)
)
```

## Tuning
- Low-traffic services: require a minimum number of requests in the short window, or page on 6h only.
- Too many pages: check whether the SLO is stricter than users need before raising thresholds.
- Missed incidents: slow burns below 1.0 never alert by design; review the budget weekly.
- Data step coarser than the short window (e.g. 15-minute buckets): the short window becomes one bucket and recovery detection is slower.

## Budget policy ideas
- Over 50% consumed: review risky changes, prefer canaries.
- Over 75%: release freeze for non-fixes, reliability work first.
- Exhausted: postmortem required; freeze until the rolling window recovers.
FILE:templates/slo-status-report.md
# SLO Status - <service> - <period or incident>

**SLI:** <what counts as good>  **SLO:** <target>% over rolling <28d/30d>
**Data:** <start> to <end> UTC, step <interval>, <n> missing intervals
**Calculator:** `slo_budget.py --slo <target> --window <window>` (exit <code>)

## Budget
| Metric | Value |
|---|---|
| SLI in period | |
| Budget consumed | |
| Budget remaining | |
| Allowed full downtime per window | |

## Burn rates
| 5m | 30m | 1h | 2h | 6h | 1d | 3d |
|---|---|---|---|---|---|---|
| | | | | | | |

## Alerts
| Policy | Status | Long / short burn |
|---|---|---|
| page 1h/5m | | |
| page 6h/30m | | |
| ticket 1d/2h | | |
| ticket 3d/6h | | |

## What happened
<worst interval, user impact, cause if known>

## Decision
- [ ] Page / incident open
- [ ] Release policy: <normal / caution / freeze> because <remaining budget>
- [ ] SLO or alert tuning: <change and reason>

## Summary for stakeholders
> <service> met <SLI>% against a <target>% target. <x>% of this window's error budget is used; at the current pace it lasts <time>. Next step: <decision>.
FILE:examples/checkout-api-requests.csv
timestamp,total,good
2026-10-18T00:00:00Z,4757,4755
2026-10-18T00:15:00Z,4525,4523
2026-10-18T00:30:00Z,4313,4311
2026-10-18T00:45:00Z,4122,4120
2026-10-18T01:00:00Z,3951,3949
2026-10-18T01:15:00Z,3803,3801
2026-10-18T01:30:00Z,3678,3676
2026-10-18T01:45:00Z,3577,3576
2026-10-18T02:00:00Z,3500,3498
2026-10-18T02:15:00Z,3448,3446
2026-10-18T02:30:00Z,3421,3419
2026-10-18T02:45:00Z,3019,3018
2026-10-18T03:00:00Z,3044,3043
2026-10-18T03:15:00Z,3093,3092
2026-10-18T03:30:00Z,3169,3168
2026-10-18T03:45:00Z,3270,3268
2026-10-18T04:00:00Z,3396,3394
2026-10-18T04:15:00Z,3547,3545
2026-10-18T04:30:00Z,3722,3720
2026-10-18T04:45:00Z,3921,3919
2026-10-18T05:00:00Z,4143,4141
2026-10-18T05:15:00Z,4388,4386
2026-10-18T05:30:00Z,4253,4251
2026-10-18T05:45:00Z,4539,4537
2026-10-18T06:00:00Z,4845,4843
2026-10-18T06:15:00Z,5168,5166
2026-10-18T06:30:00Z,5509,5507
2026-10-18T06:45:00Z,5865,5863
2026-10-18T07:00:00Z,6236,6234
2026-10-18T07:15:00Z,6619,6616
2026-10-18T07:30:00Z,7013,7009
2026-10-18T07:45:00Z,7418,7414
2026-10-18T08:00:00Z,7831,7827
2026-10-18T08:15:00Z,7850,7847
2026-10-18T08:30:00Z,8274,8271
2026-10-18T08:45:00Z,8702,8699
2026-10-18T09:00:00Z,9132,9127
2026-10-18T09:15:00Z,9561,9556
2026-10-18T09:30:00Z,9989,9984
2026-10-18T09:45:00Z,10413,10408
2026-10-18T10:00:00Z,10832,10827
2026-10-18T10:15:00Z,11245,11240
2026-10-18T10:30:00Z,11650,11645
2026-10-18T10:45:00Z,12044,12038
2026-10-18T11:00:00Z,12028,12022
2026-10-18T11:15:00Z,12398,12392
2026-10-18T11:30:00Z,12754,12748
2026-10-18T11:45:00Z,13095,13089
2026-10-18T12:00:00Z,13418,13412
2026-10-18T12:15:00Z,13724,13719
2026-10-18T12:30:00Z,14010,14003
2026-10-18T12:45:00Z,14275,14268
2026-10-18T13:00:00Z,14520,14513
2026-10-18T13:15:00Z,14742,14735
2026-10-18T13:30:00Z,14941,14934
2026-10-18T13:45:00Z,14716,14710
2026-10-18T14:00:00Z,14867,14861
2026-10-18T14:15:00Z,14993,14985
2026-10-18T14:30:00Z,15094,15086
2026-10-18T14:45:00Z,15170,15163
2026-10-18T15:00:00Z,15220,15213
2026-10-18T15:15:00Z,15244,15237
2026-10-18T15:30:00Z,15242,15236
2026-10-18T15:45:00Z,15215,15209
2026-10-18T16:00:00Z,15163,15155
2026-10-18T16:15:00Z,14686,14679
2026-10-18T16:30:00Z,14585,14578
2026-10-18T16:45:00Z,14460,14453
2026-10-18T17:00:00Z,14312,14306
2026-10-18T17:15:00Z,14141,14135
2026-10-18T17:30:00Z,13950,13866
2026-10-18T17:45:00Z,13738,13656
2026-10-18T18:00:00Z,13506,13499
2026-10-18T18:15:00Z,13257,13251
2026-10-18T18:30:00Z,12990,12984
2026-10-18T18:45:00Z,12708,12702
2026-10-18T19:00:00Z,12012,12007
2026-10-18T19:15:00Z,11702,11697
2026-10-18T19:30:00Z,11382,11376
2026-10-18T19:45:00Z,11051,11045
2026-10-18T20:00:00Z,10712,10707
2026-10-18T20:15:00Z,10367,10362
2026-10-18T20:30:00Z,10017,10013
2026-10-18T20:45:00Z,9663,9659
2026-10-18T21:00:00Z,9308,9304
2026-10-18T21:15:00Z,8952,8947
2026-10-18T21:30:00Z,8598,8594
2026-10-18T21:45:00Z,7848,7844
2026-10-18T22:00:00Z,7503,7500
2026-10-18T22:15:00Z,7164,7161
2026-10-18T22:30:00Z,6833,6830
2026-10-18T22:45:00Z,6513,6510
2026-10-18T23:00:00Z,6204,6201
2026-10-18T23:15:00Z,5907,5904
2026-10-18T23:30:00Z,5625,5622
2026-10-18T23:45:00Z,5358,5356
2026-10-19T00:00:00Z,5109,5107
2026-10-19T00:15:00Z,4877,4875
2026-10-19T00:30:00Z,4265,4263
2026-10-19T00:45:00Z,4074,4072
2026-10-19T01:00:00Z,3903,3901
2026-10-19T01:15:00Z,3755,3753
2026-10-19T01:30:00Z,3630,3628
2026-10-19T01:45:00Z,3529,3527
2026-10-19T02:00:00Z,3452,3451
2026-10-19T02:15:00Z,3400,3399
2026-10-19T02:30:00Z,3373,3371
2026-10-19T02:45:00Z,3371,3369
2026-10-19T03:00:00Z,3396,3394
2026-10-19T03:15:00Z,3045,3044
2026-10-19T03:30:00Z,3121,3120
2026-10-19T03:45:00Z,3222,3221
2026-10-19T04:00:00Z,3348,3347
2026-10-19T04:15:00Z,3499,3497
2026-10-19T04:30:00Z,3674,3672
2026-10-19T04:45:00Z,3873,3871
2026-10-19T05:00:00Z,4095,4093
2026-10-19T05:15:00Z,4340,4338
2026-10-19T05:30:00Z,4605,4603
2026-10-19T05:45:00Z,4491,4489
2026-10-19T06:00:00Z,4797,4795
2026-10-19T06:15:00Z,5120,5117
2026-10-19T06:30:00Z,5461,5458
2026-10-19T06:45:00Z,5817,5814
2026-10-19T07:00:00Z,6188,6185
2026-10-19T07:15:00Z,6571,6568
2026-10-19T07:30:00Z,6965,6962
2026-10-19T07:45:00Z,7370,7366
2026-10-19T08:00:00Z,7783,7779
2026-10-19T08:15:00Z,8202,8198
2026-10-19T08:30:00Z,8226,8222
2026-10-19T08:45:00Z,8654,8650
2026-10-19T09:00:00Z,9084,9080
2026-10-19T09:15:00Z,9513,9509
2026-10-19T09:30:00Z,9941,9936
2026-10-19T09:45:00Z,10365,10360
2026-10-19T10:00:00Z,10784,10779
2026-10-19T10:15:00Z,11197,11192
2026-10-19T10:30:00Z,11602,11597
2026-10-19T10:45:00Z,11996,11991
2026-10-19T11:00:00Z,12380,12375
2026-10-19T11:15:00Z,12350,12344
2026-10-19T11:30:00Z,12706,12700
2026-10-19T11:45:00Z,13047,13041
2026-10-19T12:00:00Z,13370,13364
2026-10-19T12:15:00Z,13676,13670
2026-10-19T12:30:00Z,13962,13956
2026-10-19T12:45:00Z,14227,14221
2026-10-19T13:00:00Z,14472,14464
2026-10-19T13:15:00Z,14694,14687
2026-10-19T13:30:00Z,14893,14886
2026-10-19T13:45:00Z,15068,15061
2026-10-19T14:00:00Z,14819,14812
2026-10-19T14:15:00Z,14945,14939
2026-10-19T14:30:00Z,15046,15040
2026-10-19T14:45:00Z,15122,15114
2026-10-19T15:00:00Z,15172,15164
2026-10-19T15:15:00Z,15196,15189
2026-10-19T15:30:00Z,15194,15187
2026-10-19T15:45:00Z,15167,15160
2026-10-19T16:00:00Z,15115,15109
2026-10-19T16:15:00Z,15038,15032
2026-10-19T16:30:00Z,14937,14929
2026-10-19T16:45:00Z,14412,14405
2026-10-19T17:00:00Z,14264,14257
2026-10-19T17:15:00Z,14093,14087
2026-10-19T17:30:00Z,13902,13896
2026-10-19T17:45:00Z,13690,13684
2026-10-19T18:00:00Z,13458,13453
2026-10-19T18:15:00Z,13209,13202
2026-10-19T18:30:00Z,12942,12936
2026-10-19T18:45:00Z,12660,12654
2026-10-19T19:00:00Z,12364,12358
2026-10-19T19:15:00Z,11654,11649
2026-10-19T19:30:00Z,11334,11329
2026-10-19T19:45:00Z,11003,10999
2026-10-19T20:00:00Z,10664,10658
2026-10-19T20:15:00Z,10319,10314
2026-10-19T20:30:00Z,9969,9964
2026-10-19T20:45:00Z,9615,9611
2026-10-19T21:00:00Z,9260,9256
2026-10-19T21:15:00Z,8904,8900
2026-10-19T21:30:00Z,8550,8547
2026-10-19T21:45:00Z,8200,8196
2026-10-19T22:00:00Z,7455,7298
2026-10-19T22:15:00Z,7116,6959
2026-10-19T22:30:00Z,6785,6629
2026-10-19T22:45:00Z,6465,6310
2026-10-19T23:00:00Z,6156,6002
2026-10-19T23:15:00Z,5859,5707
2026-10-19T23:30:00Z,5577,5426
2026-10-19T23:45:00Z,5310,5161
FILE:scripts/slo_budget.py
#!/usr/bin/env python3
"""SLO error budget and burn rate calculator (Python 3 standard library only).

Input: CSV with one row per interval, the timestamp marking the START of the interval, and request
counts. Accepted columns (case-insensitive): timestamp|time|ts, total|requests|count, and either
good|success|ok or bad|errors|failed. Timestamps: ISO 8601 (Z or offset) or Unix seconds.

Math (see references/slo-math.md):
  error budget (fraction)  = 1 - SLO                          e.g. 99.9% -> 0.001
  error rate over window   = sum(bad) / sum(total)
  burn rate                = error rate / (1 - SLO)           1.0 = spends exactly the budget over the SLO window
  budget consumed so far   = sum(bad) / ((1 - SLO) * expected requests in the SLO window)
  alert threshold          = budget share * window hours / long window hours   (14.4 = 2% of 30d in 1h)

Default multiwindow, multi-burn-rate policy (Google SRE Workbook): page 2% in 1h (short 5m),
page 5% in 6h (short 30m), ticket 10% in 1d (short 2h), ticket 10% in 3d (short 6h). An alert fires
only when BOTH the long and the short window burn faster than the threshold. Windows shorter than
the data step are rounded up to one step; windows longer than the data are reported as not evaluable.

Usage:
  python3 slo_budget.py requests.csv --slo 99.9 --window 30d
  python3 slo_budget.py requests.csv --slo 99.5 --window 28d --json
  python3 slo_budget.py --allowance --slo 99.95 --window 30d     # downtime allowed, no CSV needed
  python3 slo_budget.py --self-test
Exit code: 0 OK, 1 a page alert fires or the budget is exhausted, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import json
import re
import statistics
import sys

POLICY = [
    # (severity, long window, short window, budget share)
    ("page", "1h", "5m", 0.02),
    ("page", "6h", "30m", 0.05),
    ("ticket", "1d", "2h", 0.10),
    ("ticket", "3d", "6h", 0.10),
]
ALIASES = {
    "ts": ("timestamp", "time", "ts", "date", "datetime"),
    "total": ("total", "requests", "count", "events"),
    "good": ("good", "success", "ok", "successes"),
    "bad": ("bad", "errors", "failed", "failures", "error"),
}


class SLOError(ValueError):
    pass


def parse_duration(s):
    m = re.fullmatch(r"\s*(\d+(?:\.\d+)?)\s*([smhdw])\s*", str(s).lower())
    if not m:
        raise SLOError(f"bad duration {s!r} (use e.g. 5m, 1h, 28d)")
    return float(m.group(1)) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[m.group(2)]


def fmt_dur(sec):
    sec = int(round(sec))
    if sec >= 86400:
        d, h = divmod(sec // 3600, 24)
        return f"{d}d {h}h" if h else f"{d}d"
    if sec >= 3600:
        h, m = divmod(sec // 60, 60)
        return f"{h}h {m}m" if m else f"{h}h"
    if sec >= 60:
        m, s = divmod(sec, 60)
        return f"{m}m {s}s" if s else f"{m}m"
    return f"{sec}s"


def parse_ts(v):
    v = v.strip()
    if re.fullmatch(r"\d+(\.\d+)?", v):
        return dt.datetime.fromtimestamp(float(v), dt.timezone.utc)
    t = dt.datetime.fromisoformat(v.replace("Z", "+00:00").replace(" ", "T"))
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.astimezone(dt.timezone.utc)


def load_rows(lines):
    rd = csv.DictReader(lines)
    if not rd.fieldnames:
        raise SLOError("empty CSV")
    cols = {}
    lower = {f.strip().lower(): f for f in rd.fieldnames}
    for key, names in ALIASES.items():
        for n in names:
            if n in lower:
                cols[key] = lower[n]
                break
    if "ts" not in cols or "total" not in cols or not ({"good", "bad"} & cols.keys()):
        raise SLOError(f"need timestamp, total and good or bad columns; got {rd.fieldnames}")
    rows = []
    for i, r in enumerate(rd, 2):
        try:
            ts = parse_ts(r[cols["ts"]])
            total = int(float(r[cols["total"]]))
            bad = int(float(r[cols["bad"]])) if "bad" in cols else total - int(float(r[cols["good"]]))
        except (ValueError, TypeError) as e:
            raise SLOError(f"line {i}: {e}")
        if total < 0 or bad < 0 or bad > total:
            raise SLOError(f"line {i}: counts must satisfy 0 <= bad <= total (total={total}, bad={bad})")
        rows.append((ts, total, bad))
    if len(rows) < 2:
        raise SLOError("need at least two rows")
    rows.sort(key=lambda r: r[0])
    return rows


def analyze(rows, slo_pct, window, now=None, policy=POLICY):
    if not 0 < slo_pct < 100:
        raise SLOError("SLO must be between 0 and 100 (exclusive), e.g. 99.9")
    budget = 1 - slo_pct / 100
    win = parse_duration(window)
    diffs = [(b[0] - a[0]).total_seconds() for a, b in zip(rows, rows[1:])]
    step = statistics.median(diffs)
    if step <= 0:
        raise SLOError("duplicate timestamps")
    first = rows[0][0]
    end = (now or rows[-1][0] + dt.timedelta(seconds=step))
    covered = (end - first).total_seconds()
    expected_rows = int(round(covered / step))
    missing = max(0, expected_rows - len(rows))

    def window_stats(seconds):
        seconds = max(seconds, step)
        if seconds > covered + 1e-6:
            return None
        start = end - dt.timedelta(seconds=seconds)
        tot = sum(t for ts, t, b in rows if start <= ts < end)
        bad = sum(b for ts, t, b in rows if start <= ts < end)
        rate = bad / tot if tot else 0.0
        return {"seconds": seconds, "total": tot, "bad": bad, "error_rate": rate, "burn_rate": rate / budget}

    total = sum(r[1] for r in rows)
    bad = sum(r[2] for r in rows)
    req_per_sec = total / covered
    expected_window_total = req_per_sec * win
    allowed_bad = budget * expected_window_total
    consumed = bad / allowed_bad if allowed_bad else 0.0
    alerts = []
    for sev, lw, sw, share in policy:
        thr = share * win / parse_duration(lw)
        L, S = window_stats(parse_duration(lw)), window_stats(parse_duration(sw))
        if L is None or S is None:
            alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr, "status": "not evaluable",
                           "long_burn": None, "short_burn": None})
            continue
        fires = L["burn_rate"] >= thr and S["burn_rate"] >= thr
        alerts.append({"severity": sev, "long": lw, "short": sw, "threshold": thr,
                       "status": "FIRING" if fires else "ok",
                       "long_burn": L["burn_rate"], "short_burn": S["burn_rate"]})
    burn = {}
    for w in ("5m", "30m", "1h", "2h", "6h", "1d", "3d"):
        s = window_stats(parse_duration(w))
        burn[w] = None if s is None else s["burn_rate"]
    remaining_bad = allowed_bad - bad
    exhaust = {}
    for w in ("1h", "6h", "1d"):
        s = window_stats(parse_duration(w))
        if s and s["bad"] > 0 and remaining_bad > 0:
            exhaust[w] = remaining_bad / (s["bad"] / s["seconds"])
        elif s:
            exhaust[w] = 0.0 if remaining_bad <= 0 else None
    worst = max(rows, key=lambda r: (r[2] / r[1] if r[1] else 0, r[0]))
    return {
        "slo_percent": slo_pct, "window": window, "window_seconds": win, "step_seconds": step,
        "data_start": first.strftime("%Y-%m-%dT%H:%M:%SZ"), "data_end": end.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "rows": len(rows), "missing_intervals": missing,
        "total": total, "bad": bad, "sli_percent": 100 * (1 - bad / total) if total else 100.0,
        "allowed_bad_in_window": allowed_bad, "budget_consumed": consumed, "budget_remaining": 1 - consumed,
        "allowed_downtime_seconds": budget * win,
        "burn_rates": burn, "alerts": alerts, "exhaustion_seconds": exhaust,
        "worst_interval": {"start": worst[0].strftime("%Y-%m-%dT%H:%M:%SZ"), "total": worst[1], "bad": worst[2],
                           "error_rate": worst[2] / worst[1] if worst[1] else 0.0},
    }


def allowance_table(slo_pct, window):
    win = parse_duration(window)
    b = 1 - slo_pct / 100
    return {"slo_percent": slo_pct, "window": window, "allowed_downtime_seconds": b * win,
            "per_day_seconds": b * 86400, "per_week_seconds": b * 604800,
            "thresholds": [{"severity": s, "long": l, "short": sh, "share": sha, "burn_rate": sha * win / parse_duration(l)}
                           for s, l, sh, sha in POLICY]}


def to_markdown(r):
    pct = lambda x: f"{100 * x:.1f}%"
    out = [f"# SLO error budget report", "",
           f"**SLO:** {r['slo_percent']}% over {r['window']}  **Data:** {r['data_start']} to {r['data_end']} "
           f"({r['rows']} rows, step {fmt_dur(r['step_seconds'])}, {r['missing_intervals']} missing)",
           f"**SLI over data:** {r['sli_percent']:.4f}%  ({r['bad']:,} bad of {r['total']:,})",
           f"**Error budget:** {r['allowed_bad_in_window']:,.0f} bad requests per {r['window']} at current traffic "
           f"(= {fmt_dur(r['allowed_downtime_seconds'])} of full downtime)",
           f"**Consumed so far:** {pct(r['budget_consumed'])}  **Remaining:** {pct(r['budget_remaining'])}", "",
           "## Burn rates (1.0 = on budget)", "", "| Window | Burn rate |", "|---|---|"]
    for w, v in r["burn_rates"].items():
        out.append(f"| {w} | {'n/a' if v is None else f'{v:.2f}'} |")
    out += ["", "## Alert policy", "", "| Severity | Long / short window | Threshold | Long burn | Short burn | Status |",
            "|---|---|---|---|---|---|"]
    for a in r["alerts"]:
        lb = "n/a" if a["long_burn"] is None else f"{a['long_burn']:.2f}"
        sb = "n/a" if a["short_burn"] is None else f"{a['short_burn']:.2f}"
        out.append(f"| {a['severity']} | {a['long']} / {a['short']} | {a['threshold']:.1f} | {lb} | {sb} | {a['status']} |")
    out += ["", "## Time to exhaustion at recent burn"]
    for w, s in r["exhaustion_seconds"].items():
        out.append(f"- at last {w} rate: " + ("budget already exhausted" if s == 0 else ("no errors in window" if s is None else fmt_dur(s))))
    wi = r["worst_interval"]
    out += ["", f"Worst interval: {wi['start']} with {wi['bad']:,} bad of {wi['total']:,} ({100 * wi['error_rate']:.2f}%)"]
    return "\n".join(out)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    t0 = dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)

    def series(n, step_min, total, bad_fn):
        return [(t0 + dt.timedelta(minutes=step_min * i), total, bad_fn(i)) for i in range(n)]

    ok("duration parse", parse_duration("28d") == 28 * 86400 and parse_duration("5m") == 300)
    ok("duration format", fmt_dur(43.2 * 60) == "43m 12s" and fmt_dur(86400 * 3) == "3d")
    a = allowance_table(99.9, "30d")
    ok("99.9% of 30d allows 43m 12s", round(a["allowed_downtime_seconds"]) == 2592)
    ok("page threshold 14.4 for 30d", abs(a["thresholds"][0]["burn_rate"] - 14.4) < 1e-9)
    ok("28d threshold scales to 13.44", abs(allowance_table(99.9, "28d")["thresholds"][0]["burn_rate"] - 13.44) < 1e-9)
    flat = series(4 * 24 * 4, 15, 10000, lambda i: 10)          # 0.1% errors at 99.9 -> burn 1.0
    r = analyze(flat, 99.9, "30d")
    ok("steady burn rate is 1.0", abs(r["burn_rates"]["1h"] - 1.0) < 1e-9 and abs(r["burn_rates"]["1d"] - 1.0) < 1e-9)
    ok("no alert at burn 1.0", all(x["status"] != "FIRING" for x in r["alerts"][:3]))
    ok("burn 1.0 over 4 of 30 days consumes 4/30", abs(r["budget_consumed"] - 4 / 30) < 1e-9)
    ok("SLI computed", abs(r["sli_percent"] - 99.9) < 1e-9)
    spike = series(4 * 24 * 4, 15, 10000, lambda i: 200 if i >= 4 * 24 * 4 - 4 else 5)   # last hour 2% errors
    r = analyze(spike, 99.9, "30d")
    ok("fast burn pages (1h/5m)", r["alerts"][0]["status"] == "FIRING")
    ok("short window rounded up to step", abs(r["burn_rates"]["5m"] - 20.0) < 1e-9)
    ok("6h page not firing for 1h spike", r["alerts"][1]["status"] == "ok")
    recovered = spike[:-2] + [(spike[-2][0], 10000, 5), (spike[-1][0], 10000, 5)]
    r = analyze(recovered, 99.9, "30d")
    ok("short window resets page after recovery", r["alerts"][0]["status"] == "ok")
    short = series(8, 15, 1000, lambda i: 0)
    r = analyze(short, 99.9, "30d")
    ok("long windows not evaluable on 2h of data", r["alerts"][3]["status"] == "not evaluable" and r["burn_rates"]["3d"] is None)
    gap = flat[:10] + flat[12:]
    ok("missing intervals counted", analyze(gap, 99.9, "30d")["missing_intervals"] == 2)
    ok("exhaustion time at burn 1", abs(analyze(flat, 99.9, "30d")["exhaustion_seconds"]["1d"] - 26 * 86400) < 1)
    rows = load_rows(["time,requests,errors", "2026-01-01T00:00:00Z,100,1", "2026-01-01T00:05:00Z,100,0"])
    ok("aliases requests/errors", rows[0][1] == 100 and rows[0][2] == 1)
    rows = load_rows(["ts,total,good", "1767225600,50,49", "1767225900,50,50"])
    ok("unix timestamps and good column", rows[0][2] == 1 and rows[0][0] == t0)
    for data, label in [(["ts,total", "x,1"], "missing good/bad rejected"),
                        (["ts,total,bad", "2026-01-01T00:00:00Z,10,11", "2026-01-01T00:05:00Z,10,0"], "bad > total rejected"),
                        (["ts,total,bad", "nope,10,1", "2026-01-01T00:05:00Z,10,0"], "bad timestamp rejected")]:
        try:
            load_rows(data)
            ok(label, False)
        except SLOError:
            ok(label, True)
    try:
        analyze(flat, 100, "30d")
        ok("SLO 100 rejected", False)
    except SLOError:
        ok("SLO 100 rejected", True)
    ok("markdown has alert table", "## Alert policy" in to_markdown(analyze(flat, 99.9, "30d")))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Compute SLO error budget consumption and multiwindow burn-rate alerts.")
    ap.add_argument("file", nargs="?", help="CSV of request counts per interval ('-' for stdin)")
    ap.add_argument("--slo", type=float, default=99.9, help="target in percent (default 99.9)")
    ap.add_argument("--window", default="30d", help="SLO window, e.g. 28d or 30d (default 30d)")
    ap.add_argument("--now", help="evaluate as of this ISO time (default: end of the last interval)")
    ap.add_argument("--allowance", action="store_true", help="print allowed downtime and alert thresholds only")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    try:
        if a.allowance:
            t = allowance_table(a.slo, a.window)
            if a.json:
                print(json.dumps(t, indent=2))
            else:
                print(f"SLO {a.slo}% over {a.window}: {fmt_dur(t['allowed_downtime_seconds'])} of full downtime "
                      f"({fmt_dur(t['per_day_seconds'])}/day, {fmt_dur(t['per_week_seconds'])}/week)")
                for x in t["thresholds"]:
                    print(f"  {x['severity']:6} {x['share'] * 100:.0f}% of budget in {x['long']} (short {x['short']}): burn rate >= {x['burn_rate']:.2f}")
            return 0
        if not a.file:
            ap.error("CSV file required (or --allowance / --self-test)")
        lines = sys.stdin.read().splitlines() if a.file == "-" else open(a.file, encoding="utf-8").read().splitlines()
        r = analyze(load_rows(lines), a.slo, a.window, parse_ts(a.now) if a.now else None)
    except (OSError, SLOError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    paging = any(x["severity"] == "page" and x["status"] == "FIRING" for x in r["alerts"])
    return 1 if paging or r["budget_remaining"] <= 0 else 0


if __name__ == "__main__":
    sys.exit(main())

# Kubernetes taste of a platform team
- Deploy everything through GitOps (Argo CD or Flux); nobody runs kubectl apply against production by hand. Confidence: 0.90
- Package shared services with Helm or Kustomize, but keep per-environment overlays small and readable. Confidence: 0.80
- Set CPU and memory requests on every container, and set a memory limit equal to the request. Confidence: 0.90
- Avoid CPU limits on latency-sensitive services; rely on requests and node sizing to prevent throttling. Confidence: 0.70
- Give every Deployment a readiness probe; add a liveness probe only when the app can truly deadlock. Confidence: 0.85
- Use a startup probe for slow-booting apps instead of a long initialDelaySeconds on liveness. Confidence: 0.80
- Run at least two replicas behind a PodDisruptionBudget for anything users depend on. Confidence: 0.90
- Spread replicas across zones with topologySpreadConstraints. Confidence: 0.80
- Pin images by digest or an immutable version tag, never latest. Confidence: 0.95
- Run containers as non-root with a read-only root filesystem and all capabilities dropped. Confidence: 0.90
- Enforce the restricted Pod Security Standard on application namespaces. Confidence: 0.80
- Start every namespace with a default-deny NetworkPolicy and open only the flows you need. Confidence: 0.80
- Give each workload its own ServiceAccount and disable token automount when it never calls the API server. Confidence: 0.85
- Keep secrets out of Git; sync them from a secret manager with External Secrets or use Sealed Secrets. Confidence: 0.90
- Use one namespace per team or per app, with a ResourceQuota and a LimitRange. Confidence: 0.80
- Apply the standard app.kubernetes.io labels to every resource. Confidence: 0.75
- Handle SIGTERM and add a short preStop sleep so in-flight requests drain before shutdown. Confidence: 0.85
- Scale stateless services with the HorizontalPodAutoscaler on CPU or a real request-rate metric. Confidence: 0.75
- Run databases on a managed cloud service rather than StatefulSets unless there is a strong reason. Confidence: 0.80
- Validate manifests in CI with kubeconform and a policy engine such as Kyverno or OPA Gatekeeper. Confidence: 0.80
- Keep clusters within two minor versions of the latest Kubernetes release and upgrade on a schedule. Confidence: 0.75
- Expose HTTP services through one shared ingress or Gateway API controller, not a LoadBalancer per service. Confidence: 0.75

---
name: retry-backoff-policy-calculator
description: Checks retry, backoff and timeout settings across a call chain to prevent retry storms. Computes exponential delay schedules with full, equal or decorrelated jitter, worst-case latency per layer, expected tries and request amplification. Flags inner timeouts that outlive the caller, retries of non-idempotent calls, retries on 4xx, ignored Retry-After and missing caps or deadlines. Includes a stdlib Python calculator with --self-test.
---

# Retry and Backoff Policy Calculator

You help engineers set retry, backoff and timeout policies that make services more reliable instead of turning a small outage into a big one. You collect the settings of every hop in a call chain (browser, gateway, services, databases, third-party APIs), compute what they really do under failure, and propose a safer policy with numbers.

## Files in this skill

- `scripts/retry_policy.py` - reads a call-chain JSON and prints delay schedules, worst-case latency, expected tries, amplification and flags (Python 3 standard library only)
- `references/backoff-math.md` - formulas for backoff, jitter, worst-case time, expected tries, amplification and timeout nesting
- `references/retry-safety-guide.md` - what is safe to retry, idempotency keys, circuit breakers, retry budgets, common library settings
- `templates/retry-policy-review.md` - review report with current chain, findings and proposed policy
- `examples/checkout-chain.json` - a three-layer checkout chain with typical mistakes

## Workflow

### 1. Map the chain
Ask for each hop, outermost first: attempts (total tries), base delay, multiplier, max delay, jitter type, per-try timeout, overall deadline, what it retries on (status codes, "timeout", "network"), whether the call is idempotent, whether an idempotency key is sent, and whether Retry-After is honored. Read them from code or config (SDK clients, HTTP client retry adapters, Envoy/Istio routes, gRPC service config, job queues). Unknown values: use the library default and mark them as assumed.

### 2. Run the calculator
```
python3 scripts/retry_policy.py examples/checkout-chain.json
python3 scripts/retry_policy.py chain.json --failure-rate 0.5 --max-amplification 8
python3 scripts/retry_policy.py chain.json --json > retry-report.json
python3 scripts/retry_policy.py --self-test
```
Exit code 1 means high-severity findings, 2 means bad input.

### 3. Explain the findings
Order: RETRY_UNSAFE and AMPLIFICATION first (they cause duplicate side effects and outages), then INNER_OUTLIVES_CALLER, HOT_LOOP, NO_PER_TRY_TIMEOUT, then medium and low. Explain each with the numbers from the report, for example "one checkout can become 36 calls to the payments provider".

### 4. Propose a fixed policy
Use `references/backoff-math.md` and `references/retry-safety-guide.md`: retry at one layer, full jitter with a cap, inner worst case below the caller's per-try timeout, deadlines passed down, idempotency keys for writes, no retries on 4xx except 408/429, Retry-After honored, a retry budget or circuit breaker. Re-run the script on the proposed JSON and show before and after.

### 5. Write it up
Fill `templates/retry-policy-review.md` and give config snippets for the user's library.

## Example result
For `examples/checkout-chain.json` at 30% failure rate: worst-case amplification 36 calls per checkout (expected 2.74); the gateway layer's 9.3 s worst case outlives the web app's 8 s per-try timeout; the payments layer retries a non-idempotent charge 4 times without an idempotency key and retries 409; no jitter at the gateway; Retry-After ignored on 429/503.

## Rules

- Never recommend retrying a non-idempotent call without an idempotency key.
- Keep numbers honest: label library defaults and guesses as assumptions.
- Prefer fewer retries with good timeouts over many retries.
- Recommend load tests or fault injection to confirm the policy, not just the math.
FILE:references/backoff-math.md
# Backoff and Timeout Math

## Terms
- **attempts**: total tries including the first. attempts 3 = 1 try + 2 retries.
- **base**: first retry delay. **multiplier**: growth factor (usually 2). **cap**: max single delay.
- **per-try timeout**: how long one try may take before it is abandoned.
- **deadline**: total time budget for the call including all tries and delays.

## Delay before retry k (k = 1 .. attempts-1)
d(k) = min(cap, base x multiplier^(k-1))

| Jitter | Delay | Range | Mean |
|---|---|---|---|
| none | d | d | d |
| full | random(0, d) | 0 .. d | d/2 |
| equal | d/2 + random(0, d/2) | d/2 .. d | 0.75 d |
| decorrelated | min(cap, random(base, 3 x previous)) | base .. min(cap, base x 3^k) | simulated |

Full jitter spreads load best and is the default recommendation. Equal jitter keeps a minimum wait. Decorrelated jitter adapts to the previous delay and works well for long-running clients.

## Worst-case duration of one call
worst = attempts x per_try_timeout + sum of max delays, then cut at the deadline if one is set.
Example: 3 tries x 1 s + 100 ms + 200 ms = 3.3 s.

## Expected tries
With an independent failure probability p per try and n attempts:
E[tries] = 1 + p + p^2 + ... + p^(n-1)
p = 0.3, n = 3 -> 1.39. During a full outage p = 1 and E = n.

## Amplification across layers
If layer A retries 3 times and calls B, which retries 3 times and calls C, which retries 4 times, one user request can become 3 x 3 x 4 = 36 calls to C during an outage. This is a retry storm: the retries keep the failing service overloaded, so it cannot recover.

Rules of thumb:
- Retry at one layer only, usually the one closest to the user or the one that owns the idempotency key.
- Inner layers: attempts 1 or 2 with a short deadline.
- Keep worst-case amplification at or below about 10.
- Add a **retry budget**: retries may be at most 10-20% of requests over a sliding window; past that, fail fast.

## Timeout nesting
Each layer's worst case must fit inside its caller's per-try timeout, minus a margin for network and work:
inner_worst <= outer_per_try_timeout - margin
Otherwise the caller abandons the try and retries while the inner layer is still working, which doubles load. Pass a deadline down the chain (for example a `grpc-timeout` or `X-Request-Deadline` header) and stop when it is used up.

## Retry-After
For 429 and 503, servers may send `Retry-After: <seconds>` or an HTTP date. Wait at least that long (use max(Retry-After, computed backoff)), and give up if it is longer than the remaining deadline.
FILE:references/retry-safety-guide.md
# Retry Safety Guide

## What to retry
| Outcome | Retry? | Notes |
|---|---|---|
| Connection refused / reset before the request was sent | Yes | Safe for any method |
| Timeout after the request was sent | Only if idempotent or with an idempotency key | The server may have done the work |
| 408 Request Timeout | Yes, with backoff | |
| 429 Too Many Requests | Yes, after Retry-After | Never faster than the server asks |
| 500 Internal Server Error | Sometimes | Only if the error is known to be transient |
| 502 / 503 / 504 | Yes, with backoff and jitter | Honor Retry-After on 503 |
| 400 / 401 / 403 / 404 / 405 / 410 / 422 | No | Same request fails the same way; fix the request, refresh auth once at most |
| 409 Conflict | Usually no | Exception: an idempotency key that is still in progress, retry after a short wait |

## Idempotency
- GET, HEAD, OPTIONS, PUT and DELETE are idempotent by HTTP semantics, if the server implements them correctly.
- POST and PATCH are not. To retry them, send an **idempotency key** (a UUID per logical operation, e.g. `Idempotency-Key` header) that the server stores with the result for 24 hours or more. Generate the key once, before the first try, and reuse it for every retry.
- Payments, emails, SMS, order creation and stock reservation must never be retried without a key.

## Client checklist
1. Per-try timeout set and shorter than the caller's.
2. Overall deadline set and passed down the chain.
3. Exponential backoff with a cap and full jitter.
4. attempts <= 3 at the edge, 1-2 inside.
5. Retry only on the table's "Yes" rows.
6. Idempotency key on every retried non-idempotent call.
7. Retry budget or circuit breaker so retries stop during an outage.
8. Metrics: tries per request, retry rate, give-up count, by dependency.
9. Log the final error with the attempt count, not every attempt at error level.

## Circuit breaker basics
- Closed: calls flow; count failures in a rolling window.
- Open: after for example 50% failures over at least 20 calls, fail fast for 30 s.
- Half-open: let a few trial calls through; close if they succeed.
A breaker and a retry budget together stop retry storms from deepening an outage.

## Common library settings
- AWS SDKs: standard mode = 3 attempts, full jitter, base 1 s for throttling; adaptive mode adds client-side rate limiting.
- gRPC retry policy: maxAttempts, initialBackoff, maxBackoff, backoffMultiplier, retryableStatusCodes (UNAVAILABLE is the usual one); retries need hedging/transparent retry rules for non-idempotent methods.
- Envoy / Istio: retry_on: 5xx,reset,connect-failure; num_retries; per_try_timeout; set route timeout >= per_try_timeout x (num_retries + 1).
- Python `urllib3.Retry(total=3, backoff_factor=0.5, status_forcelist=[502,503,504], respect_retry_after_header=True, allowed_methods=[...])`.
Check the library docs for the version you use; defaults change.
FILE:templates/retry-policy-review.md
# Retry Policy Review - <service / call chain> - <date>

**Reviewer:** <name>  **Input:** <chain JSON / config files>  **Failure rate assumed:** <30%>

## Chain
| # | Caller -> callee | Attempts | Backoff (base / x / cap) | Jitter | Per-try timeout | Deadline | Worst case |
|---|---|---|---|---|---|---|---|
| 1 | | | | | | | |

**Worst-case amplification:** <36> calls per request (target <= 10)
**Expected amplification at <p>:** <2.7>

## Findings
| Severity | Layer | Code | Problem | Fix |
|---|---|---|---|---|
| high | | | | |

## Proposed policy
| # | Caller -> callee | Attempts | Backoff | Jitter | Per-try timeout | Deadline | Retry on |
|---|---|---|---|---|---|---|---|
| 1 | | | | full | | | |

**New worst-case amplification:** <n>  **New worst-case latency:** <s>

## Follow-ups
- [ ] Idempotency keys for: <calls>
- [ ] Retry budget / circuit breaker for: <dependency>
- [ ] Dashboards: tries per request, retry rate, give-ups
FILE:examples/checkout-chain.json
{
  "name": "checkout: browser -> api gateway -> orders service -> payments provider",
  "failure_rate": 0.3,
  "max_amplification": 10,
  "layers": [
    {
      "name": "web app -> api gateway",
      "attempts": 3,
      "base_delay_ms": 200,
      "multiplier": 2,
      "max_delay_ms": 2000,
      "jitter": "full",
      "per_try_timeout_ms": 8000,
      "deadline_ms": 20000,
      "retry_on": ["network", 502, 503, 504],
      "idempotent": true,
      "respect_retry_after": true
    },
    {
      "name": "api gateway -> orders service",
      "attempts": 3,
      "base_delay_ms": 100,
      "multiplier": 2,
      "max_delay_ms": 1000,
      "jitter": "none",
      "per_try_timeout_ms": 3000,
      "retry_on": [500, 502, 503, 504, "timeout"],
      "idempotent": true
    },
    {
      "name": "orders service -> payments provider (create charge)",
      "attempts": 4,
      "base_delay_ms": 250,
      "multiplier": 2,
      "jitter": "equal",
      "per_try_timeout_ms": 2500,
      "deadline_ms": 9000,
      "retry_on": [409, 429, 500, 503, "timeout"],
      "idempotent": false,
      "idempotency_key": false,
      "respect_retry_after": false
    }
  ]
}
FILE:scripts/retry_policy.py
#!/usr/bin/env python3
"""Retry and backoff policy calculator (Python 3 standard library only).

Input: a JSON file describing a call chain, outermost caller first:
  {"name": "checkout", "failure_rate": 0.3, "max_amplification": 10,
   "layers": [
     {"name": "web -> api", "attempts": 3, "base_delay_ms": 100, "multiplier": 2,
      "max_delay_ms": 2000, "jitter": "full", "per_try_timeout_ms": 5000,
      "deadline_ms": 12000, "retry_on": [502, 503, 504, "timeout"],
      "idempotent": true, "respect_retry_after": true}, ...]}
attempts = total tries including the first one (attempts 1 = no retries).
jitter: none | full | equal | decorrelated.  Delay before retry k (k = 1..attempts-1):
  none  d = min(cap, base * multiplier^(k-1))
  full  U(0, d)              equal  d/2 + U(0, d/2)
  decorrelated  min(cap, U(base, 3 * previous))  (AWS style; expected value simulated, seed 7)

Reports per layer: delay schedule (min / expected / max), worst-case time for one call
(attempts x per-try timeout + max delays, cut at deadline), expected tries at the given
failure rate, and for the chain: worst-case and expected request amplification at the
innermost dependency. Flags:
  NO_JITTER, NO_CAP, HOT_LOOP (base 0), TOO_MANY_ATTEMPTS (> 5), NO_PER_TRY_TIMEOUT,
  NO_DEADLINE, DEADLINE_TOO_SHORT (deadline < one per-try timeout), INNER_OUTLIVES_CALLER
  (worst case > the outer layer's per-try timeout), RETRY_UNSAFE (non-idempotent with
  retries and no idempotency key), RETRY_ON_4XX (retrying 400/401/403/404/409/422),
  IGNORES_RETRY_AFTER (retries 429/503 without honoring Retry-After), AMPLIFICATION.
Usage:
  python3 retry_policy.py chain.json
  python3 retry_policy.py chain.json --failure-rate 0.5 --json
  python3 retry_policy.py --self-test
Exit code: 0 OK, 1 high-severity flags found, 2 bad input.
"""
import argparse
import json
import random
import sys

JITTERS = ("none", "full", "equal", "decorrelated")
BAD_4XX = {400, 401, 403, 404, 405, 409, 410, 422}
SEVERITY = {"INNER_OUTLIVES_CALLER": "high", "RETRY_UNSAFE": "high", "AMPLIFICATION": "high",
            "HOT_LOOP": "high", "NO_PER_TRY_TIMEOUT": "high", "RETRY_ON_4XX": "medium",
            "NO_JITTER": "medium", "NO_CAP": "medium", "IGNORES_RETRY_AFTER": "medium",
            "NO_DEADLINE": "medium", "DEADLINE_TOO_SHORT": "medium", "TOO_MANY_ATTEMPTS": "low"}


class PolicyError(ValueError):
    pass


def num(layer, key, default=None, minimum=0):
    v = layer.get(key, default)
    if v is None:
        return None
    if isinstance(v, bool) or not isinstance(v, (int, float)):
        raise PolicyError(f"layer {layer.get('name', '?')!r}: {key} must be a number")
    if v < minimum:
        raise PolicyError(f"layer {layer.get('name', '?')!r}: {key} must be >= {minimum}")
    return v


def delays(layer):
    """Return [(min, expected, max)] in ms for each retry."""
    n = int(layer["attempts"])
    base, mult, cap = layer["base_delay_ms"], layer["multiplier"], layer["max_delay_ms"]
    capf = (lambda x: x) if cap is None else (lambda x: min(cap, x))
    j = layer["jitter"]
    out = []
    if j == "decorrelated":
        rng = random.Random(7)
        runs = 20000
        sums = [0.0] * (n - 1)
        for _ in range(runs):
            prev = base
            for k in range(n - 1):
                prev = capf(rng.uniform(base, prev * 3)) if base > 0 else 0.0
                sums[k] += prev
        hi = base
        for k in range(n - 1):
            hi = capf(hi * 3)
            out.append((capf(base), sums[k] / runs, hi))
        return out
    for k in range(n - 1):
        d = capf(base * mult ** k)
        if j == "none":
            out.append((d, d, d))
        elif j == "full":
            out.append((0.0, d / 2, d))
        else:
            out.append((d / 2, d * 0.75, d))
    return out


def normalize(raw):
    layers = raw.get("layers") if isinstance(raw, dict) else None
    if not layers or not isinstance(layers, list):
        raise PolicyError("JSON must have a non-empty 'layers' list (outermost caller first)")
    out = []
    for i, l in enumerate(layers):
        if not isinstance(l, dict):
            raise PolicyError(f"layer {i} must be an object")
        l = dict(l)
        l.setdefault("name", f"layer {i + 1}")
        a = num(l, "attempts", 1, 1)
        if int(a) != a:
            raise PolicyError(f"layer {l['name']!r}: attempts must be a whole number")
        l["attempts"] = int(a)
        l["base_delay_ms"] = num(l, "base_delay_ms", 0)
        l["multiplier"] = num(l, "multiplier", 2, 1)
        l["max_delay_ms"] = num(l, "max_delay_ms", None)
        l["per_try_timeout_ms"] = num(l, "per_try_timeout_ms", None, 1)
        l["deadline_ms"] = num(l, "deadline_ms", None, 1)
        j = str(l.get("jitter", "none")).lower()
        if j not in JITTERS:
            raise PolicyError(f"layer {l['name']!r}: jitter must be one of {', '.join(JITTERS)}")
        l["jitter"] = j
        l["retry_on"] = list(l.get("retry_on", []))
        l["idempotent"] = bool(l.get("idempotent", True))
        l["idempotency_key"] = bool(l.get("idempotency_key", False))
        l["respect_retry_after"] = bool(l.get("respect_retry_after", False))
        out.append(l)
    return out


def expected_tries(n, p):
    if p >= 1:
        return float(n)
    return sum(p ** k for k in range(n))


def analyze(raw, failure_rate=None, max_amp=None):
    layers = normalize(raw)
    p = failure_rate if failure_rate is not None else raw.get("failure_rate", 0.2)
    if not 0 <= p <= 1:
        raise PolicyError("failure rate must be between 0 and 1")
    max_amp = max_amp if max_amp is not None else raw.get("max_amplification", 10)
    res = {"name": raw.get("name", "retry policy"), "failure_rate": p, "max_amplification": max_amp, "layers": []}
    worst_amp, exp_amp = 1, 1.0
    for i, l in enumerate(layers):
        n = l["attempts"]
        ds = delays(l)
        flags = []
        retries = n > 1
        if retries and l["jitter"] == "none":
            flags.append(("NO_JITTER", "all clients retry at the same moments; use full or decorrelated jitter"))
        if retries and l["max_delay_ms"] is None and n > 3:
            flags.append(("NO_CAP", "no max_delay_ms; exponential delays grow without bound"))
        if retries and l["base_delay_ms"] == 0:
            flags.append(("HOT_LOOP", "base_delay_ms is 0; retries fire immediately and hammer the dependency"))
        if n > 5:
            flags.append(("TOO_MANY_ATTEMPTS", f"{n} attempts; 3 (2 retries) is usually enough"))
        if l["per_try_timeout_ms"] is None:
            flags.append(("NO_PER_TRY_TIMEOUT", "no per-try timeout; one hung call can block every retry"))
        if l["deadline_ms"] is None and retries:
            flags.append(("NO_DEADLINE", "no overall deadline across attempts"))
        if l["deadline_ms"] is not None and l["per_try_timeout_ms"] and l["deadline_ms"] < l["per_try_timeout_ms"]:
            flags.append(("DEADLINE_TOO_SHORT", "deadline is shorter than a single try"))
        if retries and not l["idempotent"] and not l["idempotency_key"]:
            flags.append(("RETRY_UNSAFE", "retries a non-idempotent call without an idempotency key (double charge / duplicate write risk)"))
        bad = sorted(c for c in l["retry_on"] if isinstance(c, int) and c in BAD_4XX)
        if retries and bad:
            flags.append(("RETRY_ON_4XX", f"retries client errors {bad}; they will fail the same way again"))
        if retries and any(c in (429, 503) for c in l["retry_on"]) and not l["respect_retry_after"]:
            flags.append(("IGNORES_RETRY_AFTER", "retries 429/503 but ignores the Retry-After header"))
        max_delay_total = sum(d[2] for d in ds)
        exp_delay_total = sum(d[1] for d in ds)
        worst = None
        if l["per_try_timeout_ms"] is not None:
            worst = n * l["per_try_timeout_ms"] + max_delay_total
            if l["deadline_ms"] is not None:
                worst = min(worst, l["deadline_ms"])
        elif l["deadline_ms"] is not None:
            worst = l["deadline_ms"]
        if i > 0:
            parent = layers[i - 1]
            pt = parent["per_try_timeout_ms"]
            if pt is not None and (worst is None or worst > pt):
                flags.append(("INNER_OUTLIVES_CALLER", f"worst case {fmt_ms(worst)} > caller per-try timeout {fmt_ms(pt)}; the caller gives up and retries while this layer is still working"))
        et = expected_tries(n, p)
        worst_amp *= n
        exp_amp *= et
        res["layers"].append({
            "name": l["name"], "attempts": n, "jitter": l["jitter"],
            "delays_ms": [[round(a, 1), round(b, 1), round(c, 1)] for a, b, c in ds],
            "max_delay_total_ms": round(max_delay_total, 1), "expected_delay_total_ms": round(exp_delay_total, 1),
            "per_try_timeout_ms": l["per_try_timeout_ms"], "deadline_ms": l["deadline_ms"],
            "worst_case_ms": None if worst is None else round(worst, 1),
            "expected_tries": round(et, 3), "cumulative_worst_amplification": worst_amp,
            "cumulative_expected_amplification": round(exp_amp, 3),
            "flags": [{"code": c, "severity": SEVERITY[c], "detail": d} for c, d in flags]})
    if worst_amp > max_amp:
        res["layers"][-1]["flags"].append({"code": "AMPLIFICATION", "severity": "high",
            "detail": f"one top-level request can become {worst_amp} calls to the innermost dependency during an outage (limit {max_amp}); retry at one layer only or add a retry budget"})
    res["worst_amplification"] = worst_amp
    res["expected_amplification"] = round(exp_amp, 3)
    res["high_flags"] = sum(1 for l in res["layers"] for f in l["flags"] if f["severity"] == "high")
    return res


def fmt_ms(v):
    if v is None:
        return "unbounded"
    if v >= 1000:
        s = f"{v / 1000:.2f}".rstrip("0").rstrip(".")
        return s + " s"
    return f"{v:g} ms"


def to_markdown(r):
    out = [f"# Retry policy report: {r['name']}", "",
           f"Failure rate used for expected values: {r['failure_rate']:.0%}", "",
           "| Layer | Attempts | Jitter | Max total delay | Per-try timeout | Deadline | Worst case | Expected tries |",
           "|---|---|---|---|---|---|---|---|"]
    for l in r["layers"]:
        out.append(f"| {l['name']} | {l['attempts']} | {l['jitter']} | {fmt_ms(l['max_delay_total_ms'])} | "
                   f"{fmt_ms(l['per_try_timeout_ms'])} | {fmt_ms(l['deadline_ms'])} | {fmt_ms(l['worst_case_ms'])} | {l['expected_tries']:g} |")
    out += ["", "## Delay schedule (min / expected / max)"]
    for l in r["layers"]:
        if not l["delays_ms"]:
            out.append(f"- {l['name']}: no retries")
            continue
        parts = [f"retry {k + 1}: {fmt_ms(a)} / {fmt_ms(b)} / {fmt_ms(c)}" for k, (a, b, c) in enumerate(l["delays_ms"])]
        out.append(f"- {l['name']}: " + "; ".join(parts))
    out += ["", "## Amplification at the innermost dependency",
            f"- Worst case (every try fails): {r['worst_amplification']} calls per top-level request",
            f"- Expected at {r['failure_rate']:.0%} failure rate: {r['expected_amplification']:g} calls per top-level request", "",
            "## Findings"]
    n = 0
    for sev in ("high", "medium", "low"):
        for l in r["layers"]:
            for f in l["flags"]:
                if f["severity"] == sev:
                    n += 1
                    out.append(f"- **{sev.upper()} {f['code']}** ({l['name']}): {f['detail']}")
    if not n:
        out.append("- No findings.")
    return "\n".join(out)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    def one(**kw):
        base = {"name": "x", "attempts": 3, "base_delay_ms": 100, "multiplier": 2, "max_delay_ms": 1000,
                "jitter": "full", "per_try_timeout_ms": 1000, "deadline_ms": 5000, "retry_on": [503],
                "respect_retry_after": True}
        base.update(kw)
        return base

    def codes(r, i=0):
        return {f["code"] for f in r["layers"][i]["flags"]}

    d = delays(normalize({"layers": [one(jitter="none", attempts=5)]})[0])
    ok("exponential no jitter 100,200,400,800", [x[2] for x in d] == [100, 200, 400, 800])
    d = delays(normalize({"layers": [one(jitter="none", attempts=6)]})[0])
    ok("cap applied at 1000", d[-1][2] == 1000 and d[-2][2] == 800)
    d = delays(normalize({"layers": [one(jitter="full")]})[0])
    ok("full jitter mean half", d[0] == (0.0, 50.0, 100))
    d = delays(normalize({"layers": [one(jitter="equal")]})[0])
    ok("equal jitter 50..100 mean 75", d[0] == (50.0, 75.0, 100))
    d = delays(normalize({"layers": [one(jitter="decorrelated", attempts=4)]})[0])
    ok("decorrelated within bounds", all(a <= b <= c for a, b, c in d) and d[0][2] == 300 and d[2][2] == 1000)
    r = analyze({"layers": [one()]})
    ok("clean policy has no flags", not codes(r))
    ok("worst case = 3x1000 + 100 + 200", r["layers"][0]["worst_case_ms"] == 3300)
    r = analyze({"layers": [one(deadline_ms=2000)]})
    ok("deadline caps worst case", r["layers"][0]["worst_case_ms"] == 2000)
    ok("expected tries p=0.5 n=3 is 1.75", expected_tries(3, 0.5) == 1.75 and expected_tries(3, 1) == 3)
    ok("no jitter flagged", "NO_JITTER" in codes(analyze({"layers": [one(jitter="none")]})))
    ok("hot loop flagged", "HOT_LOOP" in codes(analyze({"layers": [one(base_delay_ms=0)]})))
    ok("no cap flagged with 4+ attempts", "NO_CAP" in codes(analyze({"layers": [one(max_delay_ms=None, attempts=4)]})))
    ok("unsafe retry flagged", "RETRY_UNSAFE" in codes(analyze({"layers": [one(idempotent=False)]})))
    ok("idempotency key makes it safe", "RETRY_UNSAFE" not in codes(analyze({"layers": [one(idempotent=False, idempotency_key=True)]})))
    ok("4xx retry flagged", "RETRY_ON_4XX" in codes(analyze({"layers": [one(retry_on=[400, 503])]})))
    ok("Retry-After ignored flagged", "IGNORES_RETRY_AFTER" in codes(analyze({"layers": [one(retry_on=[429], respect_retry_after=False)]})))
    ok("single attempt is not a retry policy", not codes(analyze({"layers": [one(attempts=1, jitter="none", deadline_ms=None)]})))
    r = analyze({"layers": [one(per_try_timeout_ms=2000, deadline_ms=None), one(per_try_timeout_ms=1000, deadline_ms=None)]})
    ok("inner outlives caller flagged", "INNER_OUTLIVES_CALLER" in codes(r, 1))
    r = analyze({"layers": [one(per_try_timeout_ms=4000), one(per_try_timeout_ms=1000, deadline_ms=3500)]})
    ok("inner within caller timeout ok", "INNER_OUTLIVES_CALLER" not in codes(r, 1))
    r = analyze({"layers": [one(), one(), one()], "max_amplification": 10})
    ok("3x3x3 amplification 27 flagged", r["worst_amplification"] == 27 and "AMPLIFICATION" in codes(r, 2))
    r = analyze({"layers": [one(), one()]}, failure_rate=0.5)
    ok("expected amplification 1.75^2", r["expected_amplification"] == 3.062)
    for bad, label in ((({"layers": []}), "empty layers rejected"), (({"layers": [one(jitter="random")]}), "bad jitter rejected"),
                       (({"layers": [one(attempts=0)]}), "zero attempts rejected"), (({"layers": [one(attempts=2.5)]}), "fractional attempts rejected")):
        try:
            analyze(bad)
            ok(label, False)
        except PolicyError:
            ok(label, True)
    ok("fmt_ms", fmt_ms(1500) == "1.5 s" and fmt_ms(250) == "250 ms" and fmt_ms(None) == "unbounded")
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Check retry/backoff/timeout policies across a call chain.")
    ap.add_argument("file", nargs="?", help="chain JSON ('-' for stdin)")
    ap.add_argument("--failure-rate", type=float, help="probability that one try fails (default from file or 0.2)")
    ap.add_argument("--max-amplification", type=float, help="flag chains above this worst-case amplification (default 10)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.file:
        ap.error("chain JSON required (or --self-test)")
    try:
        raw = json.load(sys.stdin if a.file == "-" else open(a.file, encoding="utf-8"))
        r = analyze(raw, a.failure_rate, a.max_amplification)
    except (OSError, PolicyError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    return 1 if r["high_flags"] else 0


if __name__ == "__main__":
    sys.exit(main())

---
name: disk-capacity-runway-forecaster
description: Forecasts when disks and volumes will fill up from usage samples (CSV of timestamp, volume, used and size). Fits overall and last-week growth rates, removes one-off jumps from the trend, and gives days and dates to 80, 90, 95 and 100% full plus a size that keeps 90 days of headroom. Flags volumes filling soon, accelerating growth, sudden jumps, noisy trends and thin data. Includes a stdlib Python forecaster with --self-test.
---

# Disk Capacity Runway Forecaster

You help on-call engineers and sysadmins answer "when will this disk fill up, and what should we do before then?" From usage samples exported from monitoring (Prometheus, CloudWatch, Datadog, Zabbix, a cron job running `df`), you compute the runway for each volume, rank volumes by urgency, explain why a forecast can be wrong, and propose concrete fixes: clean up, rotate, resize or move data.

## Files in this skill

- `scripts/disk_runway.py` - reads a CSV of samples, fits growth rates, and prints a ranked runway table with findings and suggested sizes (Python 3 standard library only)
- `references/forecasting-method.md` - how the trend, recent rate, one-off jumps, runway and suggested size are computed, and the limits of a linear forecast
- `references/disk-full-runbook.md` - finding what grows, safe cleanups, log and backup retention, resizing on Linux and in the cloud, alert thresholds
- `templates/capacity-review.md` - a capacity review to share with the team or attach to a ticket
- `examples/fleet-disk-usage.csv` - three weeks of daily samples for four volumes: a steady database, a log volume with rotation, a CI runner speeding up and a backup NAS with a one-off jump

## Workflow

### 1. Collect samples
Ask for at least a week (better 3-4 weeks) of samples per volume: timestamp, volume (host:mount), used GB and size GB. Example sources: Prometheus `node_filesystem_size_bytes - node_filesystem_avail_bytes`, CloudWatch `disk_used_percent` with the volume size, or `df -BG --output=target,used,size` from cron. Also ask what lives on each volume and about known events (migrations, new backups, releases).

### 2. Run the forecaster
```
python3 scripts/disk_runway.py examples/fleet-disk-usage.csv
python3 scripts/disk_runway.py usage.csv --horizon 180 --json
python3 scripts/disk_runway.py --self-test
```
Exit code 1 means a CRITICAL or HIGH volume, 2 means bad input.

### 3. Explain and act
Start with CRITICAL and HIGH volumes. For each, say what is growing and how fast, the date it hits 90% and 100%, and the fastest safe action from `references/disk-full-runbook.md`. Treat NOISY_TREND and FEW_SAMPLES results as rough. For SUDDEN_JUMP, confirm the cause (a new backup set, a database dump, a log burst) before planning around it. Use `references/forecasting-method.md` to explain why the recent rate may win over the long-term trend.

### 4. Deliver
Fill `templates/capacity-review.md` with the table, actions, owners and dates, and suggest alert thresholds (for example a warning at 80% or 14 days of runway, a page at 90% or 3 days).

## Example result
For `examples/fleet-disk-usage.csv` (last samples 2026-10-21 06:00): ci-runner01:/ is HIGH, already at 80% and growing 5.6 GB/day in the last week vs 2.1 overall, 90% in 3.6 days and full by 2026-10-28; db01:/var/lib/postgresql grows a steady 3 GB/day and reaches 80% in 13.3 days (MEDIUM, suggested 790 GB); backup-nas had a one-off +405.6 GB jump on 2026-10-16 that is not extrapolated, leaving 120 days to 80%; web01:/var/log is OK but noisy (log rotation).

## Rules

- Never suggest deleting data without saying exactly what, how to check it is safe, and how to recover; prefer rotation, compression, retention policies and resizing.
- Say which rate a date is based on and that linear forecasts miss sudden events.
- Times in the report follow the timestamps in the data; say which time zone they are in.
- Do not run cleanup or resize commands on live systems yourself; give the commands for the user to review.
FILE:references/forecasting-method.md
# Forecasting Method

## Inputs
One row per sample: `timestamp,volume,used_gb,size_gb`. Rows may be unsorted and mixed across volumes. The latest `size_gb` is used, so a resize during the window is handled.

## Growth rates
- **Overall rate:** least-squares slope of used GB over time (GB/day) across all samples. R^2 (0 to 1) says how well a straight line fits: about 1.0 for steady growth, under 0.5 for sawtooth patterns such as log rotation or temp files.
- **Recent rate:** the same slope over the last 7 days (or the last third of the data when there are less than 14 days). It catches growth that has just sped up.
- **Runway rate:** the worse (higher) of the two. When the recent rate is more than 1.5 times the overall rate, the volume is flagged ACCELERATING.

## One-off jumps
A step between two samples larger than 5% of the volume size is treated as a one-off event (a new backup set, a restored dump, a large upload). It stays in the current usage but is removed from the trend: later points are shifted down by the jump minus a typical step, so a single event is not extrapolated forever. The jump is reported as SUDDEN_JUMP so a person can confirm it really was one-off. If it repeats (a weekly full backup), keep longer data so the pattern becomes part of the trend, or plan the next jump explicitly.

## Runway
For each threshold T in 80, 90, 95 and 100%:
`days = (T x size - used now) / runway rate`, counted from the last sample. Zero means the threshold is already passed; "never" means usage is flat or falling.
Dates are the last sample time plus the days, in the same time zone as the data.

## Suggested size
The smallest multiple of 10 GB that keeps usage under 80% for the horizon (default 90 days):
`size >= (used now + runway rate x horizon) / 0.8`, and never smaller than the current size.
Example: 128 GB used, 2 GB/day, 180 days -> (128 + 360) / 0.8 = 610 GB.

## Status levels
| Status | Rule |
|---|---|
| CRITICAL | 95% or more used, or full within 7 days |
| HIGH | 90% within 14 days |
| MEDIUM | 80% within 30 days, or accelerating, or a sudden jump |
| OK | none of the above |

## Limits
- A straight line cannot see future events: a migration, a new customer, a log level left on DEBUG.
- Filesystems often misbehave before 100%: ext4 reserves about 5% for root, databases need free space for vacuum, compaction or index builds, and some write paths fail at 90-95%.
- Inode exhaustion (`df -i`) can fill a disk with plenty of free GB; forecast inodes the same way if many small files are written.
- Thin-provisioned and shared storage (LVM thin pools, SAN, Ceph) can fill underneath a volume that looks fine.
FILE:references/disk-full-runbook.md
# Disk Full Runbook

## 1. Find what grows
```
df -h /mount; df -i /mount                       # space and inodes
sudo du -xh --max-depth=2 /mount | sort -h | tail -20
sudo find /mount -xdev -type f -size +1G -mtime -7 -exec ls -lh {} +
sudo lsof +L1                                    # deleted files still held open
journalctl --disk-usage
docker system df                                 # images, containers, volumes, build cache
```
Deleted-but-open files keep their space until the process restarts or reopens the log.

## 2. Safe quick wins (check before deleting)
| Source | Action |
|---|---|
| Application logs | logrotate with `compress`, `maxsize`, `rotate N`; ship logs to central storage |
| systemd journal | `journalctl --vacuum-size=1G`, set `SystemMaxUse=` |
| Docker / CI runners | `docker image prune -a --filter until=168h`, `docker builder prune`; clear runner work dirs |
| Package caches | `apt-get clean`, `dnf clean all`, old kernels |
| Database | archive or drop old partitions, `VACUUM` / table repack, check WAL or binlog retention and stuck replication slots |
| Backups | apply the retention policy; move old sets to object storage or cold tier |
| Core dumps, heap dumps, tmp | remove after the incident owner confirms |
Never `rm` a file a database is using; never clear WAL or binlogs by hand.

## 3. Resize
- **AWS EBS:** modify the volume size, wait for the optimizing state, then `sudo growpart /dev/nvme0n1 1` and `sudo resize2fs /dev/nvme0n1p1` (ext4) or `sudo xfs_growfs /mount` (xfs). One size increase per 6 hours per volume.
- **Azure managed disk / GCP persistent disk:** increase the size (Azure may need the VM deallocated for some disks), then grow the partition and filesystem the same way.
- **LVM:** `lvextend -r -L +100G vg/lv` grows the filesystem too.
- **Kubernetes PVC:** edit `spec.resources.requests.storage` when the StorageClass has `allowVolumeExpansion: true`.
Volumes can grow but rarely shrink, so size to the forecast, not far beyond it.

## 4. Alerts that give time to act
- Warn at 80% used or 14 days of runway; page at 90% or 3 days of runway (Prometheus: `predict_linear(node_filesystem_avail_bytes[6h], 4*3600) < 0` style rules).
- Alert on inode use above 85%.
- Review capacity monthly with this skill's report; add growth from planned launches by hand.
FILE:templates/capacity-review.md
# Capacity Review - <service / fleet> - <date>

**Data:** <source>, <first sample> to <last sample> (<time zone>)  **Horizon:** <90> days  **Reviewer:** <name>

## Runway
| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 90% on | Full on | Suggested size |
|---|---|---|---|---|---|---|---|
| | | | | | | | |

## What is growing
| Volume | Main consumer | Evidence | One-off or ongoing |
|---|---|---|---|
| | | | |

## Actions
| Volume | Action | Owner | Due | Gain |
|---|---|---|---|---|
| | | | | |

## Alerts
- Warning: <80% or 14 days runway>  - Page: <90% or 3 days runway>  - Inodes: <85%>

## Next review
<date>; re-run the forecaster with fresh samples
FILE:examples/fleet-disk-usage.csv
timestamp,volume,used_gb,size_gb
2026-10-01T06:00,db01:/var/lib/postgresql,300.0,500
2026-10-01T06:00,web01:/var/log,21.0,100
2026-10-01T06:00,ci-runner01:/,110.0,200
2026-10-01T06:00,backup-nas:/srv/backups,2100.0,4000
2026-10-02T06:00,db01:/var/lib/postgresql,303.4,500
2026-10-02T06:00,web01:/var/log,24.4,100
2026-10-02T06:00,ci-runner01:/,111.4,200
2026-10-02T06:00,backup-nas:/srv/backups,2105.4,4000
2026-10-03T06:00,db01:/var/lib/postgresql,305.7,500
2026-10-03T06:00,web01:/var/log,26.7,100
2026-10-03T06:00,ci-runner01:/,111.7,200
2026-10-03T06:00,backup-nas:/srv/backups,2109.7,4000
2026-10-04T06:00,db01:/var/lib/postgresql,309.2,500
2026-10-04T06:00,web01:/var/log,30.2,100
2026-10-04T06:00,ci-runner01:/,113.2,200
2026-10-04T06:00,backup-nas:/srv/backups,2115.2,4000
2026-10-05T06:00,db01:/var/lib/postgresql,311.9,500
2026-10-05T06:00,web01:/var/log,32.9,100
2026-10-05T06:00,ci-runner01:/,113.9,200
2026-10-05T06:00,backup-nas:/srv/backups,2119.9,4000
2026-10-06T06:00,db01:/var/lib/postgresql,315.5,500
2026-10-06T06:00,web01:/var/log,22.5,100
2026-10-06T06:00,ci-runner01:/,115.5,200
2026-10-06T06:00,backup-nas:/srv/backups,2125.5,4000
2026-10-07T06:00,db01:/var/lib/postgresql,317.6,500
2026-10-07T06:00,web01:/var/log,24.6,100
2026-10-07T06:00,ci-runner01:/,115.6,200
2026-10-07T06:00,backup-nas:/srv/backups,2129.6,4000
2026-10-08T06:00,db01:/var/lib/postgresql,321.1,500
2026-10-08T06:00,web01:/var/log,28.1,100
2026-10-08T06:00,ci-runner01:/,117.1,200
2026-10-08T06:00,backup-nas:/srv/backups,2135.1,4000
2026-10-09T06:00,db01:/var/lib/postgresql,324.3,500
2026-10-09T06:00,web01:/var/log,31.3,100
2026-10-09T06:00,ci-runner01:/,118.3,200
2026-10-09T06:00,backup-nas:/srv/backups,2140.3,4000
2026-10-10T06:00,db01:/var/lib/postgresql,326.8,500
2026-10-10T06:00,web01:/var/log,33.8,100
2026-10-10T06:00,ci-runner01:/,118.8,200
2026-10-10T06:00,backup-nas:/srv/backups,2144.8,4000
2026-10-11T06:00,db01:/var/lib/postgresql,330.0,500
2026-10-11T06:00,web01:/var/log,23.0,100
2026-10-11T06:00,ci-runner01:/,120.0,200
2026-10-11T06:00,backup-nas:/srv/backups,2150.0,4000
2026-10-12T06:00,db01:/var/lib/postgresql,333.4,500
2026-10-12T06:00,web01:/var/log,26.4,100
2026-10-12T06:00,ci-runner01:/,121.4,200
2026-10-12T06:00,backup-nas:/srv/backups,2155.4,4000
2026-10-13T06:00,db01:/var/lib/postgresql,335.7,500
2026-10-13T06:00,web01:/var/log,28.7,100
2026-10-13T06:00,ci-runner01:/,121.7,200
2026-10-13T06:00,backup-nas:/srv/backups,2159.7,4000
2026-10-14T06:00,db01:/var/lib/postgresql,339.2,500
2026-10-14T06:00,web01:/var/log,32.2,100
2026-10-14T06:00,ci-runner01:/,123.2,200
2026-10-14T06:00,backup-nas:/srv/backups,2165.2,4000
2026-10-15T06:00,db01:/var/lib/postgresql,341.9,500
2026-10-15T06:00,web01:/var/log,34.9,100
2026-10-15T06:00,ci-runner01:/,123.9,200
2026-10-15T06:00,backup-nas:/srv/backups,2169.9,4000
2026-10-16T06:00,db01:/var/lib/postgresql,345.5,500
2026-10-16T06:00,web01:/var/log,24.5,100
2026-10-16T06:00,ci-runner01:/,130.5,200
2026-10-16T06:00,backup-nas:/srv/backups,2575.5,4000
2026-10-17T06:00,db01:/var/lib/postgresql,347.6,500
2026-10-17T06:00,web01:/var/log,26.6,100
2026-10-17T06:00,ci-runner01:/,135.6,200
2026-10-17T06:00,backup-nas:/srv/backups,2579.6,4000
2026-10-18T06:00,db01:/var/lib/postgresql,351.1,500
2026-10-18T06:00,web01:/var/log,30.1,100
2026-10-18T06:00,ci-runner01:/,142.1,200
2026-10-18T06:00,backup-nas:/srv/backups,2585.1,4000
2026-10-19T06:00,db01:/var/lib/postgresql,354.3,500
2026-10-19T06:00,web01:/var/log,33.3,100
2026-10-19T06:00,ci-runner01:/,148.3,200
2026-10-19T06:00,backup-nas:/srv/backups,2590.3,4000
2026-10-20T06:00,db01:/var/lib/postgresql,356.8,500
2026-10-20T06:00,web01:/var/log,35.8,100
2026-10-20T06:00,ci-runner01:/,153.8,200
2026-10-20T06:00,backup-nas:/srv/backups,2594.8,4000
2026-10-21T06:00,db01:/var/lib/postgresql,360.1,500
2026-10-21T06:00,web01:/var/log,25.1,100
2026-10-21T06:00,ci-runner01:/,160.1,200
2026-10-21T06:00,backup-nas:/srv/backups,2600.1,4000
FILE:scripts/disk_runway.py
#!/usr/bin/env python3
"""Disk capacity runway forecaster (Python 3 standard library only).

Input: CSV with a header row and columns
  timestamp,volume,used_gb,size_gb
timestamp is ISO 8601 ("2026-10-01T00:00", "2026-10-01 06:00" or a date); one row per sample,
any order, several volumes in one file. size_gb may change (a resize); the latest value is used.

Model per volume:
  - growth rate = least-squares slope of used_gb over time (GB/day), plus R^2 for fit quality;
    one-off steps larger than 5% of the size are kept in current usage but removed from the trend
  - recent rate = slope over the last 7 days of samples (or the last third if shorter)
  - runway to 80/90/95/100% full = (threshold x size - used now) / max(rate, recent rate),
    counted from the LAST sample (not from today), so results are reproducible
  - suggested size: smallest whole 10 GB step that keeps the volume under 80% for
    --horizon days (default 90) at the worse of the two rates
Status / findings (severity):
  CRITICAL FULL_SOON       100% within 7 days, or already >= 95% used
  HIGH     RUNWAY_SHORT    90% within 14 days
  MEDIUM   RUNWAY_WATCH    80% within 30 days
  MEDIUM   ACCELERATING    recent rate > 1.5x the overall rate (and > 0.1 GB/day)
  MEDIUM   SUDDEN_JUMP     one step grew by more than 5% of the volume size
  LOW      NOISY_TREND     R^2 < 0.5: usage swings (log rotation, temp files); trust less
  LOW      SHRINKING       usage is falling; no runway
  LOW      FEW_SAMPLES     fewer than 4 samples or under 2 days of data
Usage:
  python3 disk_runway.py usage.csv
  python3 disk_runway.py usage.csv --horizon 180 --json
  python3 disk_runway.py --self-test
Exit code: 0 no CRITICAL/HIGH, 1 CRITICAL or HIGH findings, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import sys

THRESHOLDS = (0.8, 0.9, 0.95, 1.0)


class DataError(ValueError):
    pass


def parse_time(s):
    s = s.strip().replace("Z", "+00:00")
    try:
        t = dt.datetime.fromisoformat(s)
    except ValueError:
        raise DataError(f"bad timestamp {s!r}")
    return t.replace(tzinfo=None) if t.tzinfo is None else t.astimezone(dt.timezone.utc).replace(tzinfo=None)


def read_rows(text):
    rd = csv.DictReader(io.StringIO(text))
    need = {"timestamp", "volume", "used_gb", "size_gb"}
    if not rd.fieldnames or not need <= {f.strip() for f in rd.fieldnames}:
        raise DataError(f"CSV header must contain {', '.join(sorted(need))}")
    vols = {}
    for n, row in enumerate(rd, start=2):
        row = {k.strip(): (v or "").strip() for k, v in row.items() if k}
        if not any(row.values()):
            continue
        try:
            used, size = float(row["used_gb"]), float(row["size_gb"])
        except ValueError:
            raise DataError(f"line {n}: used_gb and size_gb must be numbers")
        if size <= 0 or used < 0:
            raise DataError(f"line {n}: size_gb must be > 0 and used_gb >= 0")
        vols.setdefault(row["volume"] or "(unnamed)", []).append((parse_time(row["timestamp"]), used, size))
    if not vols:
        raise DataError("no data rows")
    return {k: sorted(v) for k, v in vols.items()}


def slope(points):
    """points: [(days, value)] -> (slope per day, r2)"""
    n = len(points)
    if n < 2:
        return 0.0, 0.0
    mx = sum(p[0] for p in points) / n
    my = sum(p[1] for p in points) / n
    sxx = sum((p[0] - mx) ** 2 for p in points)
    if sxx == 0:
        return 0.0, 0.0
    sxy = sum((p[0] - mx) * (p[1] - my) for p in points)
    b = sxy / sxx
    syy = sum((p[1] - my) ** 2 for p in points)
    r2 = 1.0 if syy == 0 else (sxy * sxy) / (sxx * syy)
    return b, r2


def analyze(samples, horizon=90):
    t0 = samples[0][0]
    last_t, used, size = samples[-1]
    # one-off steps (> 5% of size in one sample) are removed from the trend, not extrapolated
    steps = [samples[i][1] - samples[i - 1][1] for i in range(1, len(samples))]
    normal = sorted(x for x in steps if x <= 0.05 * size)
    typical = normal[len(normal) // 2] if normal else 0.0
    shift, pts = 0.0, []
    for i, (t, u, _) in enumerate(samples):
        if i and steps[i - 1] > 0.05 * size:
            shift += steps[i - 1] - typical
        pts.append(((t - t0).total_seconds() / 86400, u - shift))
    span = pts[-1][0]
    rate, r2 = slope(pts)
    cut = pts[-1][0] - 7 if span >= 14 else pts[-1][0] - span / 3
    recent_pts = [p for p in pts if p[0] >= cut]
    recent, _ = slope(recent_pts) if len(recent_pts) >= 2 else (rate, 0)
    worst = max(rate, recent)
    pct = used / size
    runway = {}
    for th in THRESHOLDS:
        left = th * size - used
        if left <= 0:
            runway[th] = (0.0, last_t)
        elif worst > 1e-9:
            d = left / worst
            runway[th] = (d, last_t + dt.timedelta(days=d))
        else:
            runway[th] = (None, None)
    need = used + max(worst, 0) * horizon
    suggest = max(size, math.ceil(need / 0.8 / 10) * 10)
    findings = []

    def f(sev, code, msg):
        findings.append({"severity": sev, "code": code, "message": msg})

    d100, d90, d80 = runway[1.0][0], runway[0.9][0], runway[0.8][0]
    if pct >= 0.95 or (d100 is not None and d100 <= 7):
        f("CRITICAL", "FULL_SOON", f"{pct:.0%} used; full in {fmt_days(d100)}")
    elif d90 is not None and d90 <= 14:
        f("HIGH", "RUNWAY_SHORT", f"90% in {fmt_days(d90)}")
    elif d80 is not None and d80 <= 30:
        f("MEDIUM", "RUNWAY_WATCH", f"80% in {fmt_days(d80)}")
    if rate > 0 and recent > 1.5 * rate and recent > 0.1:
        f("MEDIUM", "ACCELERATING", f"last-week rate {recent:.2f} GB/day vs overall {rate:.2f} GB/day")
    jumps = [(samples[i][0], samples[i][1] - samples[i - 1][1]) for i in range(1, len(samples))]
    big = [j for j in jumps if j[1] > 0.05 * size]
    if big:
        t, g = max(big, key=lambda j: j[1])
        f("MEDIUM", "SUDDEN_JUMP", f"+{g:.1f} GB in one step at {t:%Y-%m-%d %H:%M} (backup, dump, log burst?)")
    if len(pts) >= 4 and r2 < 0.5 and worst > 0:
        f("LOW", "NOISY_TREND", f"R^2 {r2:.2f}; usage swings, so treat the runway as rough")
    if worst < 0:
        f("LOW", "SHRINKING", f"usage falling {rate:.2f} GB/day")
    if len(pts) < 4 or span < 2:
        f("LOW", "FEW_SAMPLES", f"{len(pts)} samples over {span:.1f} days; collect more data")
    rank = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    findings.sort(key=lambda x: rank[x["severity"]])
    return {"last_sample": last_t.strftime("%Y-%m-%d %H:%M"), "used_gb": round(used, 2), "size_gb": size,
            "used_pct": round(pct * 100, 1), "rate_gb_day": round(rate, 3), "recent_rate_gb_day": round(recent, 3),
            "r2": round(r2, 3), "samples": len(pts), "span_days": round(span, 2),
            "runway": {f"{int(th * 100)}%": {"days": None if d is None else round(d, 1),
                                             "date": None if t is None else t.strftime("%Y-%m-%d")}
                       for th, (d, t) in runway.items()},
            "suggested_size_gb": suggest, "horizon_days": horizon, "findings": findings,
            "status": findings[0]["severity"] if findings and findings[0]["severity"] in ("CRITICAL", "HIGH", "MEDIUM") else "OK"}


def fmt_days(d):
    if d is None:
        return "never at current rate"
    if d <= 0:
        return "already"
    if d < 1:
        return f"{d * 24:.0f} h"
    return f"{d:.1f} days"


def report(res, horizon):
    out = ["# Disk runway report", "", f"Runway uses the worse of the overall and last-week growth rate; suggested size keeps usage under 80% for {horizon} days.", "",
           "| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 80% | 90% | 100% | Suggested size |",
           "|---|---|---|---|---|---|---|---|---|"]
    order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "OK": 3}
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        def cell(k):
            x = r["runway"][k]
            if x["days"] is None:
                return "never"
            return "now" if x["days"] == 0 else f"{x['days']} d ({x['date']})"
        out.append(f"| {name} | {r['status']} | {r['used_gb']:g} GB ({r['used_pct']}%) | {r['size_gb']:g} GB | "
                   f"{r['rate_gb_day']} / {r['recent_rate_gb_day']} | {cell('80%')} | {cell('90%')} | {cell('100%')} | "
                   f"{r['suggested_size_gb']:g} GB |")
    out += ["", "## Findings"]
    any_f = False
    for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
        for x in r["findings"]:
            any_f = True
            out.append(f"- **{x['severity']} {x['code']}** ({name}): {x['message']}")
    if not any_f:
        out.append("- none")
    out += ["", "Last samples: " + ", ".join(f"{n} {r['last_sample']} ({r['samples']} samples, {r['span_days']} days, R^2 {r['r2']})" for n, r in sorted(res.items()))]
    return "\n".join(out) + "\n"


def run(text, horizon=90):
    if horizon <= 0:
        raise DataError("horizon must be > 0")
    return {name: analyze(s, horizon) for name, s in read_rows(text).items()}


def self_test():
    cases = []

    def ok(name, cond):
        cases.append(bool(cond))
        print(("PASS " if cond else "FAIL ") + name)

    b, r2 = slope([(0, 10), (1, 12), (2, 14)])
    ok("slope 2/day exact fit", abs(b - 2) < 1e-9 and abs(r2 - 1) < 1e-9)
    ok("flat slope 0", slope([(0, 5), (1, 5), (2, 5)])[0] == 0)
    hdr = "timestamp,volume,used_gb,size_gb\n"
    lin = hdr + "".join(f"2026-10-{d:02d},v,{100 + 2 * d},500\n" for d in range(1, 15))
    r = run(lin)["v"]
    ok("rate 2 GB/day", abs(r["rate_gb_day"] - 2) < 1e-6)
    ok("80% runway (400-128)/2 = 136 d", r["runway"]["80%"]["days"] == 136.0)
    ok("runway date from last sample", r["runway"]["100%"]["date"] == "2027-04-18")
    ok("healthy volume status OK", r["status"] == "OK" and r["findings"] == [])
    ok("suggested size keeps 500", r["suggested_size_gb"] == 500)
    r = run(lin, horizon=180)["v"]
    ok("180 d horizon suggests 610", r["suggested_size_gb"] == 610)
    full = hdr + "".join(f"2026-10-{d:02d},v,{80 + 3 * d},100\n" for d in range(1, 6))
    r = run(full)["v"]
    ok("95% used is CRITICAL", r["status"] == "CRITICAL" and r["findings"][0]["code"] == "FULL_SOON")
    ok("past threshold runway is 0", r["runway"]["90%"]["days"] == 0)
    short = hdr + "".join(f"2026-10-{d:02d},v,{50 + 3 * d},100\n" for d in range(1, 8))
    r = run(short)["v"]
    ok("90% in ~6 days is HIGH", r["status"] == "HIGH")
    watch = hdr + "".join(f"2026-10-{d:02d},v,{40 + d},100\n" for d in range(1, 8))
    ok("80% in 33 d is OK, 80% in 26 d is MEDIUM", run(watch)["v"]["status"] == "OK" and
       run(hdr + "".join(f"2026-10-{d:02d},v,{47 + d},100\n" for d in range(1, 8)))["v"]["status"] == "MEDIUM")
    acc = hdr + "".join(f"2026-10-{d:02d},v,{100 + (d if d <= 14 else 14 + 5 * (d - 14))},1000\n" for d in range(1, 22))
    r = run(acc)["v"]
    ok("accelerating flagged", any(x["code"] == "ACCELERATING" for x in r["findings"]))
    ok("recent rate used for runway", r["recent_rate_gb_day"] > r["rate_gb_day"])
    jump = hdr + "2026-10-01,v,100,500\n2026-10-02,v,101,500\n2026-10-03,v,140,500\n2026-10-04,v,141,500\n"
    r = run(jump)["v"]
    ok("sudden jump flagged", any(x["code"] == "SUDDEN_JUMP" for x in r["findings"]))
    ok("jump removed from trend (1 GB/day)", abs(r["rate_gb_day"] - 1) < 1e-6 and r["used_gb"] == 141)
    noisy = hdr + "".join(f"2026-10-{d:02d},v,{100 + (8 if d % 2 else 0) + 0.3 * d:.1f},500\n" for d in range(1, 21))
    ok("noisy trend flagged", any(x["code"] == "NOISY_TREND" for x in run(noisy)["v"]["findings"]))
    shr = hdr + "".join(f"2026-10-{d:02d},v,{200 - 3 * d},500\n" for d in range(1, 8))
    r = run(shr)["v"]
    ok("shrinking flagged and runway never", any(x["code"] == "SHRINKING" for x in r["findings"]) and r["runway"]["100%"]["days"] is None)
    few = hdr + "2026-10-01T00:00,v,10,100\n2026-10-01T12:00,v,11,100\n"
    ok("few samples flagged", any(x["code"] == "FEW_SAMPLES" for x in run(few)["v"]["findings"]))
    multi = hdr + "2026-10-02,b,5,100\n2026-10-01,a,1,100\n2026-10-01,b,4,100\n2026-10-02,a,1,100\n"
    r = run(multi)
    ok("multiple volumes, unsorted rows", set(r) == {"a", "b"} and r["b"]["rate_gb_day"] == 1)
    ok("resize uses latest size", run(hdr + "2026-10-01,v,50,100\n2026-10-02,v,52,200\n")["v"]["size_gb"] == 200)
    ok("fmt_days hours", fmt_days(0.5) == "12 h")
    for bad, label in (("a,b\n1,2\n", "missing columns"), (hdr, "no rows"), (hdr + "x,v,1,10\n", "bad timestamp"),
                       (hdr + "2026-10-01,v,abc,10\n", "bad number"), (hdr + "2026-10-01,v,1,0\n", "zero size")):
        try:
            run(bad)
            ok(label + " rejected", False)
        except DataError:
            ok(label + " rejected", True)
    print(f"{sum(cases)}/{len(cases)} passed")
    return 0 if all(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Forecast when disks fill up from usage samples.")
    ap.add_argument("file", nargs="?", help="CSV: timestamp,volume,used_gb,size_gb")
    ap.add_argument("--horizon", type=int, default=90, help="days of runway the suggested size must cover (default 90)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.file:
        ap.error("give a CSV file or --self-test")
    try:
        res = run(open(a.file, encoding="utf-8").read(), a.horizon)
    except (OSError, DataError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(res, indent=2) if a.json else report(res, a.horizon), end="\n" if a.json else "")
    return 1 if any(r["status"] in ("CRITICAL", "HIGH") for r in res.values()) else 0


if __name__ == "__main__":
    sys.exit(main())