reach.check

Two-stage CI/CD regression check and quality gate orchestration.

Orchestrate the two-stage CI/CD regression check and quality gate.

CheckAssertion

Bases: BaseModel

Single threshold assertion evaluated during empirical quality gate.

Source code in src/reach/check.py

class CheckAssertion(BaseModel):
    """Single threshold assertion evaluated during empirical quality gate."""

    model_config = ConfigDict(frozen=True)

    comparison: str
    message: str
    name: str
    observed: float
    passed: bool
    threshold: float

CheckOutcome

Bases: BaseModel

Comprehensive outcome of a two-stage quality gate check.

Source code in src/reach/check.py

class CheckOutcome(BaseModel):
    """Comprehensive outcome of a two-stage quality gate check."""

    model_config = ConfigDict(frozen=True)

    assertions: tuple[CheckAssertion, ...] = ()
    budget: int = 50
    budget_exhausted: bool = False
    exit_code: int = 0
    lint_report: LintReport
    probes_executed: int = 0
    queries_probed: int = 0
    skills_checked: int = 0
    stage_failed: CheckStage | None = None
    classification: ClassificationReport | None = None

    @property
    def passed(self) -> bool:
        """Return True if all stages and assertions passed."""
        return self.exit_code == 0

passed property

passed: bool

Return True if all stages and assertions passed.

CheckStage

Bases: StrEnum

Enumerate execution stages in the CI quality gate.

Source code in src/reach/check.py

class CheckStage(StrEnum):
    """Enumerate execution stages in the CI quality gate."""

    EMPIRICAL = "empirical"
    STATIC = "static"

EmpiricalMetrics

Bases: BaseModel

Aggregated empirical metrics observed during quality gate execution.

Source code in src/reach/check.py

class EmpiricalMetrics(BaseModel):
    """Aggregated empirical metrics observed during quality gate execution."""

    model_config = ConfigDict(frozen=True)

    recall: float
    accuracy: float
    misroute_rate: float
    entrypoint_accuracy: float = 0.0
    trajectory_reachability: float = 0.0
    step_efficiency: float = 0.0
    skill_f1: float = 0.0
    redundancy: float = 0.0

changed_skills

changed_skills(
    since: str = "HEAD~1",
    root: Path | str | None = None,
    *,
    timeout: float = 30.0,
) -> tuple[str, ...]

Identify skill names modified in git repository relative to a reference.

Source code in src/reach/check.py

