Log Distiller

---
name: log-distiller
description: >
  Turn raw structured logs into windowed metrics at ingestion — counts, error rates, and
  latency percentiles as time-series, without standing up a separate metrics pipeline. Use
  when the user has a pile of logs (file, stream, JSONL) and wants signal: "how many errors
  per minute", "p95 latency over time", "request rate by window", "chart these logs".
  SELF-BUILDING: ships no runnable code; on first use it directs the host agent to build the
  deriver in its own sandbox and prove it against a shipped test. Do NOT use for full-text
  log search (a different retrieval task).
spec_version: 1
---

# Log Distiller — self-building payload

Logs are timestamped events; metrics are aggregations over time. The host agent builds a
deriver that windows structured log records and computes per-window counts, error rates,
and latency percentiles — dashboards-ready series from a raw log pile, no metrics backend.
Ships markdown only, proven green before use. Distilled from log-to-metric-derivation
(openobserve/openobserve) and time-windowed-stream-partitioning (timeplus-io/proton).

## When to use
Summarize, chart, or alert on structured logs as metrics: rates, ratios, percentiles over
tumbling windows. Input: JSONL, list of dicts, or parsed lines with a timestamp and fields.

## How this skill runs (self-build loop)
On trigger: (1) check ./.distiller_cache/v{spec_version}/ and reuse if unchanged; (2) else
build distiller.py per the Build Brief; (3) write the Acceptance Harness verbatim and run
it — unusable until green; (4) then derive over the real logs.

## Build Brief
Construct distiller.py exposing:
derive(records: list[dict], window_s: int, ts="ts", level="level", latency="latency_ms") -> list[dict]

Each record has a numeric epoch-seconds timestamp under ts; optional level ("INFO"/"ERROR")
and numeric latency_ms. Assign each record to a tumbling window: window_start =
floor(record[ts]/window_s)*window_s. For each non-empty window compute: count; error_rate =
fraction with level=="ERROR" (0.0 if no level field) computed PER WINDOW; p50/p95 latency
via NEAREST-RANK (sort asc, rank=ceil(p/100*n), take rank-th value 1-indexed; null if no
latencies). Return {window_start,count,error_rate,p50,p95} per window, sorted asc by
window_start.

Invariants: (1) records land in correct tumbling window; (2) count exact; (3) error_rate is
per-window — an error in one window doesn't affect another's rate (load-bearing windowing-
isolation invariant); (4) p50/p95 nearest-rank match hand-computed; (5) sorted, empty
windows omitted; (6) deterministic.

## Acceptance Harness
Write verbatim to test_distiller.py and run. Pure, stdlib-only, no network.

    from distiller import derive
    RECORDS = [
        {"ts":5,"level":"INFO","latency_ms":10},
        {"ts":10,"level":"INFO","latency_ms":20},
        {"ts":20,"level":"INFO","latency_ms":30},
        {"ts":30,"level":"ERROR","latency_ms":40},
        {"ts":65,"level":"INFO","latency_ms":100},
        {"ts":70,"level":"INFO","latency_ms":200},
    ]
    def run():
        out = derive(RECORDS, 60)
        assert [w["window_start"] for w in out] == [0, 60]
        w0, w60 = out
        assert w0["count"] == 4 and w60["count"] == 2
        assert abs(w0["error_rate"] - 0.25) < 1e-9
        assert w60["error_rate"] == 0.0
        assert w0["p50"] == 20 and w0["p95"] == 40
        assert w60["p50"] == 100 and w60["p95"] == 200
        assert derive(RECORDS, 60) == out
        print(f"PASS — {len(out)} windows, error isolated to window 0, percentiles exact")
    if __name__ == "__main__": run()

Verified offline: PASS — 2 windows, error isolated to window 0, percentiles exact. A build
using a global error rate (0.167) instead of per-window fails the harness, confirming teeth.

## Notes
- Extend with sum/min/max/rate-per-second; window assignment and per-window isolation must
  stay correct.
- Streaming: emit a window once the watermark passes its end; pair with a late-data
  corrector (see Stream Reconciler) if out-of-order records matter.
- Anti-staleness: bump spec_version to rebuild against current log schemas.
- "Done" means the harness prints PASS.