perf(perf): skip rotated logs outside the requested window (#3081)

## Description

`parse_log_files(last_n_hours=N)` reads every `proxy.log*` file in full
— line by line, applying the PERF / STAGE_TIMINGS / ROUTER regexes to
each — and only then filters records against the cutoff. The cost of a
windowed query is O(retained log history), not O(window).

`/stats` is the hot caller. `_build_stats_payload` recomputes throughput
over `last_n_hours=1.0` behind a 10s cache TTL, so anything polling the
endpoint re-reads and re-regexes the entire rotated set every 10 seconds
for an answer that lives in the tail of the newest file or two.

Rotation caps the log directory at 10 MB × 5 backups
(`proxy/helpers.py`), so this is a bounded ~60 MB rather than an
unbounded leak. But it is a fixed tax that ramps up as a user's logs
fill toward that ceiling and then stays there — on a machine that has
reached the cap it is ~0.43s of pure waste on every stats rebuild.

The fix: skip any file whose mtime predates the cutoff. The logs are
append-only, so a file untouched since before the window cannot contain
a record inside it. `--hours 0` ("all data") still reads everything.

## Type of Change

- [x] Performance improvement

## Changes Made

- `parse_log_files` prunes rotated files by mtime before opening them;
files are `stat`'d once and the value reused for the ordering
(previously `stat`'d once per file anyway, as the sort key).
- A file that rotates away between `glob` and `stat` is skipped instead
of raising `OSError`.
- New `PerfReport.log_files_skipped` so coverage reporting stays honest
— `log_files_read` on its own would silently understate how much log
exists on disk. Defaulted, so existing callers are unaffected.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

Both new tests were confirmed to fail against unpatched `main`. The
windowed one fails on behavior (`total_lines_parsed`: `assert 2 == 1`),
not merely on the new field — the assertion order is deliberate, since a
read-then-filter implementation produces the same records and only
differs in work done.

### Test Output

```text
$ uv run --frozen --extra dev pytest tests/test_cli_perf_format.py \
    tests/test_proxy_dashboard_stats_cache.py tests/test_agent_savings.py -q
59 passed, 1 skipped, 1 warning in 3.36s

$ uvx ruff check headroom/perf/analyzer.py tests/test_cli_perf_format.py
All checks passed!

$ uvx ruff format --check headroom/perf/analyzer.py tests/test_cli_perf_format.py
2 files already formatted

$ uv run --frozen --extra dev mypy headroom/perf/analyzer.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- Environment: macOS 15 (arm64), Python 3.10.18, headroom-ai at
6d2254df, against a real `~/.headroom/logs` holding 54 MB across six
rotations (`proxy.log` + `.1`–`.5`) from a proxy that had been running
for weeks.
- Exact command / steps: pointed `analyzer.LOG_DIR` at the live log
directory and timed `parse_log_files(last_n_hours=1.0)` three times,
taking the median; ran it once on this branch and once with
`headroom/perf/analyzer.py` stashed back to `main`.
- Observed result: main = 0.426s median, 6 files read, 246,819 lines
parsed. This branch = 0.141s median, 2 files read, 4 skipped, 48,147
lines parsed. 3.0x faster, 80% fewer lines parsed, identical throughput
figure. The two files still read are the live log and one rotation that
had been written inside the last hour, which is correct.
- Not tested: Windows and Linux (the mtime semantics used here are
POSIX-standard and `pathlib` handles both, but I ran only macOS). No
benchmark on a log directory below the rotation ceiling — the win there
is proportionally smaller by construction, since there is less stale
history to skip.

## Runtime Rollout Safety

- Rollout-managed feature(s): none — this is a pure read-path
optimization inside the perf log parser.
- Minimum rollout channel: n/a.
- Stable/default behavior changed: no. Windowed queries return the same
records; only the work to produce them changes. `--hours 0` is
untouched.
- Kill switch / disable path: n/a — revert the commit. There is no flag
because there is no behavior to toggle.
- Unsafe override required: no.
- Qualification impact: none.
- Rollback path: single-commit revert; `PerfReport.log_files_skipped` is
a defaulted field, so no persisted or serialized data depends on it.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines

headroom/perf/analyzer.py

@@ -15,6 +15,7 @@ import os
 import re
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timedelta
+from pathlib import Path
 
 from headroom import paths as _paths
 from headroom.pricing.litellm_pricing import resolve_litellm_model
@@ -218,6 +219,10 @@ class PerfReport:
     transform_records: list[TransformRecord] = field(default_factory=list)
     toin_records: list[ToinRecord] = field(default_factory=list)
     log_files_read: int = 0
+    # Rotated files skipped unopened because they were last written before the
+    # requested window. Reported so coverage stays honest: `log_files_read` on
+    # its own would silently understate how much log exists on disk.
+    log_files_skipped: int = 0
     total_lines_parsed: int = 0
     # Window covered by the report. `requested_hours` is what the caller
     # asked for; `oldest_kept_ts` / `newest_kept_ts` are the actual
@@ -302,7 +307,31 @@ def parse_log_files(last_n_hours: float = 168.0) -> PerfReport:
             report.newest_kept_ts = ts_str
 
     # Collect log files: proxy.log, proxy.log.1, proxy.log.2, ...
-    log_files = sorted(log_dir.glob("proxy.log*"), key=lambda p: p.stat().st_mtime)
+    #
+    # A rotated file last written before the cutoff cannot contain a record
+    # inside the window, so skip it without opening it. Without this the cost
+    # of a windowed query is O(total log history) rather than O(window):
+    # `/stats` recomputes throughput over the last hour on a 10s cache TTL, so
+    # a dashboard polling it re-read and re-regexed every byte of every
+    # rotated log, forever, for an answer that lives in the tail of the newest
+    # file. Measured on a developer machine with six rotations (54 MB).
+    #
+    # mtime is the safe discriminator: the logs are append-only, so a file
+    # untouched since before the cutoff has no line written after it. Files
+    # are stat'd once and the value reused for the sort.
+    cutoff_epoch = cutoff.timestamp() if cutoff is not None else None
+    dated_files: list[tuple[float, Path]] = []
+    for path in log_dir.glob("proxy.log*"):
+        try:
+            mtime = path.stat().st_mtime
+        except OSError:
+            # Rotated away between glob and stat — nothing to read.
+            continue
+        if cutoff_epoch is not None and mtime < cutoff_epoch:
+            report.log_files_skipped += 1
+            continue
+        dated_files.append((mtime, path))
+    log_files = [path for _, path in sorted(dated_files, key=lambda pair: pair[0])]
 
     for log_file in log_files:
         report.log_files_read += 1

tests/test_cli_perf_format.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import csv
 import io
 import json
+import os
+from datetime import datetime, timedelta
 
 import pytest
 from click.testing import CliRunner
@@ -230,6 +232,70 @@ def test_parse_perf_line_preserves_client_field(monkeypatch, tmp_path):
     assert report.perf_records[0].client == "codex"
 
 
+def _perf_line(ts: datetime, client: str) -> str:
+    return (
+        f"{ts.strftime('%Y-%m-%d %H:%M:%S')},000 - headroom.proxy - INFO - "
+        f"[hr_x] PERF model=gpt-5 msgs=3 tok_before=1000 "
+        f"tok_after=90 tok_saved=910 cache_read=0 cache_write=0 "
+        f"cache_hit_pct=0 opt_ms=12 transforms=content_router client={client}\n"
+    )
+
+
+def _write_log(path, text: str, mtime: datetime) -> None:
+    path.write_text(text)
+    stamp = mtime.timestamp()
+    os.utime(path, (stamp, stamp))
+
+
+def test_windowed_parse_skips_rotated_logs_older_than_the_cutoff(monkeypatch, tmp_path):
+    """A windowed query must cost O(window), not O(total log history).
+
+    `/stats` recomputes throughput over the last hour on a 10s cache TTL, so
+    reading every rotated log each time made the endpoint slower the longer
+    the proxy had been running.
+    """
+    log_dir = tmp_path / "logs"
+    log_dir.mkdir()
+    now = datetime.now()
+    _write_log(
+        log_dir / "proxy.log.1",
+        _perf_line(now - timedelta(days=3), "stale"),
+        now - timedelta(days=3),
+    )
+    _write_log(log_dir / "proxy.log", _perf_line(now - timedelta(minutes=5), "live"), now)
+    monkeypatch.setattr(analyzer, "LOG_DIR", log_dir)
+
+    report = analyzer.parse_log_files(last_n_hours=1.0)
+
+    assert [r.client for r in report.perf_records] == ["live"]
+    # The stale file was never opened, so its lines were never even counted.
+    # Asserted before the counters below because a read-then-filter
+    # implementation also yields the right records -- only the work differs.
+    assert report.total_lines_parsed == 1
+    assert report.log_files_read == 1
+    assert report.log_files_skipped == 1
+
+
+def test_unwindowed_parse_still_reads_every_rotated_log(monkeypatch, tmp_path):
+    """`--hours 0` means "all data" and must not prune anything."""
+    log_dir = tmp_path / "logs"
+    log_dir.mkdir()
+    now = datetime.now()
+    _write_log(
+        log_dir / "proxy.log.1",
+        _perf_line(now - timedelta(days=3), "stale"),
+        now - timedelta(days=3),
+    )
+    _write_log(log_dir / "proxy.log", _perf_line(now - timedelta(minutes=5), "live"), now)
+    monkeypatch.setattr(analyzer, "LOG_DIR", log_dir)
+
+    report = analyzer.parse_log_files(last_n_hours=0)
+
+    assert {r.client for r in report.perf_records} == {"stale", "live"}
+    assert report.log_files_skipped == 0
+    assert report.log_files_read == 2
+
+
 def test_perf_csv_by_model(runner, monkeypatch):
     _patch_report(monkeypatch, _sample_report())
     result = runner.invoke(main, ["perf", "--format", "csv"])