feat: add concurrency and token/cost telemetry for llm relabel runs

datasets/AnimeName

@@ -1 +1 @@
-Subproject commit 9987cc8d7b7bf829d0022ee6e6a0b08de5327975
+Subproject commit 5de6ddeed7dafd43207953072a9e197f13b32077

reports/llm_relabel_perf_char_chunk2.json

@@ -0,0 +1,10 @@
+{
+  "wall_seconds": 586.9843375682831,
+  "rows_processed": 1000,
+  "rows_per_second": 1.703622969128493,
+  "batches_completed": 250,
+  "avg_batch_seconds": 15.12994603919983,
+  "avg_active_workers": 6.443932279510124,
+  "max_active_workers": 8,
+  "configured_workers": 8
+}

reports/llm_relabel_perf_char_chunk3_tokens.json

@@ -0,0 +1,20 @@
+{
+  "wall_seconds": 739.8576474189758,
+  "rows_processed": 1000,
+  "rows_per_second": 1.351611358601944,
+  "batches_completed": 250,
+  "avg_batch_seconds": 20.91811747932434,
+  "avg_active_workers": 7.068292213849803,
+  "max_active_workers": 8,
+  "configured_workers": 8,
+  "input_tokens": 406372,
+  "output_tokens": 351059,
+  "cached_tokens": 52096,
+  "reasoning_tokens": 220306,
+  "input_tokens_per_sec": 549.2570110177892,
+  "output_tokens_per_sec": 474.49533193943984,
+  "input_tokens_per_hour": 1977325.239664041,
+  "output_tokens_per_hour": 1708183.1949819834,
+  "usd_per_1m_input": 0.0,
+  "usd_per_1m_output": 0.0
+}

reports/llm_relabel_perf_weak_chunk2.json

@@ -0,0 +1,10 @@
+{
+  "wall_seconds": 436.6158034801483,
+  "rows_processed": 1000,
+  "rows_per_second": 2.2903431163720285,
+  "batches_completed": 250,
+  "avg_batch_seconds": 13.27953750705719,
+  "avg_active_workers": 7.603676611459359,
+  "max_active_workers": 8,
+  "configured_workers": 8
+}

reports/llm_relabel_perf_weak_chunk3_tokens.json

@@ -0,0 +1,20 @@
+{
+  "wall_seconds": 315.9569420814514,
+  "rows_processed": 1000,
+  "rows_per_second": 3.164988220901971,
+  "batches_completed": 250,
+  "avg_batch_seconds": 9.494983646392821,
+  "avg_active_workers": 7.51287982740815,
+  "max_active_workers": 8,
+  "configured_workers": 8,
+  "input_tokens": 271438,
+  "output_tokens": 173591,
+  "cached_tokens": 0,
+  "reasoning_tokens": 109518,
+  "input_tokens_per_sec": 859.0980727051892,
+  "output_tokens_per_sec": 549.413470254594,
+  "input_tokens_per_hour": 3092753.061738681,
+  "output_tokens_per_hour": 1977888.4929165384,
+  "usd_per_1m_input": 0.0,
+  "usd_per_1m_output": 0.0
+}

reports/llm_relabel_perf_weak_smoke.json

@@ -0,0 +1,10 @@
+{
+  "wall_seconds": 134.8101508617401,
+  "rows_processed": 40,
+  "rows_per_second": 0.29671356158501433,
+  "batches_completed": 10,
+  "avg_batch_seconds": 20.668276357650758,
+  "avg_active_workers": 1.5331398703993708,
+  "max_active_workers": 8,
+  "configured_workers": 8
+}

reports/llm_relabel_perf_weak_smoke_with_tokens.json

@@ -0,0 +1,20 @@
+{
+  "wall_seconds": 36.052828311920166,
+  "rows_processed": 40,
+  "rows_per_second": 1.1094829968381366,
+  "batches_completed": 10,
+  "avg_batch_seconds": 11.140262508392334,
+  "avg_active_workers": 3.0899838841684244,
+  "max_active_workers": 8,
+  "configured_workers": 8,
+  "input_tokens": 9621,
+  "output_tokens": 6818,
+  "cached_tokens": 0,
+  "reasoning_tokens": 4760,
+  "input_tokens_per_sec": 266.8583978144928,
+  "output_tokens_per_sec": 189.11137681106038,
+  "input_tokens_per_hour": 960690.232132174,
+  "output_tokens_per_hour": 680800.9565198174,
+  "usd_per_1m_input": 0.0,
+  "usd_per_1m_output": 0.0
+}

tools/llm_relabel_rows.py

@@ -13,6 +13,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 import json
 import os
 import re
+import threading
 import time
 from dataclasses import dataclass
 from pathlib import Path
