engine/core/rereads.py

"""rereads.py — re-read detection within a turn.

Pattern: the same Read `file_path` opened >= 3x WITHIN a single turn → (file, count).
Counted on the full path; the reported `file` is the basename for display, but the
count is over identical full paths (so two different files with the same basename
do not merge). Pure code, NO model.

On the fixture this fires exactly once: turn 12 → apply.js x4.
"""
from __future__ import annotations

import os
from collections import Counter
from dataclasses import dataclass
from typing import Any

_REREAD_MIN = 3


@dataclass
class Reread:
    turn: int
    file_path: str  # the full path that was re-read
    file: str  # basename, for display
    count: int


def _read_path(tc: Any):
    if tc.name != "Read":
        return None
    inp = tc.input if isinstance(tc.input, dict) else {}
    p = inp.get("file_path")
    return p if isinstance(p, str) and p else None


def detect_rereads(turn) -> list[Reread]:
    """Return re-read patterns (>= 3 reads of the same path) for one turn."""
    counts: Counter = Counter()
    for tc in turn.tools:
        p = _read_path(tc)
        if p is not None:
            counts[p] += 1
    out: list[Reread] = []
    for path, n in counts.items():
        if n >= _REREAD_MIN:
            out.append(
                Reread(turn=turn.i, file_path=path, file=os.path.basename(path), count=n)
            )
    # deterministic order: highest count first, then basename
    out.sort(key=lambda r: (-r.count, r.file))
    return out


def detect_all_rereads(turns) -> dict[int, list[Reread]]:
    """Map turn index → re-read patterns, only for turns where the pattern fires."""
    result: dict[int, list[Reread]] = {}
    for t in turns:
        rr = detect_rereads(t)
        if rr:
            result[t.i] = rr
    return result