# Concepts — orchestrating the pipeline honestly

> orchestrate adds no measurement logic; it sequences the phases and enforces,
> automatically, the guardrails that keep a run honest. This note maps the
> pipeline as a dependency graph and shows where each guardrail lives.
> Implementation: this skill's `scripts/run.py` (`needs`/`provides` resolution).

## The pipeline is a needs/provides DAG

Each phase declares what tokens it `needs` and `provides`. orchestrate orders the
phases so every `needs` is satisfied by an upstream `provides`, which is also how a
misordered or incompatible pipeline is caught *before* it runs:

```
intake            provides: project, tasks
  → implement-and-check   needs: project        provides: checked
    → baseline            needs: project, tasks  provides: splits, baseline, candidate
      → <algorithm>       needs: candidate, ...  (propose → evaluate → diagnose → gate)
        → finalize        needs: candidate       provides: report (sealed test)
          → report        reads run dir          provides: report (human summary)
```

The edges are not cosmetic: `baseline` cannot run until `implement-and-check`
emits `checked`, so the hard gate cannot be skipped; `finalize` consumes the best
candidate chosen on val, so selection happens before the test seal.

## Where each honesty guardrail lives

The pipeline's honesty is the sum of per-phase invariants, applied in order:

1. **ask-if-missing (intake)** — a missing NEEDED input is a question for the user,
   never a fabrication. Wrong here and everything downstream measures nothing.
2. **hard gate (implement-and-check)** — `cap-evolve check` must be green; the adapter
   must be implemented and the scorer deterministic before any budget is spent.
3. **freeze-once split + headroom (baseline)** — the split is written once and
   seeded; if the seed already saturates val, stop.
4. **val-only significance gate (the loop)** — acceptance is decided on val, by
   Δ > k·SE, never on train (overfit) and never on test (leak).
5. **sealed test, scored once (finalize)** — `test_used` makes re-scoring an error,
   so the headline number is unbiased.
6. **honest reading (report)** — test vs baseline, the val-test gap as overfitting,
   pass^k as reliability, uncertainty always shown.

orchestrate's contribution is that these are applied *automatically and in order*,
rather than depending on an operator to remember each one under time pressure.

## Stop rules — and why each exists

- **budget exhausted** (`max_iterations` / `max_metric_calls` / `max_usd`): the
  hard ceiling.
- **stall** (N consecutive rejects): the search has plateaued. Because the gate
  rejects non-significant gains, a run of rejects means remaining proposals are
  not clearing the noise floor — more tries mostly burn budget. Stalling early
  also limits the multiple-comparisons exposure (every extra candidate is another
  chance for a noise spike to look like a win).
- **no headroom**: the baseline already saturates val; there is nothing to gain.

Whatever the reason, the run **always ends with finalize + report**. A run that
stops without scoring the sealed test has produced edits but no result — and a
result is the only deliverable.

## Sources
- GEPA: Reflective Prompt Evolution (Agrawal et al., 2025) — the
  propose → reflect → evaluate → select loop orchestrate sequences:
  https://arxiv.org/abs/2507.19457
- τ-bench (Yao et al., 2024) — reliability (pass^k) as the end-of-run signal:
  https://arxiv.org/abs/2406.12045
- Koehn, "Statistical Significance Tests for MT Evaluation" (EMNLP 2004) — the
  significance discipline the acceptance loop enforces:
  https://aclanthology.org/W04-3250/
- Hastie, Tibshirani, Friedman, *Elements of Statistical Learning* — train/val/test
  roles the DAG encodes: https://hastie.su.domains/ElemStatLearn/

"""orchestrate composes other skills; it declares no abstract methods of its own.
check.py verifies it can resolve the manifest and read a spec."""

"""Wiring check for orchestrate."""
from __future__ import annotations
import json, sys
import _bootstrap  # noqa: F401
def main() -> int:
    rep = {"ok": False, "problems": [], "notes": []}
    try:
        import run
        if not hasattr(run, "main"):
            rep["problems"].append("run.py missing main()")
        from cap_evolve.specfile import read_yaml  # noqa: F401
        rep["notes"].append("orchestrate wiring ok")
    except Exception as e:  # noqa: BLE001
        rep["problems"].append(f"import failed: {e}")
    rep["ok"] = not rep["problems"]
    print(json.dumps(rep, indent=2))
    return 0 if rep["ok"] else 1
if __name__ == "__main__":
    sys.exit(main())

