6krrt/feedback.py

#!/usr/bin/env python3
"""Fold observed verification failures back into proficiency.

The eval harness measures models on a fixed 23-task benchmark. This measures
them on YOUR traffic, which is more predictive of routing quality and
accumulates for free as you work.

    python feedback.py --dry-run     # show what would change
    python feedback.py               # apply

Two sources, treated differently, because they are not the same kind of
evidence.

**Checks (`structural`, `local_llm`) contribute failures only.** A pass there
is weak: structural 'ok' means the code parsed, not that it was correct, and a
model emitting syntactically valid nonsense would score 1.0. Recording those
passes would flood `self_eval_score` with 1.0 samples and wash out the
benchmark's discrimination — the coding categories already sit at 1.00 for
every model, and this would spread that flatness everywhere. A failure is the
opposite: truncated or malformed output is definitive.

**Client outcomes contribute BOTH ways.** A client reporting success is not
the same claim as a parser reporting success — it ran the tests, or used the
answer, and knows the work worked. That is the only ground truth available
here, so it counts in both directions. It is also the only signal that
survives streaming, where a retry cannot reach.

Each failure is applied once. `applied_at` marks consumed rows so re-running
cannot penalize a model repeatedly for the same bad response.
"""

from __future__ import annotations

import argparse
import sqlite3
import sys
from collections import defaultdict

from config import load_config
from proficiency_store import add_self_eval

# Verdicts that count as observed failures. 'unverifiable' is excluded: it
# means the checker had nothing to say, which is not evidence about the model.
FAILURE_VERDICTS = ("truncated", "malformed", "failed")

# The one verdict that counts as a positive sample. A parser's 'ok' does not
# qualify — it means the code parsed. This means the client ran it.
SUCCESS_VERDICTS = ("succeeded",)

# Failures the model did not cause are excluded. The obvious case: a client
# that sets max_tokens=40 and gets a truncated answer caused that itself, and
# counting it would let any agent with a tight cap drag down whatever model it
# happened to route to. Found by forcing exactly that during testing.


def unapplied_failures(conn: sqlite3.Connection) -> list[sqlite3.Row]:
    """Rows that should move a model's score, with the sample each contributes.

    Named for what it mostly is. Successes only enter via client outcomes; a
    check passing is not evidence the answer was right.
    """
    conn.row_factory = sqlite3.Row
    scored = FAILURE_VERDICTS + SUCCESS_VERDICTS
    placeholders = ",".join("?" * len(scored))
    return conn.execute(
        f"""
        SELECT id, model_id, provider, task_category, kind, verdict, detail
          FROM verifications
         WHERE verdict IN ({placeholders})
           AND applied_at IS NULL
           AND task_category IS NOT NULL
           AND model_attributable = 1
         ORDER BY id
        """,
        scored,
    ).fetchall()


def summarize(
    rows: list[sqlite3.Row],
) -> dict[tuple[str, str, str], list[tuple[int, float]]]:
    """Group by (model, provider, category) as (row id, sample score) pairs."""
    grouped: dict[tuple[str, str, str], list[tuple[int, float]]] = defaultdict(list)
    for r in rows:
        score = 1.0 if r["verdict"] in SUCCESS_VERDICTS else 0.0
        grouped[(r["model_id"], r["provider"], r["task_category"])].append(
            (r["id"], score)
        )
    return grouped


def apply_failures(conn: sqlite3.Connection, cfg, grouped, dry_run: bool) -> int:
    applied = 0
    for (model_id, provider, category), pairs in sorted(grouped.items()):
        ids = [i for i, _ in pairs]
        scores = [sc for _, sc in pairs]
        wins = sum(1 for sc in scores if sc == 1.0)
        print(
            f"  {model_id:24s} {category:18s} "
            f"{len(scores) - wins} failure(s), {wins} success(es)"
        )
        if dry_run:
            continue
        # Folded into the running mean by proficiency.accumulate, so the effect
        # scales with the observed rate rather than replacing the benchmark.
        add_self_eval(conn, cfg, model_id, provider, category, scores)
        conn.executemany(
            "UPDATE verifications SET applied_at = datetime('now') WHERE id = ?",
            [(i,) for i in ids],
        )
        applied += len(ids)
    if not dry_run:
        conn.commit()
    return applied


def coverage(conn: sqlite3.Connection) -> None:
    """Report what verification has seen, so its usefulness stays visible."""
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        """
        SELECT kind, verdict, COUNT(*) n FROM verifications
         GROUP BY kind, verdict ORDER BY kind, verdict
        """
    ).fetchall()
    if not rows:
        print("  no verifications recorded yet")
        return
    total = sum(r["n"] for r in rows)
    print(f"  {'kind':16s}{'verdict':16s}{'n':>6}{'share':>9}")
    for r in rows:
        print(f"  {r['kind']:16s}{r['verdict']:16s}{r['n']:>6}{r['n']/total*100:>8.1f}%")
    unver = sum(r["n"] for r in rows if r["verdict"] == "unverifiable")
    if total and unver / total > 0.8:
        print(
            f"\n  NOTE: {unver/total*100:.0f}% of responses were unverifiable. "
            "Structural checking is not earning much here —\n"
            "  most traffic is prose. The local LLM check covers that, but only "
            "above the size threshold."
        )


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()

    cfg = load_config("config.yaml")
    conn = sqlite3.connect(cfg.database.path)

    print("verification coverage so far:")
    coverage(conn)
    print()

    rows = unapplied_failures(conn)
    if not rows:
        print("no unapplied signals — nothing to fold in")
        conn.close()
        return 0

    grouped = summarize(rows)
    print(f"{'would apply' if args.dry_run else 'applying'} "
          f"{len(rows)} sample(s) across {len(grouped)} (model, category) pair(s):")
    applied = apply_failures(conn, cfg, grouped, args.dry_run)
    conn.close()
    if not args.dry_run:
        print(f"\napplied {applied} sample(s)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())