feat(optimizer): skill-aware reflection (EmbodiSkill S_app), config-controlled and env-independent

Split failure reflections into SKILL_DEFECT (body edit) vs EXECUTION_LAPSE
(protected appendix note that re-emphasizes an existing rule, never edited
by step-level analysts). Toggle: optimizer.use_skill_aware_reflection
(default false; baseline byte-identical when off).

- optimizer/appendix.py: protected APPENDIX region (inject/extract/append
  with dedup), mirrors the slow_update protected-field pattern
- optimizer/skill_aware.py: analyst prompt augmentation, appendix_notes
  parsing, threshold-gated LLM consolidation, and a process-wide runtime
  switch (configure_skill_aware_reflection) set once by the trainer
- gradient/reflect.py: augment error/success analyst prompts at runtime;
  None-sentinel kwargs resolve from the global switch, so env adapters
  need no per-benchmark wiring (works for all envs, present and future)
- optimizer/skill.py: generalize the protected-region check to
  (slow_update, appendix); edits inside any protected region are skipped
- engine/trainer.py: inject appendix at init, flush per-step
  EXECUTION_LAPSE notes after the gate settles, optional consolidation
- tests: regression suite incl. toggle-off byte-identical guarantee and
  env-independent global-switch resolution (6/6 passing + live smoke)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

configs/_base_/default.yaml

@@ -81,6 +81,9 @@ optimizer:
   slow_update_gate_with_selection: false
   longitudinal_pair_policy: mixed  # mixed / changed / unchanged
   use_meta_skill: true
+  use_skill_aware_reflection: false  # EmbodiSkill: split failures into SKILL_DEFECT (edit body) vs EXECUTION_LAPSE (protected appendix)
+  skill_aware_appendix_source: both  # both = success+failure emit appendix notes; failure_only = only EXECUTION_LAPSE (paper-faithful)
+  skill_aware_consolidate_threshold: 0  # 0 = off; >0 = LLM-consolidate the appendix when its note count exceeds N
 
 evaluation:
   use_gate: true

scripts/train.py

@@ -245,6 +245,10 @@ def parse_args() -> argparse.Namespace:
     p.add_argument("--longitudinal_pair_policy", type=str,
                    choices=["mixed", "changed", "unchanged"])
     p.add_argument("--use_meta_skill", type=_BOOL)
+    p.add_argument("--use_skill_aware_reflection", type=_BOOL)
+    p.add_argument("--skill_aware_appendix_source", type=str,
+                   choices=["both", "failure_only"])
+    p.add_argument("--skill_aware_consolidate_threshold", type=int)
     p.add_argument("--data_path", type=str)
     p.add_argument("--split_mode", type=str,
                    choices=["ratio", "split_dir"])
