Two Inspect runs. Same accuracy 1.000, same stderr 0.000, same status: success. One of them scored three of ten.

"""inspect_denominator — read an Inspect .eval log and say what set the score was over.

Inspect is the only evaluation framework I have tested that records the loss: an
EvalLog carries both `results.total_samples` (the set the run was asked to cover)
and `results.completed_samples` (the set that was actually scored). Its default is
also safe: `fail_on_error=None` aborts the run on the first sample error.

The gap is one step downstream. Long agent runs cannot use that default — a single
flaky container would throw away hours of work — so real evals set a tolerance,
`fail_on_error=0.2` or similar. Above the default and below the tolerance the run
reports `status: success`, the headline metric is a mean over the survivors, and
`stderr` is computed over the survivors too. The counts stay in the log; the number
that gets quoted does not carry them.

This tool replaces the point estimate with the interval the metric could occupy
over the set that was asked for.

    python inspect_denominator.py logs/*.eval
    python inspect_denominator.py --selftest

Exit codes: 0 nothing lost | 2 samples left the set | 3 log unreadable |
4 the instrument failed its own probe. There is no code that means "probably fine".
"""

from __future__ import annotations

import glob
import sys
from dataclasses import dataclass

EXIT_OK, EXIT_FINDING, EXIT_UNREADABLE, EXIT_INSTRUMENT = 0, 2, 3, 4


@dataclass(frozen=True)
class Reading:
    task: str
    model: str
    status: str
    intended: int          # samples the run was asked to cover
    scored: int            # samples that produced a score
    metric: str
    reported: float        # the number the harness prints
    lo: float              # the metric if every lost sample were worst-case
    hi: float              # the metric if every lost sample were best-case

    @property
    def lost(self) -> int:
        return self.intended - self.scored

    @property
    def verdict(self) -> str:
        return "OVER SURVIVORS" if self.lost else "OVER THE FULL SET"


def bound(reported: float, scored: int, intended: int,
          worst: float = 0.0, best: float = 1.0) -> tuple[float, float]:
    """What the mean could be over `intended`, given `reported` over `scored`.

    A mean is a sum divided by a count. The sum over the scored samples is fixed at
    reported*scored; each of the lost samples contributes somewhere in [worst, best].
    Nothing here assumes anything about why a sample was lost.
    """
    if intended <= 0:
        raise ValueError("intended denominator must be positive")
    if scored > intended:
        raise ValueError("more samples scored than were asked for")
    total = reported * scored
    lost = intended - scored
    return ((total + lost * worst) / intended,
            (total + lost * best) / intended)


def read(path: str) -> list[Reading]:
    from inspect_ai.log import read_eval_log
    log = read_eval_log(path, header_only=True)
    if log.results is None:
        return []
    intended = log.results.total_samples
    scored = log.results.completed_samples
    out = []
    for sc in log.results.scores:
        for name, m in sc.metrics.items():
            if name == "stderr":
                continue
            lo, hi = bound(float(m.value), scored, intended)
            out.append(Reading(
                task=log.eval.task, model=log.eval.model, status=log.status,
                intended=intended, scored=scored, metric=f"{sc.name}.{name}",
                reported=float(m.value), lo=lo, hi=hi))
    return out


def render(r: Reading) -> str:
    head = (f"{r.task}  |  {r.model}  |  status={r.status}\n"
            f"  {r.metric:<28} reported {r.reported:.4f}\n"
            f"  denominator                  {r.scored} scored of {r.intended} asked for"
            f"   [{r.verdict}]")
    if not r.lost:
        return head
    return (head + "\n"
            f"  {r.lost} sample(s) left the set before scoring; the reported figure divides by "
            f"{r.scored}.\n"
            f"  over the set that was asked for, the same run is somewhere in "
            f"[{r.lo:.4f} .. {r.hi:.4f}]")


# --------------------------------------------------------------------------
# controls — these probe the instrument, not the text. A dead control is RED.
# --------------------------------------------------------------------------

