engine/core/loops.py

"""loops.py — retry-loop detection on a turn's Bash calls.

A REAL loop (the only kind counted, the only kind that earns advice) =
  the same EXACT full Bash command string occurs >= 2x within a turn
  AND >= 1 of those occurrences errored.

NEVER truncate the command (the POC truncated to a 55-char prefix and invented
six loops; with the exact-string rule the fixture has ZERO). Pure code, NO model.

A separate, clearly-weaker `near_identical` hint groups by a NORMALIZED command
(whitespace collapsed). It is explicitly NOT a real loop and is NEVER counted as
one — it is only a soft signal for the narrator to *consider*, never assert.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Any

_WS_RE = re.compile(r"\s+")


@dataclass
class Loop:
    """A real retry loop within a turn."""

    command: str  # the exact, un-truncated command string
    count: int  # how many times it ran in this turn
    errored: int  # how many of those runs errored


@dataclass
class NearIdentical:
    """A clearly-weaker hint: same NORMALIZED command >= 2x. NOT a real loop."""

    normalized: str
    count: int
    errored: int


@dataclass
class TurnLoops:
    turn: int
    loops: list[Loop] = field(default_factory=list)  # real loops only
    near_identical: list[NearIdentical] = field(default_factory=list)  # weak hints


def _bash_command(tc: Any) -> str:
    inp = tc.input if isinstance(tc.input, dict) else {}
    return str(inp.get("command", "") or "")


def _normalize(cmd: str) -> str:
    return _WS_RE.sub(" ", cmd).strip()


def detect_loops(turn) -> TurnLoops:
    """Detect real loops + near-identical hints for one turn's Bash calls."""
    out = TurnLoops(turn=turn.i)

    # exact full command string → list of (errored?) occurrences
    exact: dict[str, list[bool]] = {}
    norm: dict[str, list[bool]] = {}

    for tc in turn.tools:
        if tc.name != "Bash":
            continue
        cmd = _bash_command(tc)
        if not cmd:
            continue
        err = bool(getattr(tc, "errored", False))
        exact.setdefault(cmd, []).append(err)
        norm.setdefault(_normalize(cmd), []).append(err)

    # REAL loop: exact command >= 2x AND at least one errored
    for cmd, errs in exact.items():
        if len(errs) >= 2 and any(errs):
            out.loops.append(Loop(command=cmd, count=len(errs), errored=sum(errs)))

    # WEAK hint: normalized command >= 2x (regardless of error), but only if it
    # is not already a counted exact loop. Clearly weaker; never a real loop.
    exact_norms = {_normalize(c) for c, e in exact.items() if len(e) >= 2 and any(e)}
    for ncmd, errs in norm.items():
        if len(errs) >= 2 and ncmd not in exact_norms:
            out.near_identical.append(
                NearIdentical(normalized=ncmd, count=len(errs), errored=sum(errs))
            )

    return out


def detect_all_loops(turns) -> dict[int, TurnLoops]:
    """Map turn index → TurnLoops for every turn that has any Bash signal."""
    result: dict[int, TurnLoops] = {}
    for t in turns:
        tl = detect_loops(t)
        if tl.loops or tl.near_identical:
            result[t.i] = tl
    return result