From e051c720d6a26c6bf2f3d03fc0c6bffdbecf454c Mon Sep 17 00:00:00 2001 From: Oleg Silkin <97077423+RobotSail@users.noreply.github.com> Date: Fri, 14 Aug 2026 21:57:39 +0000 Subject: [PATCH] feat(skillopt): add meta-skill module for optimizer cross-epoch memory (#1236) Add a new skillopt/ package implementing optimizer-side meta-skill memory that accumulates cross-epoch learnings and feeds them back to future optimizer calls. New modules: - skillopt/optimizer/meta_skill.py: Core meta-skill generation with format_meta_skill_context(), run_meta_skill(), load_meta_skill_content(), validate_deployment_gate(), should_generate_meta_skill() - skillopt/prompts/meta_skill.md: LLM prompt template for meta-skill generation (optimizer-coach, not target-facing) - skillopt/engine/trainer.py: Epoch-boundary hook (generate_epoch_meta_skill) and epoch-start loading (load_active_meta_skill) - skillopt/gradient/reflect.py: Reflect functions with meta_skill_context parameter for in-memory prepend to optimizer prompts Safety constraints: - 3000 token cap (approximate: len // 4) - 3-epoch recency window for loading - Score-delta conditioning (skip negative-delta epochs) - Deployment gate validation (detect leakage markers in final SKILL.md) - File separation (meta_skill/ directory, never skills/) Tests: 49 unit tests covering all key functions. --- skillopt/__init__.py | 1 + skillopt/engine/__init__.py | 0 skillopt/engine/trainer.py | 133 ++++++++++ skillopt/gradient/__init__.py | 0 skillopt/gradient/reflect.py | 209 +++++++++++++++ skillopt/optimizer/__init__.py | 0 skillopt/optimizer/meta_skill.py | 244 ++++++++++++++++++ skillopt/prompts/__init__.py | 32 +++ skillopt/prompts/meta_skill.md | 42 ++++ tests/test_meta_skill.py | 420 +++++++++++++++++++++++++++++++ 10 files changed, 1081 insertions(+) create mode 100644 skillopt/__init__.py create mode 100644 skillopt/engine/__init__.py create mode 100644 skillopt/engine/trainer.py create mode 100644 skillopt/gradient/__init__.py create mode 100644 skillopt/gradient/reflect.py create mode 100644 skillopt/optimizer/__init__.py create mode 100644 skillopt/optimizer/meta_skill.py create mode 100644 skillopt/prompts/__init__.py create mode 100644 skillopt/prompts/meta_skill.md create mode 100644 tests/test_meta_skill.py diff --git a/skillopt/__init__.py b/skillopt/__init__.py new file mode 100644 index 000000000..c0939be01 --- /dev/null +++ b/skillopt/__init__.py @@ -0,0 +1 @@ +"""SkillOpt — optimizer self-improvement across training epochs.""" diff --git a/skillopt/engine/__init__.py b/skillopt/engine/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/skillopt/engine/trainer.py b/skillopt/engine/trainer.py new file mode 100644 index 000000000..29a4d9d42 --- /dev/null +++ b/skillopt/engine/trainer.py @@ -0,0 +1,133 @@ +"""Trainer-level meta-skill integration hooks. + +Provides epoch-boundary and epoch-start hooks for integrating optimizer-side +meta-skill memory into the ReflACT training loop. These functions are called +by the main trainer at specific lifecycle points. + +Epoch lifecycle with meta-skill: + +1. **Epoch start** — ``load_active_meta_skill()`` loads previous epoch's + meta-skill content so it can be passed to reflect calls. +2. **Training steps** — ``meta_skill_context`` is passed to each + ``adapter.reflect()`` call, which injects it into optimizer prompts. +3. **Epoch end** (after slow update) — ``generate_epoch_meta_skill()`` + produces updated meta-skill from adjacent-epoch comparison. +""" +from __future__ import annotations + +import json +import logging +import os + +from skillopt.optimizer.meta_skill import ( + load_meta_skill_content, + run_meta_skill, + should_generate_meta_skill, +) + +log = logging.getLogger(__name__) + + +def load_active_meta_skill( + out_root: str, + epoch: int, + *, + use_meta_skill: bool = False, +) -> str: + """Load the active meta-skill for the current epoch. + + Called at the start of each epoch. Returns the meta-skill content from + the previous epoch, or empty string if disabled or unavailable. + """ + if not use_meta_skill: + return "" + content = load_meta_skill_content(out_root, epoch - 1) + if content: + log.info( + "meta_skill.loaded", + epoch=epoch, + source_epoch=epoch - 1, + chars=len(content), + ) + return content + + +def generate_epoch_meta_skill( + out_root: str, + epoch: int, + prev_skill: str, + curr_skill: str, + comparison_pairs: list[dict], + *, + score_delta: float | None = None, + chat_fn: object | None = None, +) -> dict | None: + """Generate meta-skill at epoch boundary. + + Called after the slow update (if any) at the end of each epoch. + Handles resume safety, first-epoch skip, and score-delta conditioning. + + Returns the meta-skill result dict, or None if skipped/failed. + """ + meta_skill_dir = os.path.join(out_root, "meta_skill", f"epoch_{epoch:02d}") + done_path = os.path.join(meta_skill_dir, "meta_skill_result.json") + os.makedirs(meta_skill_dir, exist_ok=True) + + if os.path.exists(done_path): + log.info("meta_skill.resume", epoch=epoch, status="already_done") + with open(done_path) as f: + return json.load(f) + + if epoch == 1: + sentinel = {"action": "skip_first_epoch", "epoch": epoch} + with open(done_path, "w") as f: + json.dump(sentinel, f, indent=2, ensure_ascii=False) + log.info("meta_skill.skip", epoch=epoch, reason="first_epoch") + return sentinel + + if not should_generate_meta_skill(epoch, score_delta): + sentinel = { + "action": "skip_negative_delta", + "epoch": epoch, + "score_delta": score_delta, + } + with open(done_path, "w") as f: + json.dump(sentinel, f, indent=2, ensure_ascii=False) + log.info( + "meta_skill.skip", + epoch=epoch, + reason="negative_delta", + delta=score_delta, + ) + return sentinel + + prev_meta_skill = load_meta_skill_content(out_root, epoch - 1) + + result = run_meta_skill( + prev_skill=prev_skill, + curr_skill=curr_skill, + comparison_pairs=comparison_pairs, + prev_meta_skill_content=prev_meta_skill, + chat_fn=chat_fn, + ) + + if result and result.get("meta_skill_content"): + result["action"] = "write_meta_skill" + result["epoch"] = epoch + with open(done_path, "w") as f: + json.dump(result, f, indent=2, ensure_ascii=False) + log.info( + "meta_skill.generated", + epoch=epoch, + chars=len(result["meta_skill_content"]), + ) + return result + + fallback = { + "action": "generation_failed", + "epoch": epoch, + } + with open(done_path, "w") as f: + json.dump(fallback, f, indent=2, ensure_ascii=False) + log.warning("meta_skill.failed", epoch=epoch) + return None diff --git a/skillopt/gradient/__init__.py b/skillopt/gradient/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/skillopt/gradient/reflect.py b/skillopt/gradient/reflect.py new file mode 100644 index 000000000..41c72c29a --- /dev/null +++ b/skillopt/gradient/reflect.py @@ -0,0 +1,209 @@ +"""Reflect-stage meta-skill injection. + +Provides reflect wrapper functions that accept ``meta_skill_context`` and +prepend it to optimizer prompts in-memory. The on-disk analyst prompt +templates (analyst_error.md, analyst_success.md) are NEVER modified. + +The meta-skill context is prepended to the user message before trajectories, +so the optimizer sees accumulated cross-epoch guidance before analyzing the +current batch. +""" +from __future__ import annotations + +import logging + +from skillopt.optimizer.meta_skill import format_meta_skill_context + +log = logging.getLogger(__name__) + + +def reflect_on_errors( + skill_content: str, + failed_items: list[dict], + prediction_dir: str, + *, + system_prompt: str | None = None, + step_buffer_context: str = "", + meta_skill_context: str = "", + edit_budget: int = 4, + chat_fn: object | None = None, +) -> dict | None: + """Analyze failed trajectories with optional meta-skill context. + + Wraps the error analyst call, prepending meta-skill guidance to the + user prompt in-memory. Does NOT modify any prompt files on disk. + + Parameters + ---------- + skill_content: + Current skill document text. + failed_items: + Rollout result dicts for failed trajectories. + prediction_dir: + Path to predictions directory with conversation files. + system_prompt: + Custom system prompt override. + step_buffer_context: + Summary of previous steps in this epoch. + meta_skill_context: + Raw meta-skill content to prepend to the optimizer prompt. + edit_budget: + Maximum number of edits to propose. + chat_fn: + Callable ``(system, user, **kw) -> (response_text, metadata)``. + """ + if not failed_items: + return None + + user = f"## Current Skill\n{skill_content}\n\n" + user += f"## Edit Budget\nProduce at most L={edit_budget} edits.\n\n" + + if step_buffer_context.strip(): + user += f"## Previous Steps in This Epoch\n{step_buffer_context}\n\n" + + optimizer_ctx = format_meta_skill_context(meta_skill_context) + if optimizer_ctx: + user += optimizer_ctx + "\n\n" + + user += f"## Failed Trajectories ({len(failed_items)} total)\n" + user += _format_items_summary(failed_items) + + if chat_fn is None: + return None + + try: + response, _ = chat_fn( + system=system_prompt or "", + user=user, + max_completion_tokens=16384, + retries=3, + stage="analyst", + ) + from skillopt.optimizer.meta_skill import _extract_json + result = _extract_json(response) + if result: + result["source_type"] = "failure" + return result + except Exception: + log.exception("reflect_on_errors failed") + + return None + + +def reflect_on_successes( + skill_content: str, + success_items: list[dict], + prediction_dir: str, + *, + system_prompt: str | None = None, + step_buffer_context: str = "", + meta_skill_context: str = "", + edit_budget: int = 4, + chat_fn: object | None = None, +) -> dict | None: + """Analyze successful trajectories with optional meta-skill context. + + Same pattern as ``reflect_on_errors`` but for success trajectories. + Meta-skill context is prepended in-memory only. + """ + if not success_items: + return None + + user = f"## Current Skill\n{skill_content}\n\n" + user += f"## Edit Budget\nProduce at most L={edit_budget} edits.\n\n" + + if step_buffer_context.strip(): + user += f"## Previous Steps in This Epoch\n{step_buffer_context}\n\n" + + optimizer_ctx = format_meta_skill_context(meta_skill_context) + if optimizer_ctx: + user += optimizer_ctx + "\n\n" + + user += f"## Successful Trajectories ({len(success_items)} total)\n" + user += _format_items_summary(success_items) + + if chat_fn is None: + return None + + try: + response, _ = chat_fn( + system=system_prompt or "", + user=user, + max_completion_tokens=16384, + retries=3, + stage="analyst", + ) + from skillopt.optimizer.meta_skill import _extract_json + result = _extract_json(response) + if result: + result["source_type"] = "success" + return result + except Exception: + log.exception("reflect_on_successes failed") + + return None + + +def reflect_and_merge( + skill_content: str, + failure_patches: list[dict], + success_patches: list[dict], + *, + meta_skill_context: str = "", + chat_fn: object | None = None, +) -> dict | None: + """Merge failure and success patches with meta-skill context. + + Meta-skill context is prepended to the merge prompt in-memory only. + """ + user = f"## Current Skill\n{skill_content}\n\n" + + optimizer_ctx = format_meta_skill_context(meta_skill_context) + if optimizer_ctx: + user += optimizer_ctx + "\n\n" + + user += f"## Failure Patches ({len(failure_patches)} total)\n" + for i, patch in enumerate(failure_patches, 1): + user += f"### Patch {i}\n{_format_patch(patch)}\n\n" + + user += f"## Success Patches ({len(success_patches)} total)\n" + for i, patch in enumerate(success_patches, 1): + user += f"### Patch {i}\n{_format_patch(patch)}\n\n" + + if chat_fn is None: + return None + + try: + response, _ = chat_fn( + system="", + user=user, + max_completion_tokens=16384, + retries=3, + stage="merge", + ) + from skillopt.optimizer.meta_skill import _extract_json + return _extract_json(response) + except Exception: + log.exception("reflect_and_merge failed") + + return None + + +def _format_items_summary(items: list[dict]) -> str: + """Format rollout items into a compact summary.""" + parts: list[str] = [] + for item in items: + task_id = item.get("id", "?") + task_desc = item.get("task_description", item.get("instruction", "")) + fail_reason = item.get("fail_reason", "") + line = f"- Task {task_id}: {task_desc}" + if fail_reason: + line += f" (reason: {fail_reason})" + parts.append(line) + return "\n".join(parts) + + +def _format_patch(patch: dict) -> str: + """Format a single patch dict into readable text.""" + import json + return json.dumps(patch, indent=2, ensure_ascii=False) diff --git a/skillopt/optimizer/__init__.py b/skillopt/optimizer/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/skillopt/optimizer/meta_skill.py b/skillopt/optimizer/meta_skill.py new file mode 100644 index 000000000..478d269af --- /dev/null +++ b/skillopt/optimizer/meta_skill.py @@ -0,0 +1,244 @@ +"""Optimizer-side meta-skill memory for cross-epoch optimization guidance. + +Maintains a compact optimizer-facing memory distilled from adjacent-epoch +skill comparisons. Does NOT modify the target skill document. Instead, it +produces guidance meant to improve future optimizer behavior when proposing, +merging, and ranking edits. +""" +from __future__ import annotations + +import json +import logging +import os +import traceback + +from skillopt.prompts import load_prompt + +log = logging.getLogger(__name__) + +MAX_META_SKILL_TOKENS = 3000 +RECENCY_WINDOW = 3 + + +def format_meta_skill_context(meta_skill_content: str) -> str: + """Render optimizer memory into a prompt-ready context block. + + Returns an empty string when *meta_skill_content* is empty so that callers + can unconditionally prepend the result without checking. + """ + content = (meta_skill_content or "").strip() + if not content: + return "" + content = _enforce_token_cap(content) + return ( + "## Optimizer Meta Skill\n" + "This is optimizer-side memory distilled from prior epoch transitions in " + "this environment. Use it to improve how you propose, merge, and rank " + "skill edits. Prefer it when the current evidence is ambiguous, but do " + "not force it if the current trajectories clearly contradict it.\n\n" + f"{content}" + ) + + +def run_meta_skill( + prev_skill: str, + curr_skill: str, + comparison_pairs: list[dict], + *, + prev_meta_skill_content: str = "", + system_prompt: str | None = None, + chat_fn: object | None = None, +) -> dict | None: + """Produce updated optimizer-side meta-skill from adjacent epochs. + + Parameters + ---------- + prev_skill: + The last-step skill from the previous epoch. + curr_skill: + The last-step skill from the current epoch. + comparison_pairs: + Longitudinal comparison pairs (same tasks, two skill versions). + prev_meta_skill_content: + Previous epoch's meta-skill content (empty for first generation). + system_prompt: + Override the default meta-skill system prompt. + chat_fn: + Callable ``(system, user, **kw) -> (response_text, metadata)``. + When ``None`` the function returns ``None`` (dry-run / testing). + """ + actual_system = system_prompt if system_prompt is not None else load_prompt("meta_skill") + + prev_meta_section = ( + prev_meta_skill_content.strip() + if prev_meta_skill_content and prev_meta_skill_content.strip() + else "(No previous optimizer meta skill — this is the first update.)" + ) + + comparison_text = _format_comparison_text(comparison_pairs) + user = ( + f"## Previous Epoch Last-Step Skill\n{prev_skill}\n\n" + f"## Current Epoch Last-Step Skill\n{curr_skill}\n\n" + f"## Previous Optimizer Meta Skill\n" + f"The following optimizer memory was available during the current epoch. " + f"Reflect on whether it improved or harmed the quality of edits.\n\n" + f"{prev_meta_section}\n\n" + f"## Longitudinal Comparison (same tasks, two last-step skills)\n" + f"{comparison_text}" + ) + + if chat_fn is None: + return None + + try: + response, _ = chat_fn( + system=actual_system, + user=user, + max_completion_tokens=16384, + retries=3, + stage="meta_skill", + ) + result = _extract_json(response) + if result and result.get("meta_skill_content"): + content = str(result["meta_skill_content"]).strip() + content = _enforce_token_cap(content) + return { + "reasoning": str(result.get("reasoning", "")).strip(), + "meta_skill_content": content, + } + except Exception: + traceback.print_exc() + + return None + + +def load_meta_skill_content(out_root: str, epoch: int) -> str: + """Load meta-skill content from a previous epoch's result file. + + Searches the last ``RECENCY_WINDOW`` epochs and returns the most recent + content found. Returns empty string if nothing is found. + """ + if epoch <= 0: + return "" + start_epoch = max(1, epoch - RECENCY_WINDOW + 1) + for e in range(epoch, start_epoch - 1, -1): + path = os.path.join( + out_root, "meta_skill", f"epoch_{e:02d}", "meta_skill_result.json", + ) + if not os.path.exists(path): + continue + try: + with open(path) as f: + result = json.load(f) + content = str(result.get("meta_skill_content", "")).strip() + if content: + return content + except Exception: + continue + return "" + + +def validate_deployment_gate(skill_content: str) -> list[str]: + """Check final skill for meta-skill leakage markers. + + Returns a list of warnings (empty if clean). + """ + markers = ["Optimizer Meta Skill", "optimizer memory", "meta_skill"] + warnings: list[str] = [] + for marker in markers: + if marker in skill_content: + warnings.append( + f"Deployment gate: meta-skill marker '{marker}' found in final SKILL.md" + ) + return warnings + + +def should_generate_meta_skill( + epoch: int, + score_delta: float | None, +) -> bool: + """Determine whether meta-skill generation should run for this epoch. + + Requires epoch >= 2 and a non-negative score delta. + """ + if epoch < 2: + return False + if score_delta is not None and score_delta < 0: + return False + return True + + +def _enforce_token_cap(content: str) -> str: + """Truncate content to approximate token budget.""" + approx_tokens = len(content) // 4 + if approx_tokens <= MAX_META_SKILL_TOKENS: + return content + char_limit = MAX_META_SKILL_TOKENS * 4 + return content[:char_limit] + + +def _format_comparison_text(pairs: list[dict]) -> str: + """Format comparison pairs into human-readable text for the optimizer.""" + if not pairs: + return "(No comparison data available.)" + + by_cat: dict[str, list[dict]] = { + "regressed": [], + "persistent_fail": [], + "improved": [], + "stable_success": [], + } + for p in pairs: + by_cat.setdefault(p.get("category", ""), []).append(p) + + total = len(pairs) + parts = [ + f"Total samples: {total}\n" + f"- Improved (wrong->right): {len(by_cat['improved'])}\n" + f"- Regressed (right->wrong): {len(by_cat['regressed'])}\n" + f"- Persistent failures (wrong->wrong): {len(by_cat['persistent_fail'])}\n" + f"- Stable successes (right->right): {len(by_cat['stable_success'])}\n" + ] + + for cat_key, label in [ + ("regressed", "Regressions (right->wrong)"), + ("persistent_fail", "Persistent Failures (wrong->wrong)"), + ("improved", "Improvements (wrong->right)"), + ("stable_success", "Stable Successes (right->right)"), + ]: + entries = by_cat[cat_key] + if not entries: + parts.append(f"### {label}\n(none)\n") + continue + lines = [f"### {label}"] + for e in entries: + task_id = e.get("id", "?") + task_desc = e.get("task", "") + lines.append(f"- Task {task_id}: {task_desc}") + parts.append("\n".join(lines)) + + return "\n\n".join(parts) + + +def _extract_json(text: str) -> dict | None: + """Extract the first JSON object from a text response.""" + text = text.strip() + start = text.find("{") + if start == -1: + return None + depth = 0 + end = start + for i, ch in enumerate(text[start:], start): + if ch == "{": + depth += 1 + elif ch == "}": + depth -= 1 + if depth == 0: + end = i + 1 + break + if depth != 0: + return None + try: + return json.loads(text[start:end]) + except json.JSONDecodeError: + return None diff --git a/skillopt/prompts/__init__.py b/skillopt/prompts/__init__.py new file mode 100644 index 000000000..b568ddb33 --- /dev/null +++ b/skillopt/prompts/__init__.py @@ -0,0 +1,32 @@ +"""Prompt loading utilities for SkillOpt. + +Prompts are stored as ``.md`` files in this directory and loaded at runtime. +""" +from __future__ import annotations + +import os + +_PROMPTS_DIR = os.path.dirname(os.path.abspath(__file__)) +_cache: dict[str, str] = {} + + +def load_prompt(name: str) -> str: + """Load a prompt template by name from the prompts directory. + + Returns the contents of ``skillopt/prompts/{name}.md``. + Raises ``FileNotFoundError`` if the file does not exist. + """ + path = os.path.join(_PROMPTS_DIR, f"{name}.md") + if path in _cache: + return _cache[path] + if not os.path.isfile(path): + raise FileNotFoundError(f"Prompt '{name}' not found at {path}") + with open(path, encoding="utf-8") as f: + content = f.read() + _cache[path] = content + return content + + +def clear_cache() -> None: + """Clear the prompt file cache (useful for testing).""" + _cache.clear() diff --git a/skillopt/prompts/meta_skill.md b/skillopt/prompts/meta_skill.md new file mode 100644 index 000000000..2f3280b00 --- /dev/null +++ b/skillopt/prompts/meta_skill.md @@ -0,0 +1,42 @@ +You are an optimizer-coach for an AI agent skill optimization system. + +Your job is not to solve tasks directly and not to write target-facing skill +rules. Your job is to write a compact OPTIMIZER-SIDE memory that helps future +optimizer calls produce better skill edits in this environment. + +## What You Receive + +1. The previous epoch's last-step skill. +2. The current epoch's last-step skill. +3. A longitudinal comparison on the SAME sampled tasks under those two skills. +4. The previous optimizer meta skill, if one existed. + +## Your Goal + +Write a concise meta skill that improves future optimizer behavior in stages such +as failure analysis, success analysis, patch merging, and edit ranking. + +This meta skill should capture things like: +- Which kinds of edits tend to help in this environment. +- Which kinds of edits tend to be too vague, redundant, brittle, or harmful. +- What level of abstraction works best for rules here. +- What failure-repair patterns should be prioritized. +- What regression risks future optimizer calls should guard against. + +## Important Constraints + +- Address the FUTURE OPTIMIZER directly, not the target. +- Focus on how to write better edits and organize better skill updates. +- Use evidence from the adjacent-epoch comparison, not generic advice. +- Keep it compact and high-signal. Prefer a few durable principles. +- Revise or remove parts of the previous meta skill if they did not help. +- Do not output target-facing task instructions. +- Do not restate the whole skill; summarize editing strategy. +- Stay abstract: capture patterns (e.g. "conditional rules regress easily"), + not line-level specifics (e.g. "change line 42 to say X"). + +Respond ONLY with a valid JSON object: +{ + "reasoning": "", + "meta_skill_content": "" +} diff --git a/tests/test_meta_skill.py b/tests/test_meta_skill.py new file mode 100644 index 000000000..0563d08d4 --- /dev/null +++ b/tests/test_meta_skill.py @@ -0,0 +1,420 @@ +"""Tests for SkillOpt meta-skill module.""" +from __future__ import annotations + +import json + +from skillopt.optimizer.meta_skill import ( + MAX_META_SKILL_TOKENS, + format_meta_skill_context, + load_meta_skill_content, + run_meta_skill, + should_generate_meta_skill, + validate_deployment_gate, + _enforce_token_cap, + _extract_json, + _format_comparison_text, +) +from skillopt.engine.trainer import ( + generate_epoch_meta_skill, + load_active_meta_skill, +) +from skillopt.gradient.reflect import ( + reflect_on_errors, + reflect_on_successes, + reflect_and_merge, +) + + +class TestFormatMetaSkillContext: + def test_empty_input_returns_empty(self): + assert format_meta_skill_context("") == "" + + def test_none_input_returns_empty(self): + assert format_meta_skill_context(None) == "" + + def test_whitespace_only_returns_empty(self): + assert format_meta_skill_context(" \n ") == "" + + def test_non_empty_returns_formatted_block(self): + result = format_meta_skill_context("Prefer concrete edits over vague ones.") + assert "## Optimizer Meta Skill" in result + assert "Prefer concrete edits over vague ones." in result + assert "optimizer-side memory" in result + + def test_content_is_token_capped(self): + long_content = "x" * (MAX_META_SKILL_TOKENS * 4 + 1000) + result = format_meta_skill_context(long_content) + content_part = result.split("## Optimizer Meta Skill\n")[1] + header_end = content_part.find("\n\n") + 2 + raw_content = content_part[header_end:] + assert len(raw_content) // 4 <= MAX_META_SKILL_TOKENS + + +class TestRunMetaSkill: + def test_returns_none_without_chat_fn(self): + result = run_meta_skill( + prev_skill="prev", + curr_skill="curr", + comparison_pairs=[], + chat_fn=None, + ) + assert result is None + + def test_returns_parsed_result_with_mock_chat(self): + response_json = json.dumps({ + "reasoning": "Concrete edits worked better", + "meta_skill_content": "Prefer specific over vague edits.", + }) + + def mock_chat(system, user, **kwargs): + return (response_json, {}) + + result = run_meta_skill( + prev_skill="skill v1", + curr_skill="skill v2", + comparison_pairs=[{"id": "1", "category": "improved", "task": "test"}], + chat_fn=mock_chat, + ) + assert result is not None + assert result["reasoning"] == "Concrete edits worked better" + assert result["meta_skill_content"] == "Prefer specific over vague edits." + + def test_returns_none_on_empty_meta_skill_content(self): + response_json = json.dumps({ + "reasoning": "nothing useful", + "meta_skill_content": "", + }) + + def mock_chat(system, user, **kwargs): + return (response_json, {}) + + result = run_meta_skill( + prev_skill="prev", + curr_skill="curr", + comparison_pairs=[], + chat_fn=mock_chat, + ) + assert result is None + + def test_returns_none_on_chat_exception(self): + def failing_chat(system, user, **kwargs): + raise RuntimeError("API error") + + result = run_meta_skill( + prev_skill="prev", + curr_skill="curr", + comparison_pairs=[], + chat_fn=failing_chat, + ) + assert result is None + + def test_custom_system_prompt_is_used(self): + captured = {} + + def mock_chat(system, user, **kwargs): + captured["system"] = system + return (json.dumps({"reasoning": "x", "meta_skill_content": "y"}), {}) + + run_meta_skill( + prev_skill="prev", + curr_skill="curr", + comparison_pairs=[], + system_prompt="Custom prompt here", + chat_fn=mock_chat, + ) + assert captured["system"] == "Custom prompt here" + + def test_token_cap_applied_to_result(self): + long_content = "x" * (MAX_META_SKILL_TOKENS * 4 + 5000) + response_json = json.dumps({ + "reasoning": "test", + "meta_skill_content": long_content, + }) + + def mock_chat(system, user, **kwargs): + return (response_json, {}) + + result = run_meta_skill( + prev_skill="prev", + curr_skill="curr", + comparison_pairs=[], + chat_fn=mock_chat, + ) + assert result is not None + assert len(result["meta_skill_content"]) // 4 <= MAX_META_SKILL_TOKENS + + +class TestLoadMetaSkillContent: + def test_returns_empty_for_epoch_zero(self, tmp_path): + assert load_meta_skill_content(str(tmp_path), 0) == "" + + def test_returns_empty_when_no_file(self, tmp_path): + assert load_meta_skill_content(str(tmp_path), 3) == "" + + def test_loads_existing_content(self, tmp_path): + epoch_dir = tmp_path / "meta_skill" / "epoch_02" + epoch_dir.mkdir(parents=True) + result_file = epoch_dir / "meta_skill_result.json" + result_file.write_text(json.dumps({ + "meta_skill_content": "Use concrete edits.", + })) + assert load_meta_skill_content(str(tmp_path), 2) == "Use concrete edits." + + def test_recency_window_skips_old_epochs(self, tmp_path): + # Write content at epoch 1 only + epoch_dir = tmp_path / "meta_skill" / "epoch_01" + epoch_dir.mkdir(parents=True) + (epoch_dir / "meta_skill_result.json").write_text(json.dumps({ + "meta_skill_content": "Old content.", + })) + # Requesting epoch 5 should not find epoch 1 (outside 3-epoch window) + assert load_meta_skill_content(str(tmp_path), 5) == "" + + def test_recency_window_finds_recent_epoch(self, tmp_path): + epoch_dir = tmp_path / "meta_skill" / "epoch_03" + epoch_dir.mkdir(parents=True) + (epoch_dir / "meta_skill_result.json").write_text(json.dumps({ + "meta_skill_content": "Recent content.", + })) + # Requesting epoch 5 should find epoch 3 (within 3-epoch window) + assert load_meta_skill_content(str(tmp_path), 5) == "Recent content." + + def test_handles_malformed_json(self, tmp_path): + epoch_dir = tmp_path / "meta_skill" / "epoch_02" + epoch_dir.mkdir(parents=True) + (epoch_dir / "meta_skill_result.json").write_text("not valid json{{{") + assert load_meta_skill_content(str(tmp_path), 2) == "" + + +class TestValidateDeploymentGate: + def test_clean_skill_returns_no_warnings(self): + skill = "# Task Rules\n\n- Do X when Y happens.\n- Always check Z." + assert validate_deployment_gate(skill) == [] + + def test_detects_meta_skill_markers(self): + skill = "# Task Rules\n\n## Optimizer Meta Skill\nSome guidance here." + warnings = validate_deployment_gate(skill) + assert len(warnings) >= 1 + assert any("Optimizer Meta Skill" in w for w in warnings) + + def test_detects_optimizer_memory_marker(self): + skill = "This uses optimizer memory from prior epochs." + warnings = validate_deployment_gate(skill) + assert len(warnings) >= 1 + + def test_detects_meta_skill_underscore_marker(self): + skill = "The meta_skill content was generated." + warnings = validate_deployment_gate(skill) + assert len(warnings) >= 1 + + +class TestShouldGenerateMetaSkill: + def test_epoch_1_returns_false(self): + assert should_generate_meta_skill(1, score_delta=0.05) is False + + def test_epoch_0_returns_false(self): + assert should_generate_meta_skill(0, score_delta=0.1) is False + + def test_epoch_2_positive_delta_returns_true(self): + assert should_generate_meta_skill(2, score_delta=0.01) is True + + def test_epoch_2_negative_delta_returns_false(self): + assert should_generate_meta_skill(2, score_delta=-0.05) is False + + def test_epoch_2_zero_delta_returns_true(self): + assert should_generate_meta_skill(2, score_delta=0.0) is True + + def test_none_delta_returns_true(self): + assert should_generate_meta_skill(3, score_delta=None) is True + + +class TestTokenCapEnforcement: + def test_short_content_unchanged(self): + content = "Short content." + assert _enforce_token_cap(content) == content + + def test_long_content_truncated(self): + content = "x" * (MAX_META_SKILL_TOKENS * 4 + 2000) + result = _enforce_token_cap(content) + assert len(result) <= MAX_META_SKILL_TOKENS * 4 + assert len(result) // 4 <= MAX_META_SKILL_TOKENS + + def test_exact_boundary_unchanged(self): + content = "x" * (MAX_META_SKILL_TOKENS * 4) + assert _enforce_token_cap(content) == content + + +class TestExtractJson: + def test_extracts_json_from_text(self): + text = 'Here is the result: {"key": "value"} done.' + assert _extract_json(text) == {"key": "value"} + + def test_returns_none_for_no_json(self): + assert _extract_json("no json here") is None + + def test_handles_nested_json(self): + text = '{"outer": {"inner": "val"}}' + result = _extract_json(text) + assert result == {"outer": {"inner": "val"}} + + def test_returns_none_for_malformed_json(self): + assert _extract_json("{broken: json}") is None + + +class TestFormatComparisonText: + def test_empty_pairs(self): + result = _format_comparison_text([]) + assert "No comparison data" in result + + def test_formats_categories(self): + pairs = [ + {"id": "1", "category": "improved", "task": "task A"}, + {"id": "2", "category": "regressed", "task": "task B"}, + ] + result = _format_comparison_text(pairs) + assert "Improved" in result or "improved" in result.lower() + assert "Regressed" in result or "regressed" in result.lower() + assert "task A" in result + assert "task B" in result + + +class TestGenerateEpochMetaSkill: + def test_skips_first_epoch(self, tmp_path): + result = generate_epoch_meta_skill( + str(tmp_path), epoch=1, + prev_skill="prev", curr_skill="curr", + comparison_pairs=[], + ) + assert result["action"] == "skip_first_epoch" + done_path = tmp_path / "meta_skill" / "epoch_01" / "meta_skill_result.json" + assert done_path.exists() + + def test_resumes_from_existing(self, tmp_path): + meta_dir = tmp_path / "meta_skill" / "epoch_02" + meta_dir.mkdir(parents=True) + existing = {"action": "write_meta_skill", "meta_skill_content": "cached"} + (meta_dir / "meta_skill_result.json").write_text(json.dumps(existing)) + result = generate_epoch_meta_skill( + str(tmp_path), epoch=2, + prev_skill="prev", curr_skill="curr", + comparison_pairs=[], + ) + assert result["action"] == "write_meta_skill" + + def test_skips_negative_delta(self, tmp_path): + result = generate_epoch_meta_skill( + str(tmp_path), epoch=3, + prev_skill="prev", curr_skill="curr", + comparison_pairs=[], + score_delta=-0.1, + ) + assert result["action"] == "skip_negative_delta" + + def test_generates_with_chat_fn(self, tmp_path): + response = json.dumps({ + "reasoning": "test", + "meta_skill_content": "Use targeted edits.", + }) + + def mock_chat(system, user, **kwargs): + return (response, {}) + + result = generate_epoch_meta_skill( + str(tmp_path), epoch=2, + prev_skill="prev", curr_skill="curr", + comparison_pairs=[], + score_delta=0.05, + chat_fn=mock_chat, + ) + assert result is not None + assert result["action"] == "write_meta_skill" + assert result["meta_skill_content"] == "Use targeted edits." + done_path = tmp_path / "meta_skill" / "epoch_02" / "meta_skill_result.json" + assert done_path.exists() + + +class TestLoadActiveMetaSkill: + def test_returns_empty_when_disabled(self, tmp_path): + result = load_active_meta_skill(str(tmp_path), epoch=3, use_meta_skill=False) + assert result == "" + + def test_returns_empty_when_no_file(self, tmp_path): + result = load_active_meta_skill(str(tmp_path), epoch=3, use_meta_skill=True) + assert result == "" + + def test_loads_previous_epoch(self, tmp_path): + epoch_dir = tmp_path / "meta_skill" / "epoch_02" + epoch_dir.mkdir(parents=True) + (epoch_dir / "meta_skill_result.json").write_text(json.dumps({ + "meta_skill_content": "Loaded content.", + })) + result = load_active_meta_skill(str(tmp_path), epoch=3, use_meta_skill=True) + assert result == "Loaded content." + + +class TestReflectFunctions: + def test_reflect_on_errors_returns_none_without_items(self): + result = reflect_on_errors( + skill_content="skill", + failed_items=[], + prediction_dir="/tmp", + ) + assert result is None + + def test_reflect_on_errors_returns_none_without_chat_fn(self): + result = reflect_on_errors( + skill_content="skill", + failed_items=[{"id": "1", "task_description": "test"}], + prediction_dir="/tmp", + ) + assert result is None + + def test_reflect_on_errors_with_meta_skill_context(self): + captured = {} + + def mock_chat(system, user, **kwargs): + captured["user"] = user + return (json.dumps({"patch": {"edits": []}}), {}) + + reflect_on_errors( + skill_content="skill content", + failed_items=[{"id": "1", "task_description": "test"}], + prediction_dir="/tmp", + meta_skill_context="Use concrete edits.", + chat_fn=mock_chat, + ) + assert "Optimizer Meta Skill" in captured["user"] + assert "Use concrete edits." in captured["user"] + + def test_reflect_on_successes_returns_none_without_items(self): + result = reflect_on_successes( + skill_content="skill", + success_items=[], + prediction_dir="/tmp", + ) + assert result is None + + def test_reflect_and_merge_returns_none_without_chat_fn(self): + result = reflect_and_merge( + skill_content="skill", + failure_patches=[], + success_patches=[], + ) + assert result is None + + def test_reflect_and_merge_includes_meta_skill(self): + captured = {} + + def mock_chat(system, user, **kwargs): + captured["user"] = user + return (json.dumps({"merged_patch": {"edits": []}}), {}) + + reflect_and_merge( + skill_content="skill", + failure_patches=[{"edits": []}], + success_patches=[], + meta_skill_context="Merge carefully.", + chat_fn=mock_chat, + ) + assert "Optimizer Meta Skill" in captured["user"] + assert "Merge carefully." in captured["user"]