def changed_skills(
    since: str = "HEAD~1",
    root: Path | str | None = None,
    *,
    timeout: float = 30.0,
) -> tuple[str, ...]:
    """Identify skill names modified in git repository relative to a reference."""
    work_dir = Path(root).resolve() if root is not None else Path.cwd().resolve()
    clean_since = since.strip()
    if clean_since.startswith("-"):
        msg = f"git reference must not begin with a dash: {since!r}"
        raise ValueError(msg)
    try:
        completed = subprocess.run(
            ["git", "diff", "--name-only", clean_since, "--"],
            capture_output=True,
            text=True,
            cwd=work_dir,
            check=False,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as exc:
        msg = f"git diff timed out after {timeout}s against ref '{since}'"
        raise ValueError(msg) from exc
    except OSError:
        return ()

    if completed.returncode != 0:
        err = completed.stderr.strip() or "git command failed"
        msg = f"git diff failed against ref '{since}': {err}"
        err_lower = err.lower()
        if any(
            pattern in err_lower
            for pattern in ("bad revision", "unknown revision", "shallow", "ambiguous argument")
        ):
            msg += (
                f"\nHint: Git reference '{since}' may not exist in this clone. "
                "If running in CI (e.g. GitHub Actions), ensure full git history is fetched "
                "(e.g. 'fetch-depth: 0')."
            )
        raise ValueError(msg)

    discovered: set[str] = set()
    for line in completed.stdout.splitlines():
        trimmed = line.strip()
        if not trimmed:
            continue
        if match := _SKILL_PATH_PATTERN.search(trimmed):
            discovered.add(match.group(1))
        elif trimmed.endswith("SKILL.md"):
            skill_name = Path(trimmed).parent.name or Path(work_dir).resolve().name
            if skill_name:
                discovered.add(skill_name)

    return tuple(sorted(discovered))

run_check

run_check(
    *,
    skills_paths: Sequence[Path | str] | None = None,
    queries_path: Path | str | None = None,
    changed: bool = False,
    since: str = "HEAD~1",
    strict: bool = True,
    min_recall: float = 0.8,
    min_accuracy: float = 0.8,
    max_misroute: float = 0.1,
    min_entrypoint: float | None = None,
    min_reachability: float | None = None,
    min_efficiency: float | None = None,
    min_f1: float | None = None,
    max_redundancy: float | None = None,
    budget: int = 50,
    config: RunConfig | None = None,
    settings: CheckSettings | None = None,
    agent: str | None = None,
    rule_overrides: Mapping[str, Severity] | None = None,
    runtime_options: dict[str, Any] | None = None,
    global_scope: bool = False,
    confirm_callback: Callable[
        [str, list[Skill], Sequence[Path]], int
    ]
    | None = None,
) -> CheckOutcome

Execute two-stage quality gate: static lint pre-flight then empirical assertions.

Source code in src/reach/check.py

def run_check(  # noqa: PLR0913
    *,
    skills_paths: Sequence[Path | str] | None = None,
    queries_path: Path | str | None = None,
    changed: bool = False,
    since: str = "HEAD~1",
    strict: bool = True,
    min_recall: float = 0.80,
    min_accuracy: float = 0.80,
    max_misroute: float = 0.10,
    min_entrypoint: float | None = None,
    min_reachability: float | None = None,
    min_efficiency: float | None = None,
    min_f1: float | None = None,
    max_redundancy: float | None = None,
    budget: int = 50,
    config: RunConfig | None = None,
    settings: CheckSettings | None = None,
    agent: str | None = None,
    rule_overrides: Mapping[str, Severity] | None = None,
    runtime_options: dict[str, Any] | None = None,
    global_scope: bool = False,
    confirm_callback: Callable[[str, list[Skill], Sequence[Path]], int] | None = None,
) -> CheckOutcome:
    """Execute two-stage quality gate: static lint pre-flight then empirical assertions."""
    if settings is not None:
        check_settings = settings
    elif config is not None:
        check_settings = config.check
    else:
        check_settings = CheckSettings(
            min_recall=min_recall,
            min_accuracy=min_accuracy,
            max_misroute=max_misroute,
            min_entrypoint=min_entrypoint,
            min_reachability=min_reachability,
            min_efficiency=min_efficiency,
            min_f1=min_f1,
            max_redundancy=max_redundancy,
            budget=budget,
            strict=strict,
            since=since,
        )

    if check_settings.budget < 1:
        msg = "Probe budget must be at least 1"
        raise ValueError(msg)

    lint_cfg = LintSettings.from_settings(overrides=rule_overrides)
    resolved_paths, lint_report = _resolve_candidate_skills(
        skills_paths,
        lint_config=lint_cfg,
        agent=agent,
        global_scope=global_scope,
    )
    if not resolved_paths:
        scope_msg = (
            "in user global configuration (~/.agents/skills, ~/.claude/skills, etc.)"
            if global_scope
            else "in .agents/skills or skills"
        )
        msg = f"No skill paths specified and no skills found {scope_msg}."
        raise ValueError(msg)

    catalog_skills = _load_catalog_skills(resolved_paths) if changed else []
    available = {s.name for s in catalog_skills} if changed else set()
    lint_report, modified, early_outcome = _apply_changed_scope(
        lint_report, changed, check_settings.since, check_settings.budget, available
    )
    if early_outcome is not None:
        return early_outcome

    static_outcome = _check_static_gate_or_early_exit(
        lint_report, check_settings.strict, check_settings.budget, queries_path
    )
    if static_outcome is not None:
        return static_outcome
    if queries_path is None:
        msg = "queries_path must be provided for empirical assertions"
        raise ValueError(msg)

    queries_to_run, budget_exhausted = _filter_check_queries(
        queries_path,
        modified,
        changed,
        check_settings.budget,
        skills=catalog_skills,
    )

    empty_outcome = _check_empty_queries_exit(lint_report, queries_to_run, check_settings.budget)
    if empty_outcome is not None:
        return empty_outcome

    from reach.config import default_agent

    resolved_agent = agent or (config.runtime.agent if config is not None else default_agent())
    if confirm_callback is not None and resolved_agent not in ("keyword", FAKE_AGENT):
        loaded = _load_catalog_skills(resolved_paths)
        if code := confirm_callback(resolved_agent, loaded, resolved_paths):
            return CheckOutcome(
                lint_report=lint_report,
                stage_failed=CheckStage.EMPIRICAL,
                assertions=(),
                skills_checked=lint_report.skills_checked,
                queries_probed=0,
                probes_executed=0,
                budget=check_settings.budget,
                exit_code=code,
            )

    report, metrics, probes_executed = _execute_empirical_probes(
        queries_to_run,
        resolved_paths,
        agent,
        runtime_options,
        config=config,
    )

    assertions = _build_check_assertions(metrics, check_settings)

    all_passed = all(a.passed for a in assertions)

    return CheckOutcome(
        lint_report=lint_report,
        stage_failed=None if all_passed else CheckStage.EMPIRICAL,
        assertions=assertions,
        skills_checked=lint_report.skills_checked,
        queries_probed=len(queries_to_run),
        probes_executed=probes_executed,
        budget=check_settings.budget,
        budget_exhausted=budget_exhausted,
        exit_code=0 if all_passed else 2,
        classification=report,
    )