feat(sleep): record judge-side latency, requests, and model identity in replay and reports

CHANGELOG.md

@@ -7,6 +7,11 @@ All notable changes to SkillOpt are documented here. This project adheres to
 ## [Unreleased]
 
 ### Added
+- **SkillOpt-Sleep judge observability** (#283): track judge-side latency,
+  request counts, tokens used, and model identity separately in task replay,
+  consolidation, and nightly sleep reports (`report.json` and `report.md`).
+  Enables visibility into rubric evaluation overhead without conflating judge
+  costs with task generation attempts.
 - **OfficeQA Parallel Search MCP** as an opt-in `env.search_mode=parallel`
   provider for `openai_chat`/`qwen_chat`. The default stays `offline` and no
   config selects Parallel automatically. Selecting it sends the generated

skillopt_sleep/backend.py

@@ -97,6 +97,11 @@ def reflect(
     def tokens_used(self) -> int:
         return 0
 
+    @property
+    def judge_model_name(self) -> str:
+        """Model identity used when this backend acts as an evaluator/judge."""
+        return getattr(self, "model", "") or getattr(self, "name", "")
+
 
 # ── Shared scoring helpers ────────────────────────────────────────────────────
 
@@ -2326,6 +2331,14 @@ def token_delta(self) -> int:
         # shared instance state, so this branch is sequential-only - see __init__.
         return max(0, self.target.tokens_used() - getattr(self, "_target_tokens_before", 0))
 
+    @property
+    def judge_model_name(self) -> str:
+        return (
+            getattr(self.optimizer, "judge_model_name", "")
+            or getattr(self.optimizer, "model", "")
+            or getattr(self.optimizer, "name", "")
+        )
+
 
 # ── Azure OpenAI backend (gpt-5.x via managed identity) ───────────────────────
 

skillopt_sleep/consolidate.py

@@ -65,6 +65,10 @@ class ConsolidationResult:
     gate_trials: List[dict] = field(default_factory=list)
     # Steps that never started because the caller's token budget was spent.
     budget_skipped: List[str] = field(default_factory=list)
+    judge_requests: int = 0
+    judge_latency_ms: float = 0.0
+    judge_tokens: int = 0
+    judge_model: str = ""
 
 
 def _split(tasks: List[TaskRecord]) -> Tuple[List[TaskRecord], List[TaskRecord], bool]:
@@ -247,6 +251,7 @@ def _budget_spent(step: str) -> bool:
             budget_skipped=budget_skipped,
         )
 
+    all_replays: List[ReplayResult] = []
     # ── baseline on the VAL slice (the gate reference) ────────────────────
     # When the gate is OFF the user has opted out of holding out a validation set
     # (the daily-use design): we accept edits greedily and judge quality only on
@@ -258,6 +263,7 @@ def _budget_spent(step: str) -> bool:
     else:
         evlog.set_phase(backend, "baseline_val")
         base_pairs = replay_batch(backend, val_tasks, skill, memory)
+        all_replays.extend(r for _, r in base_pairs)
         base_hard, base_soft = aggregate_scores(base_pairs)
         holdout_detail = _holdout_detail(base_pairs)
     base_score = select_gate_score(base_hard, base_soft, gate_metric, gate_mixed_weight)
@@ -270,6 +276,7 @@ def _budget_spent(step: str) -> bool:
     # ── reflect over TRAIN-split failures/successes ───────────────────────
     evlog.set_phase(backend, "train")
     train_pairs = replay_batch(backend, train_tasks, skill, memory)
+    all_replays.extend(r for _, r in train_pairs)
     failures = [(t, r) for (t, r) in train_pairs if r.hard < 1.0]
     successes = [(t, r) for (t, r) in train_pairs if r.hard >= 1.0]
 
@@ -312,6 +319,7 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         trial_memory = new_doc if which == "memory" else cand_memory
         evlog.set_phase(backend, f"gate_trial:{which}")
         pairs = replay_batch(backend, val_tasks, trial_skill, trial_memory)
+        all_replays.extend(r for _, r in pairs)
         h, s = aggregate_scores(pairs)
         cand_score = select_gate_score(h, s, gate_metric, gate_mixed_weight)
         task_deltas = _task_deltas(
@@ -397,6 +405,7 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         # re-evaluate failures under the (possibly improved) skill
         evlog.set_phase(backend, "train_post_skill")
         train_pairs2 = replay_batch(backend, train_tasks, cand_skill, cand_memory)
+        all_replays.extend(r for _, r in train_pairs2)
         failures2 = [(t, r) for (t, r) in train_pairs2 if r.hard < 1.0]
         successes2 = [(t, r) for (t, r) in train_pairs2 if r.hard >= 1.0]
         edits_m = backend.reflect(
@@ -427,6 +436,7 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         else:
             evlog.set_phase(backend, "final_val")
             final_pairs = replay_batch(backend, val_tasks, cand_skill, cand_memory)
+            all_replays.extend(r for _, r in final_pairs)
         final_hard, final_soft = aggregate_scores(final_pairs)
         final_score = select_gate_score(final_hard, final_soft, gate_metric, gate_mixed_weight)
         base_gate_score = select_gate_score(base_hard, base_soft, gate_metric, gate_mixed_weight)
@@ -523,6 +533,13 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
                n_rejected=len(all_rejected),
                n_unmatched=len(all_unmatched), night=night)
 
+    j_reqs = sum(1 for r in all_replays if r.judge_model and r.judge_model != "rule")
+    j_lat = sum(r.judge_latency_ms for r in all_replays)
+    j_tok = sum(r.judge_tokens for r in all_replays)
+    j_model = next((r.judge_model for r in all_replays if r.judge_model and r.judge_model != "rule"), "")
+    if not j_model:
+        j_model = next((r.judge_model for r in all_replays if r.judge_model), "")
+
     return ConsolidationResult(
         accepted=accepted,
         gate_action=action,
@@ -541,4 +558,8 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         call_error=getattr(backend, "last_call_error", "") or "",
         holdout_leaked=holdout_leaked,
         budget_skipped=budget_skipped,
+        judge_requests=j_reqs,
+        judge_latency_ms=round(j_lat, 1),
+        judge_tokens=j_tok,
+        judge_model=j_model,
     )

skillopt_sleep/cycle.py

@@ -32,7 +32,6 @@
 from skillopt_sleep.harvest_sources import harvest_for_config, normalize_transcript_source
 from skillopt_sleep.memory import ensure_skill_scaffold
 from skillopt_sleep.mine import group_tasks_by_skill_hint, mine
-from skillopt_sleep.replay import aggregate_scores, replay_batch
 from skillopt_sleep.multi_skill import (
     BUDGET_REACHED,
     SKIPPED,
@@ -42,6 +41,7 @@
     consolidate_groups,
     skill_group_reports,
 )
+from skillopt_sleep.replay import aggregate_scores, replay_batch
 from skillopt_sleep.skill_resolver import resolve_skill, skill_search_roots
 from skillopt_sleep.staging import (
     SkillProposal,
@@ -348,6 +348,11 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str:
         f"- no-regression gate: "
         f"{'enabled' if cfg.get('gate_no_regression', False) else 'disabled'}",
         f"- tokens used: {report.tokens_used}",
+        (
+            f"- judge: {report.judge_requests} request(s), "
+            f"{report.judge_latency_ms:.1f}ms, {report.judge_tokens} tokens"
+            + (f" (model: `{_markdown_text(report.judge_model)}`)" if report.judge_model else "")
+        ),
         "",
     ]
     gate_on = str(cfg.get("gate_mode", "on")).strip().lower() not in {
@@ -964,6 +969,10 @@ def budget_exhausted() -> bool:
     report.rejected_edits = result.rejected_edits
     report.unmatched_edits = result.unmatched_edits
     report.gate_trials = redact_secrets(getattr(result, "gate_trials", []))
+    report.judge_requests = getattr(result, "judge_requests", 0)
+    report.judge_latency_ms = getattr(result, "judge_latency_ms", 0.0)
+    report.judge_tokens = getattr(result, "judge_tokens", 0)
+    report.judge_model = getattr(result, "judge_model", "")
 
     # ── held-out test measure (write-only; the gate never reads it) ──────
     # consolidate() holds the test split out entirely and documents that the
@@ -982,6 +991,13 @@ def budget_exhausted() -> bool:
         final_skill = result.new_skill if result.accepted else skill
         final_memory = result.new_memory if result.accepted else memory
         test_pairs = replay_batch(backend, test_tasks, final_skill, final_memory)
+        for _, r in test_pairs:
+            if r.judge_model and r.judge_model != "rule":
+                report.judge_requests += 1
+            report.judge_latency_ms = round(report.judge_latency_ms + r.judge_latency_ms, 1)
+            report.judge_tokens += r.judge_tokens
+            if not report.judge_model and r.judge_model:
+                report.judge_model = r.judge_model
         test_hard, test_soft = aggregate_scores(test_pairs)
         ev.log("test", "held_out_score", night=night,
                n_test=len(test_tasks),
@@ -1065,6 +1081,15 @@ def budget_exhausted() -> bool:
                 outcome = skipped_groups.get(name) or consolidated_groups.get(name)
                 if outcome is not None:
                     group_outcomes[name] = outcome
+                    if outcome.result is not None:
+                        report.judge_requests += getattr(outcome.result, "judge_requests", 0)
+                        report.judge_latency_ms = round(
+                            report.judge_latency_ms + getattr(outcome.result, "judge_latency_ms", 0.0),
+                            1,
+                        )
+                        report.judge_tokens += getattr(outcome.result, "judge_tokens", 0)
+                        if not report.judge_model and getattr(outcome.result, "judge_model", ""):
+                            report.judge_model = outcome.result.judge_model
             group_rows = skill_group_reports(group_outcomes)
             # Skill hints and isolated backend failures are both untrusted
             # free text. Scrub them before either report.json or report.md is
@@ -1214,6 +1239,10 @@ def budget_exhausted() -> bool:
                n_unmatched_edits=len(report.unmatched_edits),
                tokens_used=report.tokens_used, adopted=adopted,
                max_tokens_per_night=budget.max_tokens,
-               budget_skipped=list(budget_skipped))
+               budget_skipped=list(budget_skipped),
+               judge_requests=report.judge_requests,
+               judge_latency_ms=report.judge_latency_ms,
+               judge_tokens=report.judge_tokens,
+               judge_model=report.judge_model)
 
     return CycleOutcome(report, staging_dir, adopted, adopted_paths)

skillopt_sleep/replay.py

@@ -10,6 +10,8 @@
 """
 from __future__ import annotations
 
+import os
+from concurrent.futures import ThreadPoolExecutor
 from typing import List, Tuple
 
 from skillopt_sleep.backend import Backend
@@ -67,16 +69,41 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
             tokens = (len(skill) + len(memory) + len(task.intent) + len(response)) // 4
 
     # rule judges may need the detected tool calls; score locally when possible
+    judge_t0 = time.time()
+    judge_tokens = 0
+    judge_model = ""
     if task.reference_kind == "rule" and task.judge:
         from skillopt_sleep.judges import score_rule_judge_with_feedback
         hard, soft, rationale, optimizer_feedback = score_rule_judge_with_feedback(
             task.judge, response, tools_called
         )
+        judge_latency_ms = (time.time() - judge_t0) * 1000.0
+        judge_model = "rule"
     else:
+        eval_backend = getattr(backend, "optimizer", backend)
+        eval_token_delta_fn = getattr(eval_backend, "token_delta", None)
+        tokens_before_judge = (
+            None if eval_token_delta_fn is not None else eval_backend.tokens_used()
+        )
+
         hard, soft, rationale = backend.judge(task, response)
-        # Backend judge rationales are audit evidence and may contain provider
-        # or grader implementation details. Keep the optimizer channel generic
-        # for non-rule judges unless a future typed safe-feedback API exists.
+        judge_latency_ms = (time.time() - judge_t0) * 1000.0
+
+        if eval_token_delta_fn is not None:
+            judge_tokens = eval_token_delta_fn()
+        else:
+            judge_tokens = max(
+                0, eval_backend.tokens_used() - (tokens_before_judge or 0)
+            )
+            if judge_tokens == 0 and getattr(eval_backend, "name", "") != "mock":
+                ref_len = len(task.reference or "")
+                judge_tokens = (len(task.intent) + len(response) + ref_len) // 4
+
+        judge_model = (
+            getattr(backend, "judge_model_name", "")
+            or getattr(eval_backend, "model", "")
+            or getattr(eval_backend, "name", "")
+        )
         optimizer_feedback = (
             "The response did not satisfy the task's evaluation criteria."
             if hard < 1.0
@@ -93,7 +120,10 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
                reference_kind=task.reference_kind, sample_id=sample_id,
                hard=float(hard), soft=float(soft), why=rationale or "",
                response_head=(response or "")[:400], tools_called=tools_called,
-               tokens=int(tokens), latency_ms=round(latency_ms, 1))
+               tokens=int(tokens), latency_ms=round(latency_ms, 1),
+               judge_latency_ms=round(judge_latency_ms, 1),
+               judge_tokens=int(judge_tokens),
+               judge_model=str(judge_model))
 
     return ReplayResult(
         id=task.id,
@@ -106,14 +136,13 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
         tools_called=tools_called,
         tokens=int(tokens),
         latency_ms=round(latency_ms, 1),
+        judge_latency_ms=round(judge_latency_ms, 1),
+        judge_tokens=int(judge_tokens),
+        judge_model=str(judge_model),
         optimizer_feedback=(optimizer_feedback if hard < 1.0 else ""),
     )
 
 
-import os
-from concurrent.futures import ThreadPoolExecutor
-
-
 def replay_batch(
     backend: Backend,
     tasks: List[TaskRecord],

skillopt_sleep/types.py

@@ -115,6 +115,9 @@ class ReplayResult:
     # evidence for audit/debugging; they may contain regexes or other grader
     # implementation details and must not be used as learning context.
     optimizer_feedback: str = ""
+    judge_latency_ms: float = 0.0     # wall-clock for the judge evaluation
+    judge_tokens: int = 0             # tokens consumed by judge call
+    judge_model: str = ""             # model identity used for evaluation
 
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)
@@ -194,6 +197,10 @@ class SleepReport:
     # Per-candidate held-out comparisons, including rejected intermediate
     # skill/memory trials and the final replay.
     gate_trials: List[Dict[str, Any]] = field(default_factory=list)
+    judge_requests: int = 0
+    judge_latency_ms: float = 0.0
+    judge_tokens: int = 0
+    judge_model: str = ""
 
     def to_dict(self) -> Dict[str, Any]:
         d = asdict(self)