@@ -360,6 +364,9 @@ _LEGACY_TO_STRUCTURED: dict[str, str] = {
     "slow_update_samples": "optimizer.slow_update_samples",
     "longitudinal_pair_policy": "optimizer.longitudinal_pair_policy",
     "use_meta_skill": "optimizer.use_meta_skill",
+    "use_skill_aware_reflection": "optimizer.use_skill_aware_reflection",
+    "skill_aware_appendix_source": "optimizer.skill_aware_appendix_source",
+    "skill_aware_consolidate_threshold": "optimizer.skill_aware_consolidate_threshold",
     "use_gate": "evaluation.use_gate",
     "sel_env_num": "evaluation.sel_env_num",
     "test_env_num": "evaluation.test_env_num",
@@ -527,6 +534,7 @@ def main() -> None:
     print(f"  minibatch_size: {cfg.get('minibatch_size')}")
     print(f"  seed:           {cfg.get('seed')}")
     print(f"  meta_skill:     {cfg.get('use_meta_skill', False)}")
+    print(f"  skill_aware_reflection: {cfg.get('use_skill_aware_reflection', False)}")
     print(f"  slow_update:    {cfg.get('use_slow_update', False)}")
     print(f"  out_root:       {cfg.get('out_root')}")
     print(f"{'='*60}\n")

skillopt/config.py

@@ -119,6 +119,9 @@ _FLATTEN_MAP: dict[str, str] = {
     "optimizer.slow_update_gate_with_selection": "slow_update_gate_with_selection",
     "optimizer.longitudinal_pair_policy": "longitudinal_pair_policy",
     "optimizer.use_meta_skill": "use_meta_skill",
+    "optimizer.use_skill_aware_reflection": "use_skill_aware_reflection",
+    "optimizer.skill_aware_appendix_source": "skill_aware_appendix_source",
+    "optimizer.skill_aware_consolidate_threshold": "skill_aware_consolidate_threshold",
     "evaluation.use_gate": "use_gate",
     "evaluation.gate_metric": "gate_metric",
     "evaluation.gate_mixed_weight": "gate_mixed_weight",

skillopt/engine/trainer.py

@@ -32,6 +32,17 @@ from skillopt.optimizer.lr_autonomous import decide_autonomous_learning_rate
 from skillopt.optimizer.rewrite import rewrite_skill_from_suggestions
 from skillopt.optimizer.scheduler import build_scheduler
 from skillopt.optimizer.skill import apply_patch_with_report
+from skillopt.optimizer.appendix import (
+    append_to_appendix_field,
+    extract_appendix_notes as extract_appendix_notes_from_skill,
+    inject_empty_appendix_field,
+    _strip_all_appendix_fields,
+)
+from skillopt.optimizer.skill_aware import (
+    configure_skill_aware_reflection,
+    consolidate_appendix_notes,
+    extract_appendix_notes as extract_appendix_notes_from_result,
+)
 from skillopt.optimizer.slow_update import (
     build_comparison_pairs,
     extract_slow_update_field,
@@ -48,6 +59,7 @@ from skillopt.optimizer.update_modes import (
     short_item_summary,
 )
 from skillopt.model import (
+    chat_optimizer,
     configure_azure_openai,
     configure_claude_code_exec,
     configure_codex_exec,
@@ -838,6 +850,20 @@ class ReflACTTrainer:
 
         _save_skill(out_root, 0, skill_init)
 
+        # ── Skill-aware reflection: ensure the protected appendix (S_app)
+        # region exists on the working skill. Only current_skill carries the
+        # appendix; best_skill stays a faithful val-best snapshot (same policy
+        # as slow_update). No-op when the region already exists (resume-safe).
+        use_skill_aware = cfg.get("use_skill_aware_reflection", False)
+        # Publish the toggle process-wide so run_minibatch_reflect resolves it
+        # from config for EVERY env adapter — no per-benchmark wiring needed.
+        configure_skill_aware_reflection(
+            use_skill_aware,
+            cfg.get("skill_aware_appendix_source", "both"),
+        )
+        if use_skill_aware:
+            current_skill = inject_empty_appendix_field(current_skill)
+
         def _persist_runtime_state(last_completed_step: int) -> None:
             _save_runtime_state(
                 out_root,
@@ -1389,6 +1415,62 @@ class ReflACTTrainer:
                 ):
                     best_origin = current_origin
 
+                # ── Skill-aware reflection: flush execution-lapse reminders ──
+                # After the gate has settled current_skill, append this step's
+                # EXECUTION_LAPSE notes into the protected appendix (S_app).
+                # This bypasses the gate by design (the paper writes appendix
+                # reminders directly) and only touches current_skill, never
+                # best_skill. Body candidate evaluation already happened above
+                # and is unaffected.
+                if use_skill_aware:
+                    step_appendix_notes: list[str] = []
+                    for rp in all_raw_patches:
+                        if isinstance(rp, dict):
+                            step_appendix_notes.extend(extract_appendix_notes_from_result(rp))
+                    if step_appendix_notes:
+                        before_notes = extract_appendix_notes_from_skill(current_skill)
+                        current_skill = append_to_appendix_field(
+                            current_skill, step_appendix_notes,
+                        )
+                        after_notes = extract_appendix_notes_from_skill(current_skill)
+                        n_added = len(after_notes) - len(before_notes)
+                        step_rec["n_execution_lapse_notes"] = len(step_appendix_notes)
+                        step_rec["n_appendix_notes_added"] = n_added
+                        step_rec["n_appendix_notes_total"] = len(after_notes)
+                        with open(os.path.join(step_dir, "appendix_notes.json"), "w") as f:
+                            json.dump(
+                                {
+                                    "step_notes": step_appendix_notes,
+                                    "appendix_after": after_notes,
+                                },
+                                f, indent=2, ensure_ascii=False,
+                            )
+                        print(
+                            f"    [skill-aware] +{n_added} appendix note(s) "
+                            f"(total {len(after_notes)}) from {len(step_appendix_notes)} lapse signal(s)"
+                        )
+                        # Threshold-gated LLM consolidation (paper Eq.11): when the
+                        # appendix grows past N notes, compact it with one optimizer
+                        # call (dedupe / merge / shorten). 0 disables it. Any failure
+                        # leaves the appendix unchanged.
+                        consolidate_threshold = int(
+                            cfg.get("skill_aware_consolidate_threshold", 0) or 0
+                        )
+                        if consolidate_threshold > 0 and len(after_notes) > consolidate_threshold:
+                            compacted = consolidate_appendix_notes(
+                                after_notes, chat_fn=chat_optimizer,
+                            )
+                            if compacted and len(compacted) < len(after_notes):
+                                current_skill = append_to_appendix_field(
+                                    _strip_all_appendix_fields(current_skill), compacted,
+                                )
+                                step_rec["n_appendix_notes_consolidated"] = len(compacted)
+                                step_rec["n_appendix_notes_total"] = len(compacted)
+                                print(
+                                    f"    [skill-aware] consolidated appendix "
+                                    f"{len(after_notes)} -> {len(compacted)} notes"
+                                )
+
                 if gate_metric == "hard":
                     score_label = f"hard={cand_hard:.4f}"
                 elif gate_metric == "soft":

skillopt/gradient/reflect.py

@@ -29,6 +29,13 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from skillopt.model import chat_optimizer
 from skillopt.optimizer.meta_skill import format_meta_skill_context
+from skillopt.optimizer.skill_aware import (
+    augment_error_prompt,
+    augment_success_prompt,
+    extract_appendix_notes,
+    get_skill_aware_appendix_source,
+    is_skill_aware_enabled,
+)
 from skillopt.optimizer.update_modes import (
     get_payload_items,
     is_full_rewrite_minibatch_mode,
@@ -258,6 +265,7 @@ def run_error_analyst_minibatch(
     step_buffer_context: str = "",
     meta_skill_context: str = "",
     update_mode: str = "patch",
+    skill_aware_reflection: bool = False,
 ) -> dict | None:
     """Analyze a minibatch of failed trajectories in one optimizer call.
 
@@ -287,6 +295,11 @@ def run_error_analyst_minibatch(
     """
     mode = normalize_update_mode(update_mode)
     actual_system = _resolve_prompt(system_prompt, "analyst_error", mode)
+    # Skill-aware reflection: augment the resolved prompt at runtime so both
+    # env-specific and generic analyst prompts get the defect/lapse instruction.
+    # When the toggle is off this is a no-op (prompt byte-identical to baseline).
+    if skill_aware_reflection and not is_full_rewrite_minibatch_mode(mode):
+        actual_system = augment_error_prompt(actual_system)
 
     trajectories_text = fmt_minibatch_trajectories(items, prediction_dir)
     if not trajectories_text.strip():
@@ -325,11 +338,26 @@ def run_error_analyst_minibatch(
             stage="analyst",
         )
         result = extract_json(response)
-        if result and "patch" in result:
+        if not result:
+            return None
+        notes = extract_appendix_notes(result) if skill_aware_reflection else []
+        if "patch" in result:
             result["source_type"] = "failure"
             if not is_full_rewrite_minibatch_mode(mode):
                 truncate_payload(result["patch"], edit_budget, mode)
+            if skill_aware_reflection:
+                result["appendix_notes"] = notes
             return result
+        # Skill-aware: a batch may legitimately yield ONLY execution-lapse notes
+        # (no body edit). Return a no-op patch so the notes still reach the
+        # trainer via all_raw_patches; empty edits are dropped from the body
+        # pipeline by _normalise_patches, so body behavior is unchanged.
+        if skill_aware_reflection and notes:
+            return {
+                "source_type": "failure",
+                "patch": {"reasoning": "execution-lapse only", "edits": []},
+                "appendix_notes": notes,
+            }
     except Exception:  # noqa: BLE001
         traceback.print_exc()
     return None
@@ -346,6 +374,8 @@ def run_success_analyst_minibatch(
     step_buffer_context: str = "",
     meta_skill_context: str = "",
     update_mode: str = "patch",
+    skill_aware_reflection: bool = False,
+    emit_appendix_notes: bool = True,
 ) -> dict | None:
     """Analyze a minibatch of successful trajectories in one optimizer call.
 
@@ -365,6 +395,11 @@ def run_success_analyst_minibatch(
     """
     mode = normalize_update_mode(update_mode)
     actual_system = _resolve_prompt(system_prompt, "analyst_success", mode)
+    # Only augment + parse appendix notes on the success side when allowed.
+    # failure_only mode (paper-faithful S_app) suppresses success-side notes.
+    sa_emit = skill_aware_reflection and emit_appendix_notes
+    if sa_emit and not is_full_rewrite_minibatch_mode(mode):
+        actual_system = augment_success_prompt(actual_system)
 
     trajectories_text = fmt_minibatch_trajectories(items, prediction_dir)
     if not trajectories_text.strip():
@@ -404,6 +439,8 @@ def run_success_analyst_minibatch(
             result["source_type"] = "success"
             if not is_full_rewrite_minibatch_mode(mode):
                 truncate_payload(result["patch"], edit_budget, mode)
+            if sa_emit:
+                result["appendix_notes"] = extract_appendix_notes(result)
             return result
     except Exception:  # noqa: BLE001
         traceback.print_exc()
@@ -450,6 +487,8 @@ def run_minibatch_reflect(
     step_buffer_context: str = "",
     meta_skill_context: str = "",
     update_mode: str = "patch",
+    skill_aware_reflection: bool | None = None,
+    skill_aware_appendix_source: str | None = None,
 ) -> list[dict | None]:
     """Full minibatch reflect stage: group → parallel optimizer calls → patches.
 
@@ -484,6 +523,14 @@ def run_minibatch_reflect(
     list[dict | None]
         Patch dicts (with ``source_type`` "failure" or "success").
     """
+    # Resolve the skill-aware toggle: explicit kwargs win; otherwise fall back
+    # to the process-wide config switch set by the trainer, so the feature is
+    # env-independent and adapters need no per-benchmark wiring.
+    if skill_aware_reflection is None:
+        skill_aware_reflection = is_skill_aware_enabled()
+    if skill_aware_appendix_source is None:
+        skill_aware_appendix_source = get_skill_aware_appendix_source()
+
     os.makedirs(patches_dir, exist_ok=True)
 
     # Separate failure / success
@@ -539,6 +586,7 @@ def run_minibatch_reflect(
             trajectory_memory_context=trajectory_memory_context,
             meta_skill_context=meta_skill_context,
             update_mode=update_mode,
+            skill_aware_reflection=skill_aware_reflection,
         )
         return f"minibatch_fail_{idx:03d}", patch
 
@@ -551,6 +599,8 @@ def run_minibatch_reflect(
             trajectory_memory_context=trajectory_memory_context,
             meta_skill_context=meta_skill_context,
             update_mode=update_mode,
+            skill_aware_reflection=skill_aware_reflection,
+            emit_appendix_notes=(skill_aware_appendix_source != "failure_only"),
         )
         return f"minibatch_succ_{idx:03d}", patch
 

skillopt/optimizer/appendix.py

@@ -0,0 +1,156 @@
+"""Skill-Aware Reflection — protected appendix field (EmbodiSkill S_app).
+
+EmbodiSkill (paper 2605.10332v1) splits a skill into ``S = (S_body, S_app)``:
+the body holds the main prescriptive rules; the appendix only *emphasizes*
+existing valid rules that the executor failed to follow (EXECUTION_LAPSE), and
+**never introduces new rules**.
+
+This module owns the appendix region of the skill document. It mirrors the
+protected-field pattern of :mod:`skillopt.optimizer.slow_update`, with two
+differences:
+
+1. **Append semantics** (not replace): execution-lapse reminders accumulate
+   across steps within a run, so new notes are merged into the existing
+   appendix rather than overwriting it.
+2. **Lightweight dedup**: near-duplicate reminders are collapsed (inspired by
+   GMemory's ``_dedupe_preserve_order``) so the appendix stays compact.
+
+The appendix lives **inside** the skill markdown, between dedicated markers, so
+it is persisted by the normal ``_save_skill`` path and is resume-safe. Step-level
+analyst edits cannot modify it (enforced by the shared protected-region check in
+:mod:`skillopt.optimizer.skill`).
+
+Public API
+----------
+- :func:`has_appendix_field`        — check if markers are present
+- :func:`inject_empty_appendix_field` — add empty placeholder (skill init)
+- :func:`extract_appendix_notes`    — read current notes as a list
+- :func:`append_to_appendix_field`  — merge new notes (dedup) into the region
+"""
+from __future__ import annotations
+
+import re
+
+# ── Protected field markers ─────────────────────────────────────────────────
+
+APPENDIX_START = "<!-- APPENDIX_START -->"
+APPENDIX_END = "<!-- APPENDIX_END -->"
+
+# Heading shown inside the rendered appendix block (human-readable only).
+APPENDIX_HEADING = "## Execution Notes Appendix"
+
+# Each note is rendered as a markdown bullet so the target model reads it as
+# ordinary guidance.
+_NOTE_BULLET_PREFIX = "- "
+
+
+# ── Dedup helpers ───────────────────────────────────────────────────────────
+
+
+def _canonicalize(text: str) -> str:
+    """Normalize a note for duplicate detection (whitespace/punct/case-insensitive)."""
+    normalized = re.sub(r"\s+", " ", str(text or "").strip())
+    normalized = normalized.rstrip(" .;:,_-")
+    return normalized.casefold()
+
+
+def _dedupe_preserve_order(notes: list[str]) -> list[str]:
+    """Drop blanks and near-duplicates, preserving first-seen order."""
+    seen: set[str] = set()
+    deduped: list[str] = []
+    for note in notes:
+        text = re.sub(r"\s+", " ", str(note).strip())
+        if not text:
+            continue
+        key = _canonicalize(text)
+        if not key or key in seen:
+            continue
+        seen.add(key)
+        deduped.append(text)
+    return deduped
+
+
+# ── Field manipulation ──────────────────────────────────────────────────────
+
+
+def has_appendix_field(skill: str) -> bool:
+    return APPENDIX_START in skill and APPENDIX_END in skill
+
+
+def _render_block(notes: list[str]) -> str:
+    """Render the full marker-delimited appendix block for *notes*."""
+    lines = [APPENDIX_START, APPENDIX_HEADING]
+    for note in notes:
+        lines.append(f"{_NOTE_BULLET_PREFIX}{note}")
+    lines.append(APPENDIX_END)
+    return "\n".join(lines)
+
+
+def inject_empty_appendix_field(skill: str) -> str:
+    """Add an empty appendix placeholder at the end of *skill* (idempotent).
+
+    Mirrors ``inject_empty_slow_update_field``: called once at skill init so the
+    protected region exists before any note is written.
+    """
+    if has_appendix_field(skill):
+        return skill
+    block = f"\n\n{APPENDIX_START}\n{APPENDIX_HEADING}\n{APPENDIX_END}\n"
+    return skill.rstrip() + block
+
+
+def extract_appendix_notes(skill: str) -> list[str]:
+    """Return the current appendix notes as a list of strings (no markers/heading)."""
+    start = skill.find(APPENDIX_START)
+    end = skill.find(APPENDIX_END)
+    if start == -1 or end == -1:
+        return []
+    inner = skill[start + len(APPENDIX_START):end].strip()
+    notes: list[str] = []
+    for raw_line in inner.splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        if line == APPENDIX_HEADING or line.lstrip("#").strip() == APPENDIX_HEADING.lstrip("#").strip():
+            continue
+        if line.startswith(_NOTE_BULLET_PREFIX):
+            line = line[len(_NOTE_BULLET_PREFIX):].strip()
+        elif line.startswith("-") or line.startswith("*"):
+            line = line[1:].strip()
+        if line:
+            notes.append(line)
+    return notes
+
+
+def _strip_all_appendix_fields(skill: str) -> str:
+    """Remove every appendix marker pair (and content between) from *skill*."""
+    while True:
+        start = skill.find(APPENDIX_START)
+        if start == -1:
+            break
+        end = skill.find(APPENDIX_END, start)
+        if end == -1:
+            skill = skill[:start] + skill[start + len(APPENDIX_START):]
+            break
+        skill = skill[:end + len(APPENDIX_END)].rsplit(APPENDIX_START, 1)[0] + skill[end + len(APPENDIX_END):]
+    skill = skill.replace(APPENDIX_END, "")
+    while "\n\n\n" in skill:
+        skill = skill.replace("\n\n\n", "\n\n")
+    return skill.rstrip()
+
+
+def append_to_appendix_field(skill: str, new_notes: list[str]) -> str:
+    """Merge *new_notes* into the appendix region (dedup), returning updated skill.
+
+    - If no appendix region exists yet, one is created.
+    - Existing notes are preserved; new ones are appended after dedup against the
+      combined set, so order is stable and duplicates are dropped.
+    - Empty / whitespace-only notes are ignored. If the merged set is empty, an
+      empty placeholder region is still ensured.
+    """
+    incoming = _dedupe_preserve_order(list(new_notes or []))
+    existing = extract_appendix_notes(skill)
+    merged = _dedupe_preserve_order(existing + incoming)
+
+    base = _strip_all_appendix_fields(skill)
+    block = _render_block(merged)
+    return f"{base}\n\n{block}\n"

skillopt/optimizer/skill.py

@@ -14,25 +14,62 @@ if TYPE_CHECKING:
 SLOW_UPDATE_START = "<!-- SLOW_UPDATE_START -->"
 SLOW_UPDATE_END = "<!-- SLOW_UPDATE_END -->"
 
+# Skill-aware reflection (EmbodiSkill S_app) appendix region. Like the slow
+# update region, it is protected: step-level analyst edits must not modify it.
+APPENDIX_START = "<!-- APPENDIX_START -->"
+APPENDIX_END = "<!-- APPENDIX_END -->"
 
-def _is_in_slow_update_region(skill: str, target: str) -> bool:
-    """Check if *target* text falls within the protected slow update region."""
-    start_idx = skill.find(SLOW_UPDATE_START)
-    end_idx = skill.find(SLOW_UPDATE_END)
-    if start_idx == -1 or end_idx == -1:
+# All protected (start, end) marker pairs. Step-level edits cannot target text
+# inside any of these regions, and `append` / `insert_after`-fallback ops are
+# inserted before the earliest-occurring region so protected blocks stay at the
+# document tail. With only the slow-update region present, every helper reduces
+# to the original slow-update-only behavior (byte-identical skill output).
+_PROTECTED_REGIONS: tuple[tuple[str, str], ...] = (
+    (SLOW_UPDATE_START, SLOW_UPDATE_END),
+    (APPENDIX_START, APPENDIX_END),
+)
+
+
+def _earliest_protected_start(skill: str) -> int:
+    """Index of the earliest protected-region start marker, or -1 if none."""
+    positions = [
+        idx
+        for idx in (skill.find(start) for start, _ in _PROTECTED_REGIONS)
+        if idx != -1
+    ]
+    return min(positions) if positions else -1
+
+
+def _is_in_protected_region(skill: str, target: str) -> bool:
+    """Check if *target* text falls within any protected region."""
+    if not target:
         return False
     target_idx = skill.find(target)
     if target_idx == -1:
         return False
-    region_end = end_idx + len(SLOW_UPDATE_END)
-    return start_idx <= target_idx < region_end
+    for start_marker, end_marker in _PROTECTED_REGIONS:
+        start_idx = skill.find(start_marker)
+        end_idx = skill.find(end_marker)
+        if start_idx == -1 or end_idx == -1:
+            continue
+        region_end = end_idx + len(end_marker)
+        if start_idx <= target_idx < region_end:
+            return True
+    return False
+
+
+def _is_in_slow_update_region(skill: str, target: str) -> bool:
+    """Backward-compatible alias kept for any external callers/tests."""
+    return _is_in_protected_region(skill, target)
 
 
 def _strip_slow_update_markers(text: str) -> str:
-    """Remove any SLOW_UPDATE markers from edit content to prevent duplication."""
+    """Remove any protected-region markers from edit content to prevent duplication."""
     return (
         text.replace(SLOW_UPDATE_START, "")
             .replace(SLOW_UPDATE_END, "")
+            .replace(APPENDIX_START, "")
+            .replace(APPENDIX_END, "")
     )
 
 
@@ -54,27 +91,27 @@ def _apply_edit_with_report(skill: str, edit: EditType | dict) -> tuple[str, dic
         "status": "unknown",
     }
 
-    if target and _is_in_slow_update_region(skill, target):
-        report["status"] = "skipped_protected_slow_update_region"
+    if target and _is_in_protected_region(skill, target):
+        report["status"] = "skipped_protected_region"
         return skill, report
 
     if op == "append":
-        su_start = skill.find(SLOW_UPDATE_START)
-        if su_start != -1:
-            before = skill[:su_start].rstrip()
-            after = skill[su_start:]
-            report["status"] = "applied_append_before_slow_update"
+        prot_start = _earliest_protected_start(skill)
+        if prot_start != -1:
+            before = skill[:prot_start].rstrip()
+            after = skill[prot_start:]
+            report["status"] = "applied_append_before_protected_region"
             return before + "\n\n" + content + "\n\n" + after, report
         report["status"] = "applied_append"
         return skill.rstrip() + "\n\n" + content + "\n", report
 
     if op == "insert_after":
         if not target or target not in skill:
-            su_start = skill.find(SLOW_UPDATE_START)
-            if su_start != -1:
-                before = skill[:su_start].rstrip()
-                after = skill[su_start:]
-                report["status"] = "applied_insert_after_fallback_before_slow_update"
+            prot_start = _earliest_protected_start(skill)
+            if prot_start != -1:
+                before = skill[:prot_start].rstrip()
+                after = skill[prot_start:]
+                report["status"] = "applied_insert_after_fallback_before_protected_region"
                 return before + "\n\n" + content + "\n\n" + after, report
             report["status"] = "applied_insert_after_fallback_append"
             return skill.rstrip() + "\n\n" + content + "\n", report

skillopt/optimizer/skill_aware.py

@@ -0,0 +1,206 @@
+"""Skill-Aware Reflection — analyst prompt augmentation (EmbodiSkill).
+
+When ``use_skill_aware_reflection`` is enabled, the failure/success analysts are
+asked to additionally classify each reflection by EmbodiSkill type and to route
+**EXECUTION_LAPSE** reflections (the skill rule is correct, the executor just
+failed to follow it) into a separate ``appendix_notes`` list instead of the body
+patch. This module owns:
+
+1. the instruction text appended to the resolved analyst system prompt, and
+2. extraction of ``appendix_notes`` from the analyst JSON response.
+
+Design notes
+------------
+- The suffix is appended **at runtime, gated by the toggle**, so env-specific and
+  generic analyst prompts are augmented uniformly and — when the toggle is off —
+  remain byte-identical to baseline.
+- Discrimination follows the paper / GMemory: ``SKILL_DEFECT`` = the skill rule is
+  wrong / missing / underspecified (→ body edit); ``EXECUTION_LAPSE`` = the rule
+  is valid but the agent didn't follow it (→ appendix reminder, body untouched).
+  **When unsure, default to EXECUTION_LAPSE** (protect the body — never delete a
+  valid rule over a one-off execution slip).
+- Success reflections are labeled DISCOVERY / OPTIMIZATION for logging only; their
+  edit behavior is unchanged.
+"""
+from __future__ import annotations
+
+
+# ── Runtime switch (config-driven, env-independent) ─────────────────────────
+#
+# The trainer calls :func:`configure_skill_aware_reflection` once at startup
+# from the resolved config. ``run_minibatch_reflect`` then picks these values
+# up automatically, so env adapters never need to thread the toggle through —
+# the feature is controlled purely by ``optimizer.use_skill_aware_reflection``
+# regardless of benchmark. Mirrors the ``configure_azure_openai`` pattern in
+# :mod:`skillopt.model`. Explicit kwargs at a call site still take precedence
+# (backward compatible).
+
+_RUNTIME: dict = {"enabled": False, "appendix_source": "both"}
+
+
+def configure_skill_aware_reflection(
+    enabled: bool,
+    appendix_source: str = "both",
+) -> None:
+    """Set the process-wide skill-aware reflection switch from config."""
+    _RUNTIME["enabled"] = bool(enabled)
+    _RUNTIME["appendix_source"] = str(appendix_source or "both")
+
+
+def is_skill_aware_enabled() -> bool:
+    return bool(_RUNTIME["enabled"])
+
+
+def get_skill_aware_appendix_source() -> str:
+    return str(_RUNTIME["appendix_source"])
+
+
+# ── Prompt suffixes ─────────────────────────────────────────────────────────
+
+# Appended to the FAILURE analyst system prompt when the toggle is on.
+ERROR_SUFFIX = """
+
+## Skill-Aware Reflection (EmbodiSkill)
+
+Before proposing body edits, classify EACH failure pattern as one of:
+
+- **SKILL_DEFECT**: the current skill is wrong, missing, or underspecified for
+  this situation — i.e. an agent that *followed the skill* would still fail, or
+  the skill gives no relevant guidance. These become normal body `edits`.
+- **EXECUTION_LAPSE**: the skill ALREADY contains a relevant, correct rule that
+  would have avoided the failure, but the agent did not follow it (e.g. ignored a
+  rule, malformed output, copied the feedback text verbatim, emitted a non-action
+  token like "stop", or otherwise broke execution unrelated to skill content).
+
+Discrimination test: "Is there a rule in the current skill that, if followed,
+prevents this failure?" If yes → EXECUTION_LAPSE. If no (rule absent/wrong) →
+SKILL_DEFECT. **When genuinely unsure, choose EXECUTION_LAPSE** — do not edit or
+delete a valid rule over a one-off execution slip.
+
+Routing:
+- SKILL_DEFECT → put the fix in `patch.edits` (body), as usual.
+- EXECUTION_LAPSE → put a concise reminder in `appendix_notes` (a flat list of
+  strings). DO NOT add a body edit for it. Each note should re-emphasize the
+  existing valid rule the agent failed to follow; it must NOT introduce a new
+  rule. Keep notes short, concrete, and reusable.
+
+Add `appendix_notes` as a TOP-LEVEL key of your JSON output (a sibling of
+`patch`), e.g. `"appendix_notes": ["Follow the existing X rule before Y."]`.
+Use `[]` when there is no execution lapse. Body edits and appendix notes are
+independent: a batch may yield only edits, only notes, both, or neither.
+"""
+
+# Appended to the SUCCESS analyst system prompt when the toggle is on.
+SUCCESS_SUFFIX = """
+
+## Skill-Aware Reflection (EmbodiSkill)
+
+For each proposed edit, optionally label its `reflection_type` for logging:
+- **DISCOVERY**: a useful new rule not yet in the skill (typically an `append`).
+- **OPTIMIZATION**: a better way to perform an existing rule (typically a
+  `replace` of that rule).
+
+This labeling does not change edit behavior. You may also add a top-level
+`appendix_notes` list (flat strings) if a successful trajectory reveals an
+existing valid rule worth re-emphasizing; otherwise use `[]`.
+"""
+
+
+def augment_error_prompt(system_prompt: str) -> str:
+    """Append the failure-analyst skill-aware instruction."""
+    return system_prompt.rstrip() + "\n" + ERROR_SUFFIX
+
+
+def augment_success_prompt(system_prompt: str) -> str:
+    """Append the success-analyst skill-aware instruction."""
+    return system_prompt.rstrip() + "\n" + SUCCESS_SUFFIX
+
+
+# ── Response parsing ────────────────────────────────────────────────────────
+
+
+def extract_appendix_notes(result: dict | None) -> list[str]:
+    """Pull a clean list of appendix-note strings from an analyst JSON result.
+
+    Tolerant of shape: accepts a top-level ``appendix_notes`` list, a single
+    string, or items wrapped in dicts with a ``note``/``content`` field. Returns
+    ``[]`` for anything missing or malformed (so a non-compliant model degrades
+    gracefully to baseline body-only behavior).
+    """
+    if not isinstance(result, dict):
+        return []
+    raw = result.get("appendix_notes")
+    if raw is None:
+        return []
+    if isinstance(raw, str):
+        raw = [raw]
+    if not isinstance(raw, list):
+        return []
+    notes: list[str] = []
+    for item in raw:
+        if isinstance(item, str):
+            text = item.strip()
+        elif isinstance(item, dict):
+            text = str(item.get("note") or item.get("content") or "").strip()
+        else:
+            text = ""
+        if text:
+            notes.append(text)
+    return notes
+
+
+# ── Appendix consolidation (threshold-gated, paper Eq.11 UpdateSkillAppendix) ──
+
+_CONSOLIDATE_SYSTEM = (
+    "You compact the Execution Notes Appendix of an agent skill. Each note "
+    "re-emphasizes an existing skill rule the agent failed to follow. Your job "
+    "is a periodic compaction pass: remove duplicates and redundant overlap, "
+    "merge near-identical reminders into one, and simplify phrasing while keeping "
+    "each note concrete and operational. Do not invent new rules. Preserve the "
+    "distinct actionable content. Return valid JSON only."
+)
+
+
+def consolidate_appendix_notes(
+    notes: list[str],
+    *,
+    chat_fn,
+    max_completion_tokens: int = 4096,
+) -> list[str]:
+    """LLM-consolidate appendix notes: dedupe / merge / compact.
+
+    Mirrors GMemory ``_maybe_refactor_execution_notes`` and paper Eq.11. ``chat_fn``
+    is the optimizer chat callable ``(system, user, max_completion_tokens, retries,
+    stage) -> (text, meta)``. On ANY failure (parse, empty, exception) the original
+    notes are returned unchanged, so consolidation can never lose the appendix.
+    """
+    from skillopt.utils import extract_json  # local import to avoid cycles
+
+    clean = [str(n).strip() for n in (notes or []) if str(n).strip()]
+    if len(clean) < 2:
+        return clean
+
+    numbered = "\n".join(f"{i}. {n}" for i, n in enumerate(clean, 1))
+    user = (
+        f"## Current Execution Notes ({len(clean)} total)\n{numbered}\n\n"
+        "Compact these into a shorter list without losing distinct actionable "
+        "information. Merge duplicates and near-duplicates; keep each note short, "
+        "concrete, and reusable. Return valid JSON only with this schema:\n"
+        '{ "appendix_notes": ["compacted note 1", "compacted note 2"] }'
+    )
+    try:
+        response, _ = chat_fn(
+            system=_CONSOLIDATE_SYSTEM,
+            user=user,
+            max_completion_tokens=max_completion_tokens,
+            retries=2,
+            stage="appendix_consolidate",
+        )
+        result = extract_json(response)
+        compacted = extract_appendix_notes(result)
+        # Guard: only accept a non-empty result that actually shrinks the set.
+        if compacted and len(compacted) <= len(clean):
+            return compacted
+    except Exception:  # noqa: BLE001
+        pass
+    return clean

tests/test_skill_aware_reflection.py

@@ -0,0 +1,274 @@
+"""Standalone regression + function tests for skill-aware reflection.
+
+Run directly (no pytest needed):
+    python tests/test_skill_aware_reflection.py
+
+Covers:
+1. Toggle-OFF byte-identical guarantee for skill.py edit application
+   (slow-update-only behavior must be unchanged).
+2. Appendix module: inject / append / dedup / extract / accumulate.
+3. Appendix-region protection from step-level edits.
+4. Coexistence of appendix + slow_update regions.
+5. reflect.py prompt augmentation + appendix_notes parsing (no LLM call).
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+# Ensure THIS repo's skillopt is imported (not an installed copy) when the
+# file is run directly: script mode puts tests/ on sys.path, not the repo root.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+def _reference_old_apply(skill: str, edit: dict) -> str:
+    """Reproduce the ORIGINAL slow-update-only edit behavior inline."""
+    SU_START = "<!-- SLOW_UPDATE_START -->"
+    SU_END = "<!-- SLOW_UPDATE_END -->"
+    op = edit.get("op", "")
+    content = edit.get("content", "").strip().replace(SU_START, "").replace(SU_END, "")
+    target = edit.get("target", "")
+    si = skill.find(SU_START)
+    ei = skill.find(SU_END)
+
+    def in_su(t: str) -> bool:
+        if si == -1 or ei == -1:
+            return False
+        ti = skill.find(t)
+        if ti == -1:
+            return False
+        return si <= ti < ei + len(SU_END)
+
+    if target and in_su(target):
+        return skill
+    if op == "append":
+        s = skill.find(SU_START)
+        if s != -1:
+            return skill[:s].rstrip() + "\n\n" + content + "\n\n" + skill[s:]
+        return skill.rstrip() + "\n\n" + content + "\n"
+    if op == "insert_after":
+        if not target or target not in skill:
+            s = skill.find(SU_START)
+            if s != -1:
+                return skill[:s].rstrip() + "\n\n" + content + "\n\n" + skill[s:]
+            return skill.rstrip() + "\n\n" + content + "\n"
+        idx = skill.index(target) + len(target)
+        nl = skill.find("\n", idx)
+        at = nl + 1 if nl != -1 else len(skill)
+        return skill[:at] + "\n" + content + "\n" + skill[at:]
+    if op == "replace":
+        if not target or target not in skill:
+            return skill
+        return skill.replace(target, content, 1)
+    if op == "delete":
+        if not target or target not in skill:
+            return skill
+        return skill.replace(target, "", 1)
+    return skill
+
+
+def test_toggle_off_byte_identical() -> None:
+    from skillopt.optimizer.skill import _apply_edit_with_report
+
+    SU_START = "<!-- SLOW_UPDATE_START -->"
+    SU_END = "<!-- SLOW_UPDATE_END -->"
+    skill = (
+        "# QA Skill\n\n## Rules\n- Prefer shortest answer span.\n"
+        "- Use clue wording to constrain answer type.\n\n"
+        f"{SU_START}\nSome slow update guidance here.\n{SU_END}\n"
+    )
+    edits = [
+        {"op": "append", "content": "- New rule appended."},
+        {"op": "insert_after", "target": "## Rules", "content": "- Inserted rule."},
+        {"op": "insert_after", "target": "NONEXISTENT", "content": "- Fallback rule."},
+        {"op": "replace", "target": "Prefer shortest answer span.", "content": "Prefer the exact minimal span."},
+        {"op": "delete", "target": "- Use clue wording to constrain answer type."},
+        {"op": "replace", "target": "Some slow update guidance here.", "content": "HACKED"},
+        {"op": "delete", "target": "Some slow update guidance here."},
+    ]
+    for e in edits:
+        new_skill, _ = _apply_edit_with_report(skill, e)
+        old_skill = _reference_old_apply(skill, e)
+        assert new_skill == old_skill, f"byte mismatch for {e['op']}"
+    print("PASS  test_toggle_off_byte_identical")
+
+
+def test_appendix_module() -> None:
+    from skillopt.optimizer.appendix import (
+        has_appendix_field, inject_empty_appendix_field,
+        extract_appendix_notes, append_to_appendix_field, APPENDIX_START,
+    )
+    skill = "# QA Skill\n\n- Prefer shortest answer span."
+    s1 = inject_empty_appendix_field(skill)
+    assert has_appendix_field(s1) and extract_appendix_notes(s1) == []
+    assert inject_empty_appendix_field(s1) == s1  # idempotent
+    s2 = append_to_appendix_field(s1, ["Go to fridge for ice water.", "No stop token."])
+    assert extract_appendix_notes(s2) == ["Go to fridge for ice water.", "No stop token."]
+    s3 = append_to_appendix_field(s2, ["go to fridge for ice water", "Check sheet range."])
+    assert extract_appendix_notes(s3) == [
+        "Go to fridge for ice water.", "No stop token.", "Check sheet range.",
+    ], "near-duplicate must be dropped"
+    assert s3.count(APPENDIX_START) == 1, "exactly one region after accumulation"
+    assert "# QA Skill" in s3 and "Prefer shortest answer span" in s3
+    assert extract_appendix_notes(append_to_appendix_field(s1, ["  ", "", "real"])) == ["real"]
+    print("PASS  test_appendix_module")
+
+
+def test_appendix_protection() -> None:
+    from skillopt.optimizer.skill import _apply_edit_with_report
+    from skillopt.optimizer.appendix import append_to_appendix_field, inject_empty_appendix_field
+
+    skill = inject_empty_appendix_field("# QA Skill\n\n- Rule one.")
+    skill = append_to_appendix_field(skill, ["Follow rule one before acting."])
+    for e in (
+        {"op": "delete", "target": "Follow rule one before acting."},
+        {"op": "replace", "target": "Follow rule one before acting.", "content": "HACK"},
+    ):
+        new, rep = _apply_edit_with_report(skill, e)
+        assert new == skill, f"appendix must be protected from {e['op']}"
+        assert rep["status"] == "skipped_protected_region"
+    new, rep = _apply_edit_with_report(skill, {"op": "replace", "target": "Rule one.", "content": "Rule 1."})
+    assert "Rule 1." in new and "Follow rule one before acting." in new
+    print("PASS  test_appendix_protection")
+
+
+def test_coexistence_with_slow_update() -> None:
+    from skillopt.optimizer.skill import _apply_edit_with_report
+    from skillopt.optimizer.appendix import (
+        inject_empty_appendix_field, append_to_appendix_field, extract_appendix_notes,
+    )
+    from skillopt.optimizer.slow_update import (
+        inject_empty_slow_update_field, replace_slow_update_field, extract_slow_update_field,
+    )
+    skill = inject_empty_appendix_field("# QA Skill\n\n- Rule one.")
+    skill = append_to_appendix_field(skill, ["Follow rule one."])
+    skill = inject_empty_slow_update_field(skill)
+    skill = replace_slow_update_field(skill, "Slow guidance v2.")
+    assert extract_appendix_notes(skill) == ["Follow rule one."]
+    assert extract_slow_update_field(skill) == "Slow guidance v2."
+    # both regions protected
+    n1, r1 = _apply_edit_with_report(skill, {"op": "delete", "target": "Follow rule one."})
+    n2, r2 = _apply_edit_with_report(skill, {"op": "replace", "target": "Slow guidance v2.", "content": "X"})
+    assert n1 == skill and n2 == skill
+    # append lands before both regions (body stays at top)
+    n3, _ = _apply_edit_with_report(skill, {"op": "append", "content": "- Rule two."})
+    assert n3.find("- Rule two.") < n3.find("<!-- APPENDIX_START -->")
+    assert n3.find("- Rule two.") < n3.find("<!-- SLOW_UPDATE_START -->")
+    print("PASS  test_coexistence_with_slow_update")
+
+
+def test_reflect_parsing_and_augment() -> None:
+    import inspect
+    import skillopt.gradient.reflect as R
+    from skillopt.optimizer.skill_aware import extract_appendix_notes, augment_error_prompt
+
+    for fn in ("run_error_analyst_minibatch", "run_success_analyst_minibatch"):
+        sig = inspect.signature(getattr(R, fn))
+        assert "skill_aware_reflection" in sig.parameters
+        assert sig.parameters["skill_aware_reflection"].default is False, f"{fn} default must be False"
+    # run_minibatch_reflect uses a None sentinel: explicit kwarg wins, else the
+    # process-wide config switch (configure_skill_aware_reflection) decides.
+    sig = inspect.signature(R.run_minibatch_reflect)
+    assert sig.parameters["skill_aware_reflection"].default is None
+    assert sig.parameters["skill_aware_appendix_source"].default is None
+    assert extract_appendix_notes({"appendix_notes": ["a", "b"]}) == ["a", "b"]
+    assert extract_appendix_notes({"appendix_notes": "x"}) == ["x"]
+    assert extract_appendix_notes({"appendix_notes": [{"note": "n"}, {"content": "c"}, {}]}) == ["n", "c"]
+    assert extract_appendix_notes({}) == [] and extract_appendix_notes(None) == []
+    aug = augment_error_prompt("ORIG")
+    assert aug.startswith("ORIG") and "SKILL_DEFECT" in aug and "EXECUTION_LAPSE" in aug
+    print("PASS  test_reflect_parsing_and_augment")
+
+
+def test_global_switch_env_independent() -> None:
+    """The config switch alone must drive SAR for ANY env adapter (no kwargs)."""
+    from unittest import mock
+    import skillopt.gradient.reflect as R
+    from skillopt.optimizer.skill_aware import (
+        configure_skill_aware_reflection,
+        get_skill_aware_appendix_source,
+        is_skill_aware_enabled,
+    )
+
+    # configure() round-trip.
+    configure_skill_aware_reflection(True, "failure_only")
+    assert is_skill_aware_enabled() and get_skill_aware_appendix_source() == "failure_only"
+    configure_skill_aware_reflection(False)
+    assert not is_skill_aware_enabled() and get_skill_aware_appendix_source() == "both"
+
+    # run_minibatch_reflect with NO skill-aware kwargs (adapter-style call):
+    # capture what it forwards to the analyst workers under each switch state.
+    import tempfile
+    captured: dict = {}
+
+    def fake_error_analyst(*args, **kwargs):
+        captured["skill_aware_reflection"] = kwargs.get("skill_aware_reflection")
+        return None
+
+    def run_once() -> None:
+        captured.clear()
+        with mock.patch.object(R, "run_error_analyst_minibatch", fake_error_analyst), \
+             tempfile.TemporaryDirectory() as tmp:
+            R.run_minibatch_reflect(
+                results=[{"id": "t1", "hard": 0, "soft": 0.0}],
+                skill_content="# Skill",
+                prediction_dir=tmp,
+                patches_dir=tmp,
+                workers=1,
+                failure_only=True,
+                minibatch_size=8,
+            )
+
+    try:
+        configure_skill_aware_reflection(True, "both")
+        run_once()
+        assert captured.get("skill_aware_reflection") is True, \
+            "switch ON must reach the analyst without adapter wiring"
+
+        configure_skill_aware_reflection(False)
+        run_once()
+        assert captured.get("skill_aware_reflection") is False, \
+            "switch OFF must keep the analyst at baseline"
+
+        # Explicit kwarg still overrides the global switch (backward compat).
+        captured.clear()
+        with mock.patch.object(R, "run_error_analyst_minibatch", fake_error_analyst), \
+             tempfile.TemporaryDirectory() as tmp:
+            R.run_minibatch_reflect(
+                results=[{"id": "t1", "hard": 0, "soft": 0.0}],
+                skill_content="# Skill",
+                prediction_dir=tmp,
+                patches_dir=tmp,
+                workers=1,
+                failure_only=True,
+                minibatch_size=8,
+                skill_aware_reflection=True,
+            )
+        assert captured.get("skill_aware_reflection") is True
+    finally:
+        configure_skill_aware_reflection(False)
+    print("PASS  test_global_switch_env_independent")
+
+
+def main() -> int:
+    tests = [
+        test_toggle_off_byte_identical,
+        test_appendix_module,
+        test_appendix_protection,
+        test_coexistence_with_slow_update,
+        test_reflect_parsing_and_augment,
+        test_global_switch_env_independent,
+    ]
+    failed = 0
+    for t in tests:
+        try:
+            t()
+        except AssertionError as exc:
+            failed += 1
+            print(f"FAIL  {t.__name__}: {exc}")
+    print(f"\n{len(tests) - failed}/{len(tests)} passed")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())