"""orchestrate — resolve capevolve.yaml + the manifest into a validated run plan.

Builds the ordered sequence from the manifest + spec (NOT a hardcoded list) and
validates the needs/provides DAG: walking the sequence, each step's ``needs``
tokens must already be satisfied by an upstream step's ``provides`` (or be
externally supplied — ``project``/``tasks`` come from intake/the adapter). A typo
in a meta ``needs``/``provides`` token, or a step ordered before its producer,
fails the plan loudly instead of silently running a broken pipeline.

The sequence now includes intake + the cap-evolve check gate before baseline. With
``--execute`` it hands off to ``cap-evolve run``.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import _bootstrap  # noqa: F401

from cap_evolve.specfile import read_yaml

# Tokens not produced by any skill — supplied by intake / the project adapter.
EXTERNAL_TOKENS = {"project", "tasks"}

# The fixed pipeline shape (phases + the spec-selected algorithm). Capability +
# optimizer skills are not sequenced steps — they are *bound into* the algorithm
# step (the algorithm calls the optimizer; the capability defines the edit surface).
PHASE_ORDER = ["intake", "implement-and-check", "baseline", "<algorithm>", "finalize", "report"]


def _skills_dir() -> Path | None:
    here = Path(__file__).resolve()
    for parent in here.parents:
        if (parent / "_registry" / "manifest.json").exists():
            return parent
        if (parent / "skills" / "_registry" / "manifest.json").exists():
            return parent / "skills"
    return None


def _load_manifest(skills_dir: Path) -> dict:
    return json.loads((skills_dir / "_registry" / "manifest.json").read_text())["skills"]


def _resolve_algorithm(name: str) -> str:
    """Old hill-climb skill names collapse to the one ``hill-climb`` skill."""
    if name in ("all-at-once", "cyclic", "hardest-first"):
        return "hill-climb"
    return name or "hill-climb"


def build_sequence(spec: dict) -> list[str]:
    algo = _resolve_algorithm(spec.get("algorithm_skill", "hill-climb"))
    return [algo if s == "<algorithm>" else s for s in PHASE_ORDER]


def validate_dag(sequence: list[str], manifest: dict) -> dict:
    """Check that every step's needs are satisfied by an upstream provides.

    Returns ``{ok, satisfied: [...], problems: [...]}``.
    """
    available = set(EXTERNAL_TOKENS)
    problems, trace = [], []
    for step in sequence:
        s = manifest.get(step)
        if s is None:
            problems.append(f"step {step!r} is not in the manifest")
            continue
        needs = list(s.get("needs", []) or [])
        missing = [n for n in needs if n not in available]
        if missing:
            problems.append(
                f"step {step!r} needs {missing} which no upstream step provides "
                f"(available so far: {sorted(available)})")
        trace.append({"step": step, "needs": needs,
                      "provides": list(s.get("provides", []) or []),
                      "missing": missing})
        available.update(s.get("provides", []) or [])
    return {"ok": not problems, "trace": trace, "problems": problems}


def main(argv=None) -> int:
    p = argparse.ArgumentParser(prog="orchestrate")
    p.add_argument("--spec", default=".capevolve/project/capevolve.yaml")
    p.add_argument("--project", default=".capevolve/project")
    p.add_argument("--execute", action="store_true", help="run the plan via `cap-evolve run`")
    args = p.parse_args(argv)

    spec = read_yaml(Path(args.spec).read_text()) if Path(args.spec).exists() else {}
    skills_dir = _skills_dir()
    manifest = _load_manifest(skills_dir) if skills_dir else {}

    sequence = build_sequence(spec)
    dag = validate_dag(sequence, manifest) if manifest else {
        "ok": False, "problems": ["no manifest found — run build_manifest.py"], "trace": []}

    plan = {
        "sequence": sequence,
        "dag_valid": dag["ok"],
        "dag": dag,
        "capabilities": spec.get("capabilities"),
        "optimizer": spec.get("optimizer_skill"),
        "algorithm": _resolve_algorithm(spec.get("algorithm_skill", "hill-climb")),
        "focus": spec.get("algorithm_focus"),
        "gate_mode": spec.get("gate_mode", "auto"),  # auto → harness uses the paired per-task gate
        "budget": {"max_iterations": spec.get("max_iterations"), "stall": spec.get("stall")},
        "rule": "cap-evolve check must be green before baseline; test scored once at finalize.",
    }

    if not dag["ok"]:
        print(json.dumps(plan, indent=2))
        return 1

    if args.execute:
        from cap_evolve.cli import main as cli_main
        return cli_main(["run", "--spec", args.spec, "--project", args.project])

    print(json.dumps(plan, 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())