feat(metrics): per-(task version, agent, model) trial metrics

oddish/alembic/versions/tvm_metrics_001_add_task_version_model_metrics.py

@@ -0,0 +1,126 @@
+"""Add per-(task version, agent, model) trial metrics.
+
+Sibling of ``task_version_browse_summaries`` one grain finer, so pass rate and
+trajectory length can be selected per model instead of only filtered on.
+Backfill is deliberately NOT done here -- the table is inert until
+``backfill_task_version_model_metrics`` populates it, which keeps this
+migration fast on a hot ``trials`` table.
+
+Revision ID: tvm_metrics_001
+Revises: agentcap01
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "tvm_metrics_001"
+down_revision: Union[str, Sequence[str], None] = "agentcap01"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_TABLE = "task_version_model_metrics"
+
+
+def _table_exists(name: str) -> bool:
+    return name in sa.inspect(op.get_bind()).get_table_names()
+
+
+def _index_exists(table: str, name: str) -> bool:
+    return name in {
+        item["name"] for item in sa.inspect(op.get_bind()).get_indexes(table)
+    }
+
+
+def _counter(name: str) -> sa.Column:
+    return sa.Column(name, sa.Integer(), nullable=False, server_default="0")
+
+
+def upgrade() -> None:
+    if not _table_exists(_TABLE):
+        op.create_table(
+            _TABLE,
+            sa.Column("task_version_id", sa.String(160), nullable=False),
+            sa.Column("agent", sa.String(128), nullable=False),
+            # NULL model is a real state on older trials; the empty string keeps
+            # it inside the primary key without inventing a fake model name.
+            sa.Column("model", sa.String(256), nullable=False, server_default=""),
+            sa.Column("task_id", sa.String(128), nullable=False),
+            # scored population
+            _counter("n_pass"),
+            _counter("n_partial"),
+            _counter("n_fail"),
+            # terminal but unscored, attributed by harbor_stage
+            _counter("n_unscored_agent"),
+            _counter("n_unscored_env"),
+            _counter("n_unscored_verify"),
+            _counter("n_cancelled_user"),
+            _counter("n_cancelled_reaped"),
+            _counter("n_cancelled_other"),
+            _counter("n_skipped"),
+            _counter("n_scoreless"),
+            _counter("n_unscored_unknown"),
+            # never in any denominator
+            _counter("n_inflight"),
+            # metric sums
+            sa.Column(
+                "sum_reward", sa.Float(), nullable=False, server_default="0"
+            ),
+            _counter("n_reward_present"),
+            sa.Column(
+                "sum_runtime", sa.Float(), nullable=False, server_default="0"
+            ),
+            _counter("n_runtime_present"),
+            _counter("n_with_trajectory"),
+            # step distribution -- nullable: NULL means "never measured", which
+            # is distinct from a measured zero. See spec 4.6.
+            _counter("n_steps_present"),
+            sa.Column("sum_steps", sa.BigInteger(), nullable=False, server_default="0"),
+            sa.Column("steps_all_min", sa.Integer()),
+            sa.Column("steps_all_p50", sa.Integer()),
+            sa.Column("steps_all_max", sa.Integer()),
+            _counter("steps_pass_n"),
+            sa.Column("steps_pass_min", sa.Integer()),
+            sa.Column("steps_pass_p50", sa.Integer()),
+            sa.Column("steps_pass_max", sa.Integer()),
+            _counter("steps_fail_n"),
+            sa.Column("steps_fail_min", sa.Integer()),
+            sa.Column("steps_fail_p50", sa.Integer()),
+            sa.Column("steps_fail_max", sa.Integer()),
+            _counter("steps_partial_n"),
+            sa.Column(
+                "updated_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.text("NOW()"),
+            ),
+            sa.PrimaryKeyConstraint("task_version_id", "agent", "model"),
+            sa.ForeignKeyConstraint(
+                ["task_version_id"], ["task_versions.id"], ondelete="CASCADE"
+            ),
+            sa.ForeignKeyConstraint(["task_id"], ["tasks.id"], ondelete="CASCADE"),
+        )
+
+    # Ratios are computed at read time rather than stored: Postgres generated
+    # columns cannot reference other generated columns, and every denominator
+    # here is a sum of counters that a partial index can cover just as well.
+    if not _index_exists(_TABLE, "idx_tvm_metrics_model"):
+        op.create_index("idx_tvm_metrics_model", _TABLE, ["model", "agent"])
+    if not _index_exists(_TABLE, "idx_tvm_metrics_task_id"):
+        op.create_index("idx_tvm_metrics_task_id", _TABLE, ["task_id"])
+    # Anomaly triage (spec 4.7) seeks on a low passing-step minimum.
+    if not _index_exists(_TABLE, "idx_tvm_metrics_steps_pass_min"):
+        op.create_index(
+            "idx_tvm_metrics_steps_pass_min",
+            _TABLE,
+            ["steps_pass_min"],
+            postgresql_where=sa.text("steps_pass_n > 0"),
+        )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_tvm_metrics_steps_pass_min", table_name=_TABLE)
+    op.drop_index("idx_tvm_metrics_task_id", table_name=_TABLE)
+    op.drop_index("idx_tvm_metrics_model", table_name=_TABLE)
+    op.drop_table(_TABLE)

oddish/src/oddish/core/backfill_task_version_model_metrics.py

@@ -0,0 +1,103 @@
+"""Populate task_version_model_metrics for task versions that have no row yet.
+
+Batched and resumable: each batch commits on its own and the next run picks up
+where this one stopped, because progress is the presence of rows rather than a
+stored cursor. Safe to re-run -- a recomputed group produces an identical row.
+
+    uv run python -m oddish.core.backfill_task_version_model_metrics --limit 500
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+
+from sqlalchemy import func, select
+
+from oddish.core.task_version_model_metrics import (
+    metrics_query,
+    refresh_task_version_model_metrics,
+)
+from oddish.db import TaskVersionModel, TaskVersionModelMetricsModel, get_session
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_BATCH = 200
+
+
+async def _next_version_ids(session, after_id: str | None, batch: int) -> list[str]:
+    """Next page of task versions in id order.
+
+    Keyset, not "versions missing a row": a version whose trials are all out of
+    scope legitimately produces no rows, so a missing-row cursor would hand back
+    the same page forever.
+    """
+    query = select(TaskVersionModel.id).order_by(TaskVersionModel.id.asc())
+    if after_id is not None:
+        query = query.where(TaskVersionModel.id > after_id)
+    rows = await session.execute(query.limit(batch))
+    return [str(value) for value in rows.scalars().all()]
+
+
+async def backfill(
+    batch: int = DEFAULT_BATCH,
+    limit: int | None = None,
+    after_id: str | None = None,
+) -> int:
+    """Recompute metrics for every task version. Returns the number processed."""
+    processed = 0
+    cursor = after_id
+    while limit is None or processed < limit:
+        size = batch if limit is None else min(batch, limit - processed)
+        async with get_session() as session:
+            version_ids = await _next_version_ids(session, cursor, size)
+            if not version_ids:
+                break
+            await refresh_task_version_model_metrics(session, version_ids)
+            written = (
+                await session.execute(
+                    select(func.count())
+                    .select_from(TaskVersionModelMetricsModel)
+                    .where(
+                        TaskVersionModelMetricsModel.task_version_id.in_(version_ids)
+                    )
+                )
+            ).scalar_one()
+            await session.commit()
+        cursor = version_ids[-1]
+        processed += len(version_ids)
+        logger.info(
+            "%d task versions (%d metric rows); %d total; resume after %s",
+            len(version_ids),
+            written,
+            processed,
+            cursor,
+        )
+        if len(version_ids) < size:
+            break
+    return processed
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--batch", type=int, default=DEFAULT_BATCH)
+    parser.add_argument(
+        "--limit", type=int, default=None, help="stop after N task versions"
+    )
+    parser.add_argument(
+        "--after-id", default=None, help="resume after this task version id"
+    )
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    total = asyncio.run(
+        backfill(batch=args.batch, limit=args.limit, after_id=args.after_id)
+    )
+    logger.info("done: %d task versions", total)
+
+
+if __name__ == "__main__":
+    main()
+
+
+__all__ = ["backfill", "metrics_query"]

oddish/src/oddish/core/task_browse_summary.py

@@ -11,6 +11,7 @@
     browse_trial_scope,
     trial_bucket_label,
 )
+from oddish.core.task_version_model_metrics import refresh_task_version_model_metrics
 from oddish.db import (
     TaskBrowseSummaryModel,
     TaskVersionModel,
@@ -174,3 +175,8 @@ async def refresh_task_browse_summaries(
             | {"updated_at": func.now()},
         )
     )
+
+    # The per-model rollup is invalidated by exactly the same events, so it
+    # rides this refresh rather than duplicating the hook at all 18 call sites.
+    # Runs under the advisory locks taken above.
+    await refresh_task_version_model_metrics(session, list(summaries))