# SkillOpt — concepts

Load this when you need to change `skillopt_loop` or reason about its rollout
cost. To *run* the algorithm, `SKILL.md` is enough.

SkillOpt (arXiv:2605.23904, *Executive Strategy for Self-Evolving Agent Skills*)
is a single-lineage capability optimizer: like `hill-climb` the parent is always
the current best, but the run is organized into epochs over mini-batches with a
decaying integer edit budget and one extra consolidation step per epoch. It edits
whatever the selected capability owns; nothing in the loop assumes a skill
package, despite the name.

## The loop

`cap_evolve.skillopt.skillopt_loop(adapter, *, run_dir, optimizer, current_val, …)`:

1. Init memory + version store (`harness._init_memory_store`). Compute
   `steps_per_epoch = ceil(len(train) / (batch_size · accumulation))`,
   `total_steps = epochs · steps_per_epoch`, and the integer edit-budget schedule
   `build_schedule(lr_schedule, max=edit_budget, min=min_edit_budget, total=total_steps)`
   (default `cosine`, 4 → 2).
2. **Each epoch**: shuffle the train ids (`random.Random(1000 + epoch)`, so a
   rerun reproduces), reset the per-epoch `step_buffer` and `rejected_this_epoch`,
   and record the epoch-start candidate id.
3. **Each step**: take the accumulation window of the shuffled order as the
   mini-batch and build the instruction string —
   `harness._focus_instructions(current_val, focus_ids=minibatch_ids, label)` plus
   the SkillOpt block (`L`, the rejected ids to avoid this epoch, the unsolved
   failure patterns). **Both the focus filter and the failure-pattern filter are
   currently empty by construction — see the gap below.** Parent is always the
   current best. `harness.run_step(...)` does the rest: materialize, optimize,
   evaluate on val, apply the gate, snapshot + set best on accept, write
   RejectedMemory/History.
4. Append a bounded record to `step_buffer`
   (`{step, epoch, accepted, n_fail, failure_patterns, rejected id + val Δ}`),
   capped at ≤3 task ids per pattern, ≤10 patterns, ≤12 steps
   (`_MAX_*` in `skillopt.py:58-60`), and reset each epoch so the prompt cannot
   balloon. Note `_MAX_BUFFER_STEPS` — a *step* cap — is reused to slice the
   *reject* list at `skillopt.py:122`.
5. Update `current_val` only on accept.
6. **End of epoch** (from epoch 2, unless `--no-slow-update`): the gated
   consolidation step below.
7. Return a result dict shaped like `hill_climb_loop`'s, plus `epochs`,
   `edit_budget_schedule`, `epoch_stats` and `slow_updates`.

### Gap: the mini-batch is not actually in focus (issue #371)

`minibatch_ids` come from `run_dir.read_splits().train` (`skillopt.py:230`) but
are used to filter the parent's **val** per-task rows (`skillopt.py:291`,
`:315`, `:319` → `harness.py:2011-2012`). `make_splits` assigns train/val/test as
disjoint slices of one shuffled list (`splits.py:117-119`), so every filter
yields nothing. Observed on `SyntheticAdapter(n=12)`:

```
train: ['t1','t9','t8','t5','t10','t2']   val: ['t3','t7','t4']   train ∩ val: set()
Focus: epoch 1/1 step 1/3 (mini-batch of 2 train tasks, L=4).
Current val reward 0.000: 0 solid / 0 flaky / 0 failing of 0 tasks.
'## Failure patterns still unsolved' ever rendered? False
```