@@ -67,6 +68,61 @@ class Row:
     record: Dict[str, Any]
 
 
+class ConcurrentMeter:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self.current_active = 0
+        self.max_active = 0
+        self.active_time_accum = 0.0
+        self.last_ts = time.time()
+
+    def _accumulate(self, now: float) -> None:
+        dt = now - self.last_ts
+        if dt > 0:
+            self.active_time_accum += self.current_active * dt
+            self.last_ts = now
+
+    def task_start(self) -> None:
+        now = time.time()
+        with self._lock:
+            self._accumulate(now)
+            self.current_active += 1
+            if self.current_active > self.max_active:
+                self.max_active = self.current_active
+
+    def task_end(self) -> None:
+        now = time.time()
+        with self._lock:
+            self._accumulate(now)
+            if self.current_active > 0:
+                self.current_active -= 1
+
+    def snapshot(self) -> Dict[str, float]:
+        now = time.time()
+        with self._lock:
+            self._accumulate(now)
+            return {
+                "current_active": float(self.current_active),
+                "max_active": float(self.max_active),
+                "active_time_accum": float(self.active_time_accum),
+                "timestamp": now,
+            }
+
+
+@dataclass
+class UsageStats:
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cached_tokens: int = 0
+    reasoning_tokens: int = 0
+
+    def add(self, other: "UsageStats") -> None:
+        self.input_tokens += int(other.input_tokens)
+        self.output_tokens += int(other.output_tokens)
+        self.cached_tokens += int(other.cached_tokens)
+        self.reasoning_tokens += int(other.reasoning_tokens)
+
+
 def parse_args() -> argparse.Namespace:
     p = argparse.ArgumentParser(description="Relabel selected JSONL rows via Responses API")
     p.add_argument("--input", required=True, help="Input JSONL")
@@ -84,6 +140,9 @@ def parse_args() -> argparse.Namespace:
     p.add_argument("--concurrency", type=int, default=4, help="Parallel request workers")
     p.add_argument("--max-rows", type=int, default=0, help="Optional cap; 0 means no cap")
     p.add_argument("--skip-selected", type=int, default=0, help="Skip this many selected rows before processing")
+    p.add_argument("--min-token-len", type=int, default=0, help="Only process rows with token length >= this value")
+    p.add_argument("--max-token-len", type=int, default=0, help="Only process rows with token length <= this value (0 = no limit)")
+    p.add_argument("--sort-by", choices=("none", "token_len_asc"), default="none", help="Optional ordering of selected rows")
     p.add_argument("--retries", type=int, default=3, help="Retries per batch")
     p.add_argument("--sleep-ms", type=int, default=150, help="Delay between successful calls")
     p.add_argument("--prompt-cache-key", default="anifilebert-relabel-v1", help="Stable prompt cache key")
@@ -91,6 +150,9 @@ def parse_args() -> argparse.Namespace:
     p.add_argument("--reasoning-effort", default="medium", help="Reasoning effort (e.g. low/medium/high)")
     p.add_argument("--checkpoint-rows", type=int, default=100, help="Write checkpoint every N processed rows")
     p.add_argument("--failure-log", default="reports/llm_relabel_failures.log", help="Failure log path")
