diff --git a/CHANGELOG.md b/CHANGELOG.md index 35205bcc..89287369 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py index 429802db..7e413379 100644 --- a/skillopt_sleep/backend.py +++ b/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) ─────────────────────── diff --git a/skillopt_sleep/consolidate.py b/skillopt_sleep/consolidate.py index 3b63520a..f239d453 100644 --- a/skillopt_sleep/consolidate.py +++ b/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, ) diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index d19f9893..48f84312 100644 --- a/skillopt_sleep/cycle.py +++ b/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) diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py index 6eba0b86..c87adbbb 100644 --- a/skillopt_sleep/replay.py +++ b/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], diff --git a/skillopt_sleep/types.py b/skillopt_sleep/types.py index 150a4f58..3a08cbdc 100644 --- a/skillopt_sleep/types.py +++ b/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) diff --git a/tests/test_sleep_judge_observability.py b/tests/test_sleep_judge_observability.py new file mode 100644 index 00000000..ca1d7cd6 --- /dev/null +++ b/tests/test_sleep_judge_observability.py @@ -0,0 +1,161 @@ +"""Tests for judge-side observability in replay, consolidate, and cycle reports (#283). + +Pure-stdlib (unittest), deterministic, no API key, no external dependencies. +""" +from __future__ import annotations + +import tempfile +import time +import unittest +from typing import Tuple + +from skillopt_sleep.backend import DualBackend, MockBackend +from skillopt_sleep.config import load_config +from skillopt_sleep.consolidate import consolidate +from skillopt_sleep.cycle import _render_report_md, run_sleep_cycle +from skillopt_sleep.replay import replay_one +from skillopt_sleep.types import SleepReport, TaskRecord + + +class InstrumentedJudgeBackend(MockBackend): + """Mock backend that tracks judge token counts and latency.""" + + def __init__(self, model_name: str = "mock-judge-v1"): + super().__init__() + self.name = model_name + self._tokens = 0 + + def tokens_used(self) -> int: + return self._tokens + + @property + def judge_model_name(self) -> str: + return self.name + + def judge(self, task: TaskRecord, response: str) -> Tuple[float, float, str]: + # Simulate LLM judge token usage and small latency + self._tokens += 42 + time.sleep(0.005) + return super().judge(task, response) + + +class TestJudgeObservability(unittest.TestCase): + def test_replay_one_rule_judge_records_rule_model_and_zero_tokens(self): + backend = MockBackend() + task = TaskRecord( + id="t1", + project="test", + intent="write summary", + reference_kind="rule", + judge={"operator": "section_present", "target": "Summary"}, + ) + res = replay_one(backend, task, skill="# Summary\nhello", memory="") + self.assertEqual(res.judge_model, "rule") + self.assertEqual(res.judge_tokens, 0) + self.assertGreaterEqual(res.judge_latency_ms, 0.0) + + def test_replay_one_rubric_judge_measures_latency_and_tokens(self): + backend = InstrumentedJudgeBackend(model_name="rubric-eval-llm") + task = TaskRecord( + id="t2", + project="test", + intent="implement feature", + reference_kind="rubric", + reference="clean architecture", + ) + res = replay_one(backend, task, skill="", memory="") + self.assertEqual(res.judge_model, "rubric-eval-llm") + self.assertEqual(res.judge_tokens, 42) + self.assertGreater(res.judge_latency_ms, 0.0) + + def test_dual_backend_judge_model_attribution(self): + target = MockBackend() + target.name = "llama-3-8b" + optimizer = MockBackend() + optimizer.name = "gpt-4o-judge" + + dual = DualBackend(target, optimizer) + self.assertEqual(dual.judge_model_name, "gpt-4o-judge") + + def test_consolidate_aggregates_judge_metrics(self): + backend = InstrumentedJudgeBackend(model_name="judge-agg-model") + tasks = [ + TaskRecord( + id="t1", + project="test", + intent="p1", + reference_kind="rubric", + reference="ref1", + split="val", + ), + TaskRecord( + id="t2", + project="test", + intent="p2", + reference_kind="rubric", + reference="ref2", + split="val", + ), + ] + result = consolidate( + backend, + tasks, + skill="", + memory="", + gate_mode="on", + ) + self.assertGreaterEqual(result.judge_requests, 2) + self.assertGreaterEqual(result.judge_tokens, 84) + self.assertGreater(result.judge_latency_ms, 0.0) + self.assertEqual(result.judge_model, "judge-agg-model") + + def test_render_report_md_includes_judge_observability(self): + report = SleepReport( + night=1, + project="test", + tokens_used=1500, + judge_requests=4, + judge_latency_ms=312.5, + judge_tokens=850, + judge_model="gpt-4o", + ) + rendered = _render_report_md(report, {}) + self.assertIn("- judge: 4 request(s), 312.5ms, 850 tokens (model: `gpt-4o`)", rendered) + + def test_run_sleep_cycle_records_judge_metrics(self): + backend = InstrumentedJudgeBackend(model_name="cycle-judge-model") + tasks = [ + TaskRecord( + id="t1", + project="test", + intent="prompt 1", + reference_kind="rubric", + reference="expected 1", + split="train", + ), + TaskRecord( + id="t2", + project="test", + intent="prompt 2", + reference_kind="rubric", + reference="expected 2", + split="test", + ), + ] + with tempfile.TemporaryDirectory() as td: + cfg = load_config(project=td, gate_mode="off") + outcome = run_sleep_cycle( + cfg, + seed_tasks=tasks, + backend=backend, + dry_run=True, + ) + report = outcome.report + self.assertGreaterEqual(report.judge_requests, 1) + self.assertGreater(report.judge_latency_ms, 0.0) + self.assertGreater(report.judge_tokens, 0) + self.assertEqual(report.judge_model, "cycle-judge-model") + + +if __name__ == "__main__": + unittest.main()