So `n_fail` is always 0 and step 4's `failure_patterns` is always `[]`. Fixing
this means either evaluating the mini-batch on **train** and reflecting on that
result (gepa's `_eval_minibatch` exists for exactly this) or deriving the focus
ids from val. The two readings imply different rollout costs; #371 is the
decision.

## Textual learning rate (edit budget)

`core/cap_evolve/lr_schedule.py`. Integers only — you cannot make 2.7 edits —
clamped to `[min_lr, max_lr]`, `total_steps <= 0` yields `[]`, and `constant` or
`total_steps == 1` sits at `max_lr` (`lr_schedule.py:42-55`). Worked values,
`max=4 min=2 total=12`:

```
constant  [4,4,4,4,4,4,4,4,4,4,4,4]
linear    [4,4,4,3,3,3,3,3,3,2,2,2]
cosine    [4,4,4,4,3,3,3,3,2,2,2,2]
```

`linear` and `cosine` differ at 2 of 12 positions. Over a 3-value integer range
the schedule choice is close to a no-op; `SKILL.md` § "The textual learning rate"
argues why the decay is a heuristic rather than a demonstrated mechanism.

`L` reaches the optimizer as prose only; nothing clips the edit count. The
`requested_edits` / `applied_changes` pair logged at `skillopt.py:338` is not a
guardrail: `_changed_components` (`:398-427`) counts files whose bytes differ,
and `applied_changes` is read by nothing in `core/`, `dashboard/` or `skills/`.

## The within-epoch buffer

Two per-epoch bounded structures appended to the next step's prompt:

- **rejected-edit list** — each rejected candidate's id and val Δ
  (`skillopt.py:120-125`). It does not carry the *content* of the rejected edit,
  so an optimizer cannot avoid an approach it was never shown. The `LEDGER.md`
  that `run_step` injects into every workdir already lists each prior edit's
  outcome and the tasks it broke and fixed, run-global rather than per-epoch —
  strictly stronger. Treat this list as redundant.
- **failure-pattern block** — failing feedback clustered by a normalized 8-word
  prefix, infra-errored tasks dropped via `raw.errored` (`skillopt.py:65-93`).
  Sound logic, currently fed an empty list (see the gap above).

## The epoch-boundary consolidation step

From epoch 2, compare the epoch-start candidate against the current best and
bucket each task:

- **regressed** — passed at epoch start, now failing;
- **persistent_fail** — failing both times;
- **stable_success** — passing both times;
- **improved** — reward rose. Computed with a bare `if` before the exclusive
  chain (`skillopt.py:184-191`), so a 0.2 → 0.5 task lands in both `improved` and
  `persistent_fail`; `_slow_update_instructions` (`:139-166`) renders only the
  first three. Read `improved` as an overlapping counter, not a partition member.

The longitudinal instruction ("fix the REGRESSIONS and chip at the PERSISTENT
failures without breaking any STABLE SUCCESS") goes through one ordinary
`harness.run_step` (`skillopt.py:479-485`) — same val gate, never force-accepted.
`skills/algorithms/skillopt/scripts/check.py:108-112` asserts the step carries a
gate decision, which is the regression guard on that property.

### Gap: the cost and the comparison

- The re-evaluation calls `harness.evaluate_candidate(..., split="train")` twice
  with **no** `ids=` (`skillopt.py:458-463`) and only then filters to the ~20
  sampled ids (`:464-466`). Budget `2 · len(train) · n_trials` rollouts per
  epoch boundary, not `2 · sample · n_trials`.
- The re-eval is guarded by `prev_epoch_best_id != run_dir.best_id`
  (`skillopt.py:455`). If the epoch accepted nothing, both sides remain
  `current_val` — the identical list — so the buckets are computed over **val**
  tasks and report 0 regressed / 0 improved, while the logged `sample=` is a
  train count that was never scored (`:469-470`). The consolidation step runs
  anyway, on an instruction describing no actual change.
- `skillopt.py:147` writes "the skill" into that prompt for every capability
  type. An algorithm must not name a capability it cannot know.

## Where it lives

- Loop: `core/cap_evolve/skillopt.py` (`skillopt_loop`).
- Schedule: `core/cap_evolve/lr_schedule.py` (`build_schedule`).
- The shared step: `core/cap_evolve/harness.py` (`run_step`,
  `evaluate_candidate`, `_focus_instructions`, `_init_memory_store`) — its
  contract is documented by `algorithms/hill-climb`.

"""skillopt has no per-skill abstract methods beyond the project adapter.

Like hill-climb, it composes the contract methods (tasks/run_target/score/
materialize) via the shared harness; the optimizer skill supplies the proposer
and the capability skill owns the editable surface. The only "policy" this
algorithm carries is its schedule defaults (the textual learning rate), so
``check.py`` verifies the epoch/step loop + schedule + buffer + gated slow-update
behaviorally rather than asserting an implementation here.
"""

from __future__ import annotations

from pathlib import Path

# Defaults for the textual learning rate (integer edit budget) and the loop shape.
DEFAULT_POLICY = {
    "epochs": 4,
    "edit_budget": 4,
    "min_edit_budget": 2,
    "lr_schedule": "cosine",
    "slow_update": True,
}


def materialize(capability_dir: Path) -> dict:  # noqa: ARG001
    return {}

"""Behavioral contract for skillopt.

Drives the real ``skillopt_loop`` end-to-end with the offline MOCK optimizer on a
tiny synthetic adapter (zero API) and asserts the SkillOpt mechanics actually
hold — not merely that ``run.py`` imports:

  * the epoch/step loop runs end-to-end (epochs × mini-batches produce steps);
  * the edit-budget (textual learning rate) schedule DECAYS (cosine: start>end);
  * the within-epoch rejected-edit buffer is populated AND bounded;
  * the epoch-boundary slow update is GATED on val (it appears as a normal,
    gate-decided step — never force-accepted);
  * the test split is NEVER consumed (seal stays unused).
"""

from __future__ import annotations

import shutil
import sys
import tempfile
from pathlib import Path

import _bootstrap  # noqa: F401

from cap_evolve.skillcheck import (
    Checker, import_run, make_mock_optimizer, SyntheticAdapter, seed_capability_dir,
)


def main() -> int:
    c = Checker("skillopt")
    run = import_run()
    c.require_main(run)

    from cap_evolve import skillopt
    from cap_evolve import RunDir, Budget, harness
    from cap_evolve.lr_schedule import build_schedule

    c.check(hasattr(skillopt, "skillopt_loop"), "core missing skillopt.skillopt_loop")
    c.check(set(run.SCHEDULES) == {"constant", "linear", "cosine"},
            f"unexpected schedules: {run.SCHEDULES}", note=f"schedules: {run.SCHEDULES}")

    # The textual learning rate (edit budget) decays under cosine.
    sched = build_schedule("cosine", max_lr=4, min_lr=2, total_steps=8)
    c.check(sched and sched[0] > sched[-1],
            f"edit-budget schedule did not decay: {sched}",
            note=f"edit-budget anneal (cosine 4->2 over 8): {sched}")

    tmp = Path(tempfile.mkdtemp(prefix="skillopt_chk_"))
    try:
        adapter = SyntheticAdapter(n=8)
        seed = seed_capability_dir(tmp, level=0)
        run_dir = RunDir.create(tmp / ".capevolve", ts="chk",
                                budget=Budget(max_iterations=50))
        harness.ensure_splits(adapter, run_dir, seed=0)
        base = harness.baseline(adapter, seed, run_dir=run_dir)

        # 2 epochs so the epoch-2 boundary triggers a slow update; small batch so we
        # get multiple steps per epoch (exercising the buffer).
        result = skillopt.skillopt_loop(
            adapter, run_dir=run_dir, optimizer=make_mock_optimizer(bump=1),
            current_val=base, epochs=2, batch_size=2, accumulation=1,
            edit_budget=4, min_edit_budget=2, lr_schedule="cosine",
            gate_kwargs={"mode": "significant", "k_se": 1.0},
            slow_update=True, slow_update_sample=4, store=None,
        )

        # 1) the epoch/step loop ran end to end
        c.check(len(result["steps"]) >= 2 and result["epochs"] == 2,
                f"loop did not run epochs×steps: {result.get('epochs')}, "
                f"{len(result.get('steps', []))} steps",
                note=f"ran {result['epochs']} epochs, {len(result['steps'])} steps, "
                     f"{result['accepts']} accepts")
        c.check(result["best_val"] >= base.reward,
                f"best_val regressed below baseline: {result['best_val']} < {base.reward}")

        # 2) the schedule in the result decays (cosine over the run)
        rs = result["edit_budget_schedule"]
        c.check(rs and rs[0] >= rs[-1] and min(rs) >= 2,
                f"result edit-budget schedule wrong: {rs}")

        # 3) the rejected-edit buffer was populated AND bounded. We assert this from
        # the events: skillopt_step events carry accept; once the synthetic adapter
        # plateaus (all 8 tasks solved at level 8), further steps reject — those feed
        # the per-epoch buffer. We re-run a longer single-epoch loop that plateaus.
        run_dir2 = RunDir.create(tmp / ".capevolve", ts="chk2", budget=Budget(max_iterations=50))
        harness.ensure_splits(adapter, run_dir2, seed=0)
        base2 = harness.baseline(adapter, seed, run_dir=run_dir2)
        result2 = skillopt.skillopt_loop(
            adapter, run_dir=run_dir2, optimizer=make_mock_optimizer(bump=0),  # never improves
            current_val=base2, epochs=1, batch_size=2, accumulation=1,
            edit_budget=4, min_edit_budget=2, lr_schedule="cosine",
            gate_kwargs={"mode": "significant", "k_se": 1.0}, slow_update=False, store=None,
        )
        rejects = sum(1 for s in result2["steps"] if not s["accepted"])
        c.check(rejects >= 1, "a non-improving optimizer produced no rejects to buffer",
                note=f"non-improving optimizer → {rejects} rejected edits buffered")
        # bounded: the module caps the buffer; assert the cap constant is finite+small
        c.check(0 < skillopt._MAX_BUFFER_STEPS <= 50,
                f"buffer cap unreasonable: {skillopt._MAX_BUFFER_STEPS}",
                note=f"per-epoch buffer bounded to {skillopt._MAX_BUFFER_STEPS} steps")

        # 4) the slow update is GATED on val (appeared as a gate-decided step with a
        # decision, not a force-accept). Find it in the first run's slow_updates.
        su = result["slow_updates"]
        c.check(len(su) >= 1, "no slow update ran at the epoch-2 boundary",
                note=f"slow updates: {su}")
        if su:
            slow_step = next((s for s in result["steps"]
                              if s.get("step_in_epoch") == "slow"), None)
            c.check(slow_step is not None and "decision" in slow_step,
                    "slow-update step missing a gate decision (force-accepted?)",
                    note="slow update is gated on val (carries a gate decision)")

        # 5) test was NEVER consumed
        splits = run_dir.read_splits()
        c.check(not splits.test_used, "skillopt consumed the sealed test split",
                note="test split sealed throughout (never consumed)")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    return c.emit()


if __name__ == "__main__":
    sys.exit(main())

"""skillopt — a disciplined single-lineage climber with a textual learning rate.

SkillOpt (arXiv:2605.23904) runs epochs × mini-batches: each step focuses the
optimizer on one mini-batch of train tasks under a shrinking integer **edit
budget** L (the textual learning rate, on a ``constant|linear|cosine`` schedule),
keeps a within-epoch rejected-edit + failure-pattern buffer in the prompt, and
ends each epoch with ONE extra *gated* slow/meta update that fixes longitudinal
regressions. Parent is always the current best (single lineage). Gated on val;
test sealed.

Thin wrapper over ``cap_evolve.skillopt.skillopt_loop``.
"""

from __future__ import annotations

import argparse
import json
import shlex
import sys
from pathlib import Path

import _bootstrap  # noqa: F401

from cap_evolve import RunDir, harness, skillopt
from cap_evolve.check import load_adapter
from cap_evolve.loop import SplitResult
from cap_evolve.store import make_store

ALGO = "skillopt"
SCHEDULES = ("constant", "linear", "cosine")


def main(argv=None) -> int:
    p = argparse.ArgumentParser(prog=ALGO)
    p.add_argument("--run-dir", required=True)
    p.add_argument("--project", required=True)
    p.add_argument("--optimizer", required=True, help="optimizer cmd with {workdir} {prompt}")
    p.add_argument("--epochs", type=int, default=4)
    # Accepted (and ignored) for `cap-evolve run` compatibility: the generic
    # sequencer passes --max-iterations to every algorithm, but skillopt is
    # epoch-driven — set --epochs to control the step count.
    p.add_argument("--max-iterations", type=int, default=0,
                   help="IGNORED — this loop is epoch-driven; use --epochs/--batch-size")
    p.add_argument("--batch-size", type=int, default=None,
                   help="train tasks per mini-batch (default min(8, len(train)))")
    p.add_argument("--accumulation", type=int, default=1,
                   help="mini-batches accumulated per step (default 1)")
    # the textual learning rate = integer edit budget; --lr is an alias
    p.add_argument("--edit-budget", "--lr", dest="edit_budget", type=int, default=4,
                   help="max edits per step at the start (textual learning rate)")
    p.add_argument("--min-edit-budget", type=int, default=2,
                   help="edit budget floor at the end of the schedule")
    p.add_argument("--lr-schedule", default="cosine", choices=SCHEDULES,
                   help="how the edit budget decays over the run")
    p.add_argument("--n-trials", type=int, default=1)
    p.add_argument("--workers", type=int, default=1,
                   help="concurrent rollouts per evaluation (1 = serial, the default). "
                        "Only safe when the adapter's run_target is thread-safe.")
    p.add_argument("--gate-mode", default="auto",
                   help="auto = let the engine pick the paired gate (recommended; candidate & current share val tasks); or significant|paired|strict|threshold")
    p.add_argument("--k-se", type=float, default=1.0)
    su = p.add_mutually_exclusive_group()
    su.add_argument("--slow-update", dest="slow_update", action="store_true", default=True,
                    help="run the gated epoch-boundary slow/meta update (default on)")
    su.add_argument("--no-slow-update", dest="slow_update", action="store_false",
                    help="disable the slow update")
    p.add_argument("--slow-update-sample", type=int, default=20,
                   help="train ids sampled for the longitudinal slow-update compare")
    p.add_argument("--no-regression", action="store_true",
                   help="reject candidates that break a passing val task")
    p.add_argument("--store", default="git", help="git|copy|command")
    p.add_argument("--store-commit-cmd", default=None)
    p.add_argument("--resume", action="store_true",
                   help="continue from the run's current best instead of baseline")
    # The shared optimizer read-context flags — identical set to hill-climb/gepa.
    harness.OptimizerContext.add_arguments(p)
    p.add_argument("--protected-paths", default="",
                   help="comma-separated globs sealing the eval surface (scorer/gold/tasks/"
                        "tests). 'default' expands to the built-in set. Empty = off. A "
                        "candidate that edits one is INDECISIVE, not scored 0.0.")
    args = p.parse_args(argv)

    run_dir = RunDir.open(Path(args.run_dir))
    # Process-wide rollout concurrency for every evaluation this loop runs.
    harness.DEFAULT_WORKERS = max(1, args.workers)

    try:
        from capevolve_telemetry import load_observers_from_state
        for obs in load_observers_from_state(run_dir.load_observer_state()):
            run_dir.add_observer(obs)
    except Exception:  # noqa: BLE001
        pass

    if harness.DEFAULT_WORKERS > 1:
        run_dir.log_event("parallel", workers=harness.DEFAULT_WORKERS, algorithm=ALGO)
    store = make_store({"store": args.store, "store_commit_cmd": args.store_commit_cmd}, run_dir.root)
    adapter = load_adapter(Path(args.project))
    optimizer = harness.optimizer_from_command(shlex.split(args.optimizer))
    if args.resume and run_dir.best_id:
        current_val = harness.split_result_from_rollouts(run_dir, run_dir.best_id, "val")
    else:
        current_val = SplitResult.from_dict(
            json.loads((run_dir.root / "baseline.json").read_text())["val"])

    result = skillopt.skillopt_loop(
        adapter, run_dir=run_dir, optimizer=optimizer, current_val=current_val,
        epochs=args.epochs, batch_size=args.batch_size, accumulation=args.accumulation,
        edit_budget=args.edit_budget, min_edit_budget=args.min_edit_budget,
        lr_schedule=args.lr_schedule, n_trials=args.n_trials,
        gate_kwargs=({"k_se": args.k_se} if args.gate_mode == "auto"
                     else {"mode": args.gate_mode, "k_se": args.k_se}),
        no_regression=args.no_regression, slow_update=args.slow_update,
        slow_update_sample=args.slow_update_sample, algorithm=ALGO, store=store,
        ctx=harness.OptimizerContext.from_args(args, run_dir=run_dir),
        protected_patterns=harness.parse_protected_paths(args.protected_paths),
    )
    run_dir.close_observers()

    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Thin shim: locate cap_evolve, then defer to cap_evolve._bootstrap.

Skill scripts ``import _bootstrap`` first. The real path-resolution logic lives
ONCE in ``cap_evolve._bootstrap`` (so it can't drift across skills); this shim
only has to find that package, which means a minimal upward walk for ``core/`` —
the single bit of bootstrapping that genuinely must run before cap_evolve is
importable. Everything else delegates.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path


def _seed_path() -> None:
    """Minimal: put a dir containing the cap_evolve package on sys.path.

    ``CAPEVOLVE_CORE`` is honoured BEFORE any ambient import. An editable install of a
    *different* cap-evolve checkout registers a ``sys.meta_path`` finder, which outranks
    both ``sys.path`` and ``PYTHONPATH`` — so "cap_evolve imports fine" is not evidence that
    it imports the checkout you are standing in. Deferring to the ambient package here made
    an explicit override unreachable, and the symptom was a stale core silently answering
    for this one (``ModuleNotFoundError: cap_evolve.constraints`` from a checkout that
    predates that module). An explicit env var wins.
    """
    env = os.environ.get("CAPEVOLVE_CORE")
    want = Path(env).resolve() if env else None
    if want and (want / "cap_evolve" / "__init__.py").exists():
        loaded = sys.modules.get("cap_evolve")
        already = getattr(loaded, "__file__", None)
        if already and Path(already).resolve().parent.parent == want:
            return                      # right checkout already imported: touch nothing
        p = str(want)
        if p in sys.path:
            sys.path.remove(p)
        sys.path.insert(0, p)
        if loaded is not None:
            # Evicting a module makes a re-import yield a DIFFERENT object, so anything
            # already holding a reference fails an `is` check. Only ever do it when the
            # loaded package really is the wrong checkout — otherwise this "fix" becomes
            # the bug (it broke two identity assertions in core/tests exactly once).
            for name in [m for m in sys.modules
                         if m == "cap_evolve" or m.startswith("cap_evolve.")]:
                sys.modules.pop(name, None)
        for finder in list(sys.meta_path):
            if "cap_evolve" in getattr(finder, "MAPPING", {}):
                sys.meta_path.remove(finder)
        return
    # A checkout's own core outranks an ambient install. Without this, a skill script run
    # from checkout X silently executed against checkout Y's cap_evolve (an editable install
    # registers a sys.meta_path finder, which outranks sys.path), and the only symptom was
    # missing modules — or, worse, a green result measured against the wrong tree.
    here = Path(__file__).resolve()
    own = next((p / "core" for p in here.parents
                if (p / "core" / "cap_evolve" / "__init__.py").exists()), None)
    if own is not None:
        os.environ.setdefault("CAPEVOLVE_CORE", str(own))
        return _seed_path()
    try:
        import cap_evolve  # noqa: F401
        return
    except Exception:
        pass
    cands = []
    for parent in here.parents:
        cands.append(parent / "core")
        cands.append(parent)
    for c in cands:
        if (c / "cap_evolve" / "__init__.py").exists():
            p = str(c)
            if p not in sys.path:
                sys.path.insert(0, p)
            return


_seed_path()
from cap_evolve._bootstrap import ensure_core  # noqa: E402

# Anchor the upward walk at THIS skill script's location (not the core module's).
ensure_core(Path(__file__).resolve())