+    p.add_argument("--perf-log", default="", help="Optional JSON perf summary path")
+    p.add_argument("--usd-per-1m-input", type=float, default=0.0, help="Input token price (USD per 1M tokens)")
+    p.add_argument("--usd-per-1m-output", type=float, default=0.0, help="Output token price (USD per 1M tokens)")
     p.add_argument(
         "--user-agent",
         default="Codex Desktop/0.133.0-alpha.1 (Windows 10.0.22631; x86_64) unknown (Codex Desktop; 26.519.41501)",
@@ -214,6 +276,18 @@ def append_failure_log(path: str, message: str) -> None:
         f.write(message.rstrip() + "\n")
 
 
+def parse_usage(response_obj: Dict[str, Any]) -> UsageStats:
+    usage = response_obj.get("usage", {}) or {}
+    in_details = usage.get("input_tokens_details", {}) or {}
+    out_details = usage.get("output_tokens_details", {}) or {}
+    return UsageStats(
+        input_tokens=int(usage.get("input_tokens", 0) or 0),
+        output_tokens=int(usage.get("output_tokens", 0) or 0),
+        cached_tokens=int(in_details.get("cached_tokens", 0) or 0),
+        reasoning_tokens=int(out_details.get("reasoning_tokens", 0) or 0),
+    )
+
+
 def relabel_batch(
     api_base: str,
     api_key: str,
@@ -225,7 +299,7 @@ def relabel_batch(
     user_agent: str,
     retries: int,
     failure_log: str,
-) -> Dict[int, List[str]]:
+) -> tuple[Dict[int, List[str]], UsageStats]:
     url = f"{api_base.rstrip('/')}/responses"
     headers = {
         "Authorization": f"Bearer {api_key}",
@@ -259,6 +333,7 @@ def relabel_batch(
             resp = requests.post(url, headers=headers, json=body, timeout=120)
             resp.raise_for_status()
             obj = resp.json()
+            usage_stats = parse_usage(obj)
             try:
                 parsed = extract_function_args(obj, "submit_labels")
             except Exception:
@@ -300,7 +375,7 @@ def relabel_batch(
                 )
                 raise ValueError(f"incomplete/invalid rows from model: missing={missing}")
 
-            return mapping
+            return mapping, usage_stats
         except Exception as exc:  # noqa: BLE001
             last_error = exc
             # Some compatible gateways may not support prompt caching or reasoning fields.
@@ -336,8 +411,9 @@ def process_batch_with_fallback(
     retries: int,
     failure_log: str,
 ) -> List[tuple[Row, List[str]]]:
+    usage_total = UsageStats()
     try:
-        mapping = relabel_batch(
+        mapping, usage = relabel_batch(
             api_base=api_base,
             api_key=api_key,
             model=model,
@@ -349,11 +425,12 @@ def process_batch_with_fallback(
             retries=retries,
             failure_log=failure_log,
         )
+        usage_total.add(usage)
     except RuntimeError:
         mapping = {}
         for idx, row in enumerate(batch):
             try:
-                single = relabel_batch(
+                single, usage = relabel_batch(
                     api_base=api_base,
                     api_key=api_key,
                     model=model,
@@ -365,6 +442,7 @@ def process_batch_with_fallback(
                     retries=max(retries, 4),
                     failure_log=failure_log,
                 )
+                usage_total.add(usage)
                 mapping[idx] = single[0]
             except RuntimeError as exc:
                 append_failure_log(
@@ -372,7 +450,45 @@ def process_batch_with_fallback(
                     f"[row-skip] file_id={row.record.get('file_id')} line={row.line_no} reason={exc}",
                 )
                 mapping[idx] = row.record.get("labels", [])
-    return [(batch[row_id], labels) for row_id, labels in mapping.items()]
+    return [(batch[row_id], labels) for row_id, labels in mapping.items()], usage_total
+
+
+def process_batch_timed(
+    meter: ConcurrentMeter,
+    api_base: str,
+    api_key: str,
+    model: str,
+    batch: Sequence[Row],
+    prompt_cache_key: str,
+    prompt_cache_retention: str,
+    reasoning_effort: str,
+    user_agent: str,
+    retries: int,
+    failure_log: str,
+) -> Dict[str, Any]:
+    meter.task_start()
+    t0 = time.time()
+    try:
+        updates, usage = process_batch_with_fallback(
+            api_base=api_base,
+            api_key=api_key,
+            model=model,
+            batch=batch,
+            prompt_cache_key=prompt_cache_key,
+            prompt_cache_retention=prompt_cache_retention,
+            reasoning_effort=reasoning_effort,
+            user_agent=user_agent,
+            retries=retries,
+            failure_log=failure_log,
+        )
+        return {
+            "updates": updates,
+            "elapsed": time.time() - t0,
+            "batch_size": len(batch),
+            "usage": usage,
+        }
+    finally:
+        meter.task_end()
 
 
 def main() -> None:
@@ -385,6 +501,18 @@ def main() -> None:
     output_path = Path(args.output)
 
     all_records, selected_rows = load_rows(input_path, args.selector)
+    if args.min_token_len > 0 or args.max_token_len > 0:
+        filtered: List[Row] = []
+        for row in selected_rows:
+            tok_len = len(row.record.get("tokens", []))
+            if tok_len < args.min_token_len:
+                continue
+            if args.max_token_len > 0 and tok_len > args.max_token_len:
+                continue
+            filtered.append(row)
+        selected_rows = filtered
+    if args.sort_by == "token_len_asc":
+        selected_rows.sort(key=lambda r: len(r.record.get("tokens", [])))
     if args.skip_selected > 0:
         selected_rows = selected_rows[args.skip_selected:]
     if args.max_rows > 0:
@@ -404,10 +532,16 @@ def main() -> None:
     ]
 
     done_rows = 0
+    wall_start = time.time()
+    meter = ConcurrentMeter()
+    total_batch_elapsed = 0.0
+    completed_batches = 0
+    usage_total = UsageStats()
     with ThreadPoolExecutor(max_workers=concurrency) as executor:
         futures = [
             executor.submit(
-                process_batch_with_fallback,
+                process_batch_timed,
+                meter,
                 api_base=args.api_base,
                 api_key=api_key,
                 model=args.model,
@@ -422,14 +556,35 @@ def main() -> None:
             for batch in batches
         ]
         for fut in as_completed(futures):
-            updates = fut.result()
+            result = fut.result()
+            updates = result["updates"]
+            total_batch_elapsed += float(result["elapsed"])
+            completed_batches += 1
+            usage_total.add(result["usage"])
             for row, new_labels in updates:
                 rec = row.record
                 if rec.get("labels") != new_labels:
                     rec["labels"] = new_labels
                     changed += 1
             done_rows += len(updates)
-            print(f"processed={done_rows}/{total} changed={changed}")
+            snap = meter.snapshot()
+            wall_elapsed = max(1e-9, snap["timestamp"] - wall_start)
+            rows_per_sec = done_rows / wall_elapsed
+            avg_active = snap["active_time_accum"] / wall_elapsed
+            in_tok_per_sec = usage_total.input_tokens / wall_elapsed
+            out_tok_per_sec = usage_total.output_tokens / wall_elapsed
+            hourly_usd = 0.0
+            if args.usd_per_1m_input > 0 or args.usd_per_1m_output > 0:
+                cost = (usage_total.input_tokens / 1_000_000.0) * args.usd_per_1m_input + (
+                    usage_total.output_tokens / 1_000_000.0
+                ) * args.usd_per_1m_output
+                hourly_usd = cost / wall_elapsed * 3600.0
+            print(
+                f"processed={done_rows}/{total} changed={changed} "
+                f"rows_per_sec={rows_per_sec:.2f} active_now={int(snap['current_active'])} "
+                f"avg_active={avg_active:.2f} max_active={int(snap['max_active'])}/{concurrency} "
+                f"in_tok_s={in_tok_per_sec:.1f} out_tok_s={out_tok_per_sec:.1f} usd_h={hourly_usd:.3f}"
+            )
             if args.checkpoint_rows > 0 and (done_rows % args.checkpoint_rows == 0 or done_rows == total):
                 write_jsonl(output_path, all_records)
             if args.sleep_ms > 0:
@@ -437,6 +592,44 @@ def main() -> None:
 
     # rows in selected_rows reference dicts in all_records by identity, so changes are already reflected.
     write_jsonl(output_path, all_records)
+    wall_total = time.time() - wall_start
+    final_snap = meter.snapshot()
+    avg_active = final_snap["active_time_accum"] / max(1e-9, wall_total)
+    perf_summary = {
+        "wall_seconds": wall_total,
+        "rows_processed": done_rows,
+        "rows_per_second": done_rows / max(1e-9, wall_total),
+        "batches_completed": completed_batches,
+        "avg_batch_seconds": total_batch_elapsed / max(1, completed_batches),
+        "avg_active_workers": avg_active,
+        "max_active_workers": int(final_snap["max_active"]),
+        "configured_workers": concurrency,
+        "input_tokens": usage_total.input_tokens,
+        "output_tokens": usage_total.output_tokens,
+        "cached_tokens": usage_total.cached_tokens,
+        "reasoning_tokens": usage_total.reasoning_tokens,
+        "input_tokens_per_sec": usage_total.input_tokens / max(1e-9, wall_total),
+        "output_tokens_per_sec": usage_total.output_tokens / max(1e-9, wall_total),
+        "input_tokens_per_hour": usage_total.input_tokens / max(1e-9, wall_total) * 3600.0,
+        "output_tokens_per_hour": usage_total.output_tokens / max(1e-9, wall_total) * 3600.0,
+        "usd_per_1m_input": args.usd_per_1m_input,
+        "usd_per_1m_output": args.usd_per_1m_output,
+    }
+    if args.usd_per_1m_input > 0 or args.usd_per_1m_output > 0:
+        total_cost = (usage_total.input_tokens / 1_000_000.0) * args.usd_per_1m_input + (
+            usage_total.output_tokens / 1_000_000.0
+        ) * args.usd_per_1m_output
+        perf_summary["usd_total"] = total_cost
+        perf_summary["usd_per_hour"] = total_cost / max(1e-9, wall_total) * 3600.0
+    if args.perf_log:
+        p = Path(args.perf_log)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(json.dumps(perf_summary, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(
+        f"perf wall={wall_total:.1f}s rows_per_sec={perf_summary['rows_per_second']:.2f} "
+        f"avg_active={avg_active:.2f} max_active={int(final_snap['max_active'])}/{concurrency} "
+        f"in_tok_s={perf_summary['input_tokens_per_sec']:.1f} out_tok_s={perf_summary['output_tokens_per_sec']:.1f}"
+    )
     print(f"done selected_rows={total} changed_rows={changed} output={output_path}")