def selftest() -> int:
    ok = True

    def check(label: str, got, want) -> None:
        nonlocal ok
        good = got == want
        ok = ok and good
        print(f"  {'LIVE' if good else 'DEAD'}  {label}\n        got {got!r}  want {want!r}")

    print("controls")
    # C1 the whole point: 1.0000 over 3 of 10 must not stay 1.0000
    check("C1  1.0000 scored over 3 of 10 must open to [0.3, 1.0]",
          tuple(round(x, 4) for x in bound(1.0, 3, 10)), (0.3, 1.0))
    # C2 nothing lost must collapse the interval onto the point — the gate must not
    #    cry over an honest run
    check("C2  nothing lost must leave the number alone",
          tuple(round(x, 4) for x in bound(0.42, 50, 50)), (0.42, 0.42))
    # C3 a zero must stay reachable from below
    check("C3  0.0000 over 3 of 10 must open upward only",
          tuple(round(x, 4) for x in bound(0.0, 3, 10)), (0.0, 0.7))
    # C4 the interval must contain the reported figure whenever any sample survived
    lo, hi = bound(0.8, 4, 9)
    check("C4  the reported figure must lie inside its own interval",
          lo <= 0.8 <= hi, True)
    # C5 an impossible denominator must raise, not return something plausible
    try:
        bound(1.0, 11, 10)
        check("C5  scored > intended must be refused", "returned a value", "ValueError")
    except ValueError:
        check("C5  scored > intended must be refused", "ValueError", "ValueError")
    # C6 a total of zero must be refused rather than divided by
    try:
        bound(1.0, 0, 0)
        check("C6  an empty set must be refused, not scored", "returned a value", "ValueError")
    except ValueError:
        check("C6  an empty set must be refused, not scored", "ValueError", "ValueError")
    # C7 the checker itself must be able to say DEAD. A suite that can only pass is
    #    not a suite. This runs check() against an answer that is wrong on purpose and
    #    demands the word DEAD back, without letting it condemn the run.
    import io
    import contextlib
    buf, saved = io.StringIO(), ok
    with contextlib.redirect_stdout(buf):
        check("negative control", tuple(round(x, 4) for x in bound(1.0, 3, 10)), (1.0, 1.0))
    said_dead, condemned = "DEAD" in buf.getvalue(), not ok
    ok = saved
    check("C7  the checker must report DEAD on a wrong answer, and doing so must "
          "condemn the run", (said_dead, condemned), (True, True))

    print("\nSELFTEST " + ("OK  7 controls live" if ok
                           else "FAILED - do not trust any verdict above"))
    return EXIT_OK if ok else EXIT_INSTRUMENT


def main(argv: list[str]) -> int:
    if not argv or argv[0] in ("-h", "--help"):
        print(__doc__)
        return EXIT_OK
    if argv[0] == "--selftest":
        return selftest()

    paths: list[str] = []
    for a in argv:
        paths.extend(sorted(glob.glob(a)) or [a])
    if not paths:
        print("no logs matched", file=sys.stderr)
        return EXIT_UNREADABLE

    finding = False
    for p in paths:
        try:
            readings = read(p)
        except Exception as e:                     # noqa: BLE001 — reported, never swallowed
            print(f"{p}\n  UNREADABLE  {type(e).__name__}: {e}")
            return EXIT_UNREADABLE
        if not readings:
            print(f"{p}\n  no results in this log (status only)")
            continue
        print(p)
        for r in readings:
            print(render(r))
            finding = finding or bool(r.lost)
        print()
    return EXIT_FINDING if finding else EXIT_OK


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

"""Produce two real Inspect logs: one that loses samples, one that loses none.

No API cost — everything runs against mockllm/model. The honest run is the point:
a gate that only ever fires has not been tested. Both logs are then read back by
inspect_denominator.py, which must find the first and stay silent on the second.
"""
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
os.environ.setdefault("INSPECT_LOG_DIR", os.path.join(HERE, "logs"))

from inspect_ai import Task, task, eval                      # noqa: E402
from inspect_ai.dataset import Sample                        # noqa: E402
from inspect_ai.solver import solver, TaskState, Generate    # noqa: E402
from inspect_ai.scorer import scorer, accuracy, stderr, Score, Target  # noqa: E402

N = 10


@solver
def flaky(fail_ids):
    async def solve(state: TaskState, generate: Generate) -> TaskState:
        if int(state.sample_id) in fail_ids:
            raise RuntimeError("upstream 503 — sample never ran")
        state.output.completion = "yes"
        return state
    return solve


@scorer(metrics=[accuracy(), stderr()])
def always_right():
    async def score(state: TaskState, target: Target) -> Score:
        return Score(value="C")
    return score


def build(name, fail_ids):
    @task
    def t():
        return Task(
            name=name,
            dataset=[Sample(id=i, input="q", target="yes") for i in range(1, N + 1)],
            solver=flaky(fail_ids),
            scorer=always_right(),
        )
    return t()


if __name__ == "__main__":
    # seven of ten die inside the solver; the tolerance is set above the real error
    # rate, which is what a long agent run has to do to survive one flaky container
    eval(build("shrinking", set(range(1, 8))), model="mockllm/model",
         fail_on_error=0.75, display="plain")
    # nothing dies — the negative control, on a real log rather than an assertion
    eval(build("intact", set()), model="mockllm/model",
         fail_on_error=0.75, display="plain")
    print("\nlogs written to", os.environ["INSPECT_LOG_DIR"], file=sys.stderr)