`reach.catalog`¶

Filesystem scanning, discovery, and YAML frontmatter parsing routines.

Load skills from a repository root and compose them into catalogs for evaluation.

DEFAULT_SWEEP_SCALES `module-attribute` ¶

DEFAULT_SWEEP_SCALES: tuple[int, ...] = (
    _CANONICAL_LOG_STEPS[3:10]
)

CorpusScalingPlan ¶

Bases: BaseModel

Encapsulate precomputed distance geometry and nested catalogs for a scaling sweep.

Source code in src/reach/catalog.py

class CorpusScalingPlan(BaseModel):
    """Encapsulate precomputed distance geometry and nested catalogs for a scaling sweep."""

    model_config = ConfigDict(frozen=True)

    skills: tuple[Skill, ...]
    skill_names: tuple[str, ...]
    distance_matrix: tuple[tuple[float, ...], ...]
    similarity_matrix: tuple[tuple[float, ...], ...]
    sequence: tuple[str, ...]
    catalogs: tuple[Catalog, ...]
    anchor_skills: tuple[str, ...] | None = None

    @classmethod
    def create(
        cls,
        skills: Sequence[Skill],
        scales: Sequence[int],
        anchor_skills: Sequence[str] | None = None,
        scorer: Scorer | None = None,
    ) -> CorpusScalingPlan:
        """Construct a scaling plan by computing distance geometry and k-Center ordering once."""
        unique_skills = _deduplicate_skills(skills)
        names, dist, sim = _compute_cosine_bm25_distance_matrix(unique_skills, scorer=scorer)
        name_to_idx = {name: i for i, name in enumerate(names)}

        resolved_anchors: tuple[str, ...] | None = None
        if anchor_skills is not None:
            valid_anchors = tuple(dict.fromkeys(a for a in anchor_skills if a in name_to_idx))
            if valid_anchors:
                resolved_anchors = valid_anchors

        seq = _build_scaling_sequence(names, dist, sim, name_to_idx, resolved_anchors)
        catalogs = _build_nested_catalogs(seq, scales)

        return cls(
            skills=tuple(unique_skills),
            skill_names=names,
            distance_matrix=tuple(tuple(row) for row in dist),
            similarity_matrix=tuple(tuple(row) for row in sim),
            sequence=seq,
            catalogs=tuple(catalogs),
            anchor_skills=resolved_anchors,
        )

    def queries_for_scale(
        self,
        catalog: Catalog,
        raw_query_set: QuerySet,
        anchor_skills: Sequence[str] | None = None,
    ) -> QuerySet:
        """Slice query set into in-scope reachability probes for catalog."""
        effective_anchors = anchor_skills if anchor_skills is not None else self.anchor_skills
        return build_corpus_scaling_queries(
            scale_skills=catalog.skills,
            raw_query_set=raw_query_set,
            anchor_skills=effective_anchors,
        )

create `classmethod` ¶

create(
    skills: Sequence[Skill],
    scales: Sequence[int],
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> CorpusScalingPlan

Construct a scaling plan by computing distance geometry and k-Center ordering once.

Source code in src/reach/catalog.py

@classmethod
def create(
    cls,
    skills: Sequence[Skill],
    scales: Sequence[int],
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> CorpusScalingPlan:
    """Construct a scaling plan by computing distance geometry and k-Center ordering once."""
    unique_skills = _deduplicate_skills(skills)
    names, dist, sim = _compute_cosine_bm25_distance_matrix(unique_skills, scorer=scorer)
    name_to_idx = {name: i for i, name in enumerate(names)}

    resolved_anchors: tuple[str, ...] | None = None
    if anchor_skills is not None:
        valid_anchors = tuple(dict.fromkeys(a for a in anchor_skills if a in name_to_idx))
        if valid_anchors:
            resolved_anchors = valid_anchors

    seq = _build_scaling_sequence(names, dist, sim, name_to_idx, resolved_anchors)
    catalogs = _build_nested_catalogs(seq, scales)

    return cls(
        skills=tuple(unique_skills),
        skill_names=names,
        distance_matrix=tuple(tuple(row) for row in dist),
        similarity_matrix=tuple(tuple(row) for row in sim),
        sequence=seq,
        catalogs=tuple(catalogs),
        anchor_skills=resolved_anchors,
    )

queries_for_scale ¶

queries_for_scale(
    catalog: Catalog,
    raw_query_set: QuerySet,
    anchor_skills: Sequence[str] | None = None,
) -> QuerySet

Slice query set into in-scope reachability probes for catalog.

Source code in src/reach/catalog.py

def queries_for_scale(
    self,
    catalog: Catalog,
    raw_query_set: QuerySet,
    anchor_skills: Sequence[str] | None = None,
) -> QuerySet:
    """Slice query set into in-scope reachability probes for catalog."""
    effective_anchors = anchor_skills if anchor_skills is not None else self.anchor_skills
    return build_corpus_scaling_queries(
        scale_skills=catalog.skills,
        raw_query_set=raw_query_set,
        anchor_skills=effective_anchors,
    )

ResolvedTarget ¶

Bases: BaseModel

Structured resolution of a user-specified skill target and optional catalog.

Source code in src/reach/catalog.py

class ResolvedTarget(BaseModel):
    """Structured resolution of a user-specified skill target and optional catalog."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    skill_name: str
    catalog_path: Path | None = None
    manifest_path: Path | None = None

build_catalogs ¶

build_catalogs(
    skills: Sequence[Skill],
    mode: CatalogMode,
    size: int = 20,
    rivals: int = 10,
    seed: int = 0,
    scorer: Scorer | None = None,
    target_skill: str | None = None,
) -> list[Catalog]

Assemble skill catalogs from a corpus using the specified cataloging mode.

Source code in src/reach/catalog.py

def build_catalogs(
    skills: Sequence[Skill],
    mode: CatalogMode,
    size: int = 20,
    rivals: int = 10,
    seed: int = 0,
    scorer: Scorer | None = None,
    target_skill: str | None = None,
) -> list[Catalog]:
    """Assemble skill catalogs from a corpus using the specified cataloging mode."""
    if not skills:
        return []
    match mode:
        case CatalogMode.ALL:
            return [
                Catalog(
                    id="all",
                    mode=mode,
                    skills=tuple(s.name for s in skills),
                ),
            ]
        case CatalogMode.SINGLETON:
            return [Catalog(id=f"singleton:{s.name}", mode=mode, skills=(s.name,)) for s in skills]
        case CatalogMode.NEIGHBORHOOD:
            return build_neighborhood_catalogs(
                skills,
                size=size,
                rivals=rivals,
                seed=seed,
                scorer=scorer,
            )
        case CatalogMode.SWEEP:
            scales = resolve_sweep_scales(len(skills))
            if target_skill is not None:
                return build_scaling_catalogs(
                    skills,
                    target_skill=target_skill,
                    scales=scales,
                    seed=seed,
                    scorer=scorer,
                )
            return build_corpus_scaling_catalogs(
                skills,
                scales=scales,
                scorer=scorer,
            )

        case _:
            msg = f"unsupported catalog mode: {mode}"
            raise ValueError(msg)

build_corpus_scaling_catalogs ¶

build_corpus_scaling_catalogs(
    skills: Sequence[Skill],
    scales: Sequence[int],
    ordered_names: Sequence[str] | None = None,
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> list[Catalog]

Generate deterministic nested catalogs for whole-corpus capacity evaluation.

Source code in src/reach/catalog.py

def build_corpus_scaling_catalogs(
    skills: Sequence[Skill],
    scales: Sequence[int],
    ordered_names: Sequence[str] | None = None,
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> list[Catalog]:
    """Generate deterministic nested catalogs for whole-corpus capacity evaluation."""
    if not skills:
        return []

    if ordered_names is not None:
        catalogs: list[Catalog] = []
        for k in scales:
            count = max(1, min(k, len(ordered_names)))
            chosen = ordered_names[:count]
            catalogs.append(
                Catalog(
                    id=f"sweep:corpus:{count}",
                    mode=CatalogMode.SWEEP,
                    skills=tuple(sorted(chosen)),
                    target=None,
                )
            )
        return catalogs

    plan = CorpusScalingPlan.create(
        skills=skills,
        scales=scales,
        anchor_skills=anchor_skills,
        scorer=scorer,
    )
    return list(plan.catalogs)

build_corpus_scaling_queries ¶

build_corpus_scaling_queries(
    scale_skills: Sequence[str],
    raw_query_set: QuerySet,
    anchor_skills: Sequence[str] | None = None,
) -> QuerySet

Slice query set into in-scope reachability probes for installed skills.

Source code in src/reach/catalog.py

def build_corpus_scaling_queries(
    scale_skills: Sequence[str],
    raw_query_set: QuerySet,
    anchor_skills: Sequence[str] | None = None,
) -> QuerySet:
    """Slice query set into in-scope reachability probes for installed skills."""
    scale_set = set(scale_skills)
    target_skills = scale_set & set(anchor_skills) if anchor_skills is not None else scale_set
    in_scope_queries = [q for q in raw_query_set.queries if q.expected_skill in target_skills]
    return QuerySet(
        catalog_id=f"sweep:corpus:{len(scale_skills)}",
        queries=tuple(in_scope_queries),
        provenance=raw_query_set.provenance,
    )

build_corpus_scaling_sequence ¶

build_corpus_scaling_sequence(
    skills: Sequence[Skill],
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> tuple[str, ...]

Order skills using Farthest-First Traversal (k-Center) on Cosine-BM25 distance.

Source code in src/reach/catalog.py

def build_corpus_scaling_sequence(
    skills: Sequence[Skill],
    anchor_skills: Sequence[str] | None = None,
    scorer: Scorer | None = None,
) -> tuple[str, ...]:
    """Order skills using Farthest-First Traversal (k-Center) on Cosine-BM25 distance."""
    plan = CorpusScalingPlan.create(
        skills=skills,
        scales=(),
        anchor_skills=anchor_skills,
        scorer=scorer,
    )
    return plan.sequence

build_neighborhood_catalogs ¶

build_neighborhood_catalogs(
    skills: Sequence[Skill],
    size: int = 20,
    rivals: int = 10,
    seed: int = 0,
    scorer: Scorer | None = None,
) -> list[Catalog]

Generate fixed-size catalogs per skill containing target, rivals, and filler.

Source code in src/reach/catalog.py

def build_neighborhood_catalogs(
    skills: Sequence[Skill],
    size: int = 20,
    rivals: int = 10,
    seed: int = 0,
    scorer: Scorer | None = None,
) -> list[Catalog]:
    """Generate fixed-size catalogs per skill containing target, rivals, and filler."""
    if not skills:
        return []
    if size < MIN_NEIGHBORHOOD_SIZE:
        msg = f"a neighborhood needs at least 2 skills, got {size}"
        raise ValueError(msg)
    if rivals < 1:
        msg = f"a neighborhood needs at least 1 rival, got {rivals}"
        raise ValueError(msg)

    by_name = {s.name: s for s in skills}
    unique_skills = sorted(by_name.values(), key=lambda s: s.name)
    ranker = scorer or Bm25Scorer.from_skills(unique_skills)
    catalogs = []
    for skill in unique_skills:
        ranked: list[str] = []
        seen: set[str] = {skill.name}
        for name, _ in ranker.rank(skill, unique_skills):
            if name not in seen:
                seen.add(name)
                ranked.append(name)

        chosen = [skill.name, *ranked[:rivals]]

        rng = Random(f"{seed}:{skill.name}")  # noqa: S311 (deterministic benchmark sampling)
        chosen_set = set(chosen)
        pool = [s for s in ranked[rivals:] if s not in chosen_set]
        rng.shuffle(pool)
        chosen.extend(pool[: max(0, size - len(chosen))])

        catalogs.append(
            Catalog(
                id=f"neighborhood:{skill.name}",
                mode=CatalogMode.NEIGHBORHOOD,
                skills=tuple(sorted(chosen)),
                target=skill.name,
            ),
        )
    return catalogs

build_scaling_catalogs ¶

build_scaling_catalogs(
    skills: Sequence[Skill],
    target_skill: str,
    scales: Sequence[int],
    rivals_share: float = 0.5,
    seed: int = 0,
    scorer: Scorer | None = None,
) -> list[Catalog]

Generate multi-scale catalogs for a target skill across requested scales.

Source code in src/reach/catalog.py

def build_scaling_catalogs(
    skills: Sequence[Skill],
    target_skill: str,
    scales: Sequence[int],
    rivals_share: float = 0.5,
    seed: int = 0,
    scorer: Scorer | None = None,
) -> list[Catalog]:
    """Generate multi-scale catalogs for a target skill across requested scales."""
    if not skills:
        return []
    by_name = {s.name: s for s in skills}
    if target_skill not in by_name:
        msg = f"target skill {target_skill!r} not in skills"
        raise KeyError(msg)

    unique_skills = list(by_name.values())
    target_obj = by_name[target_skill]
    ranker = scorer or Bm25Scorer.from_skills(unique_skills)
    ranked = [name for name, _ in ranker.rank(target_obj, unique_skills) if name != target_skill]

    rng = Random(f"{seed}:{target_skill}")  # noqa: S311 (deterministic benchmark sampling)
    filler_order = list(ranked)
    rng.shuffle(filler_order)

    unique_sorted_scales = sorted({max(1, k) for k in scales})
    chosen_by_scale: dict[int, tuple[str, ...]] = {}
    current_chosen: list[str] = [target_skill]
    current_set: set[str] = {target_skill}

    for k in unique_sorted_scales:
        if k <= 1:
            chosen_by_scale[k] = (target_skill,)
            continue

        r = max(1, round((k - 1) * rivals_share))
        _extend_unique_up_to(current_chosen, current_set, ranked[:r], k)
        _extend_unique_up_to(current_chosen, current_set, filler_order, k)
        chosen_by_scale[k] = tuple(sorted(current_chosen[:k]))

    return [
        Catalog(
            id=f"sweep:{target_skill}:{max(1, raw_k)}",
            mode=CatalogMode.SWEEP,
            skills=chosen_by_scale[max(1, raw_k)],
            target=target_skill,
        )
        for raw_k in scales
    ]

corpus_digest ¶

corpus_digest(skills: Sequence[Skill]) -> str

Compute deterministic 12-char SHA-256 digest of corpus names/descriptions.

Parameters:

Name	Type	Description	Default
`skills`	`Sequence[Skill]`	Sequence of resident Skill objects.	required

Returns:

Type	Description
`str`	A 12-character hexadecimal SHA-256 digest identifying the corpus selection surface.

Source code in src/reach/catalog.py

def corpus_digest(skills: Sequence[Skill]) -> str:
    """Compute deterministic 12-char SHA-256 digest of corpus names/descriptions.

    Args:
        skills: Sequence of resident Skill objects.

    Returns:
        A 12-character hexadecimal SHA-256 digest identifying the corpus selection surface.
    """
    material = "\n".join(
        f"{skill.name}\n{skill.description}" + ("" if skill.model_invocable else HIDDEN_MARKER)
        for skill in sorted(skills, key=lambda s: s.name)
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:12]

deduplicate_skills ¶

deduplicate_skills(skills: Sequence[Skill]) -> list[Skill]

Filter duplicate skills by name, preserving first insertion order.

Source code in src/reach/catalog.py

def deduplicate_skills(skills: Sequence[Skill]) -> list[Skill]:
    """Filter duplicate skills by name, preserving first insertion order."""
    unique_skills: list[Skill] = []
    seen: set[str] = set()
    for s in skills:
        if s.name not in seen:
            seen.add(s.name)
            unique_skills.append(s)
    return unique_skills

determine_min_scale ¶

determine_min_scale(total_skills: int) -> int

Determine optimal baseline starting scale based on corpus size.

Source code in src/reach/catalog.py

def determine_min_scale(total_skills: int) -> int:
    """Determine optimal baseline starting scale based on corpus size."""
    if total_skills > LARGE_CORPUS_THRESHOLD:
        return 10
    if total_skills > MEDIUM_CORPUS_THRESHOLD:
        return 5
    if total_skills > SMALL_CORPUS_THRESHOLD:
        return 2
    return 1

find_cluster_medoids ¶

find_cluster_medoids(
    skills: Sequence[Skill],
    k: int,
    scorer: Scorer | None = None,
) -> tuple[str, ...]

Find k representative skill medoids across modularity clusters.

Partition skills into communities via modularity optimization, then select the central medoid skill from each cluster (maximizing intra-cluster BM25 similarity). If fewer than k clusters exist, iteratively select the farthest remaining skills from the chosen cohort to ensure maximal vocabulary diversity.

Parameters:

Name	Type	Description	Default
`skills`	`Sequence[Skill]`	The corpus of skills to partition and select from.	required
`k`	`int`	The desired number of anchor medoid skills.	required
`scorer`	`Scorer \| None`	Optional BM25 scorer for computing skill distances.	`None`

Returns:

Type	Description
`tuple[str, ...]`	Tuple of up to k representative skill names.

Source code in src/reach/catalog.py

def find_cluster_medoids(
    skills: Sequence[Skill],
    k: int,
    scorer: Scorer | None = None,
) -> tuple[str, ...]:
    """Find k representative skill medoids across modularity clusters.

    Partition skills into communities via modularity optimization, then select
    the central medoid skill from each cluster (maximizing intra-cluster BM25 similarity).
    If fewer than k clusters exist, iteratively select the farthest remaining
    skills from the chosen cohort to ensure maximal vocabulary diversity.

    Args:
        skills: The corpus of skills to partition and select from.
        k: The desired number of anchor medoid skills.
        scorer: Optional BM25 scorer for computing skill distances.

    Returns:
        Tuple of up to k representative skill names.
    """
    if k <= 0 or not skills:
        return ()

    unique_skills = _deduplicate_skills(skills)
    names, dist, sim = _compute_cosine_bm25_distance_matrix(unique_skills, scorer=scorer)
    n = len(names)
    if n <= k:
        return names

    from reach.cluster import cluster_skills

    partition = cluster_skills(unique_skills, resolution=1.5, max_clusters=k)
    name_to_idx = {name: i for i, name in enumerate(names)}
    chosen_indices = _extract_cluster_medoid_indices(partition.clusters, name_to_idx, sim)

    order = _farthest_first_traversal(dist, chosen_indices, min(k, n))
    return tuple(names[i] for i in order)

find_skill_manifest ¶

find_skill_manifest(skill_path: Path) -> Path | None

Traverse upward from SKILL.md or directory to locate nearest skills.json or lockfile.

Source code in src/reach/catalog.py

def find_skill_manifest(skill_path: Path) -> Path | None:
    """Traverse upward from SKILL.md or directory to locate nearest skills.json or lockfile."""
    resolved = resolve_path(skill_path)
    current = resolved if resolved.is_dir() else resolved.parent
    try:
        mtime_ns = current.stat().st_mtime_ns
    except OSError:
        mtime_ns = 0
    return _find_manifest_for_dir(current, mtime_ns)

generate_log_scales ¶

generate_log_scales(
    total_skills: int, min_scale: int | None = None
) -> tuple[int, ...]

Generate human-friendly logarithmic sweep scales up to total_skills.

Source code in src/reach/catalog.py

def generate_log_scales(
    total_skills: int,
    min_scale: int | None = None,
) -> tuple[int, ...]:
    """Generate human-friendly logarithmic sweep scales up to total_skills."""
    if total_skills <= 0:
        msg = f"total_skills must be positive, got {total_skills}"
        raise ValueError(msg)
    if total_skills == 1:
        return (1,)

    start = min_scale if min_scale is not None else determine_min_scale(total_skills)
    scales = [s for s in _CANONICAL_LOG_STEPS if start <= s < total_skills]
    if (not scales or scales[0] != start) and (start < total_skills and start not in scales):
        scales.insert(0, start)
    if total_skills not in scales:
        scales.append(total_skills)
    return tuple(sorted(set(scales)))

load_registry_skills ¶

load_registry_skills(
    project: str,
    location: str = "global",
    publisher: str | None = None,
    fresh: bool = False,
    no_cache: bool = False,
    cache_ttl_seconds: int = 300,
    cache_root: Path | str | None = None,
) -> list[Skill]

Fetch and load skills from Google Cloud Agent Registry via local cache mirror.

Parameters:

Name	Type	Description	Default
`project`	`str`	Google Cloud project ID.	required
`location`	`str`	Registry location (default: 'global').	`'global'`
`publisher`	`str \| None`	Optional publisher filter.	`None`
`fresh`	`bool`	If True, bypass metadata TTL and query live.	`False`
`no_cache`	`bool`	If True, run in ephemeral memory/tempdir.	`False`
`cache_ttl_seconds`	`int`	TTL in seconds for metadata cache validity.	`300`
`cache_root`	`Path \| str \| None`	Optional custom cache directory.	`None`

Returns:

Type	Description
`list[Skill]`	Sorted list of resident Skill objects.

Source code in src/reach/catalog.py

def load_registry_skills(
    project: str,
    location: str = "global",
    publisher: str | None = None,
    fresh: bool = False,
    no_cache: bool = False,
    cache_ttl_seconds: int = 300,
    cache_root: Path | str | None = None,
) -> list[Skill]:
    """Fetch and load skills from Google Cloud Agent Registry via local cache mirror.

    Args:
        project: Google Cloud project ID.
        location: Registry location (default: 'global').
        publisher: Optional publisher filter.
        fresh: If True, bypass metadata TTL and query live.
        no_cache: If True, run in ephemeral memory/tempdir.
        cache_ttl_seconds: TTL in seconds for metadata cache validity.
        cache_root: Optional custom cache directory.

    Returns:
        Sorted list of resident Skill objects.
    """
    from reach.registry import RegistryCacheManager

    manager = RegistryCacheManager(cache_root=cache_root)
    return manager.resolve_skills(
        project=project,
        location=location,
        publisher=publisher,
        fresh=fresh,
        no_cache=no_cache,
        cache_ttl_seconds=cache_ttl_seconds,
    )

load_skills ¶

load_skills(root: Path | str) -> list[Skill]

Load and parse all skills under a directory root, sorted by skill name.

Deduplicates skills sharing the same name by selecting the shortest path.

Parameters:

Name	Type	Description	Default
`root`	`Path \| str`	Directory path containing skill subdirectories or SKILL.md files.	required

Returns:

Type	Description
`list[Skill]`	Sorted list of resident Skill objects found under the directory.

Raises:

Type	Description
`NotADirectoryError`	If the resolved path does not exist or is not a directory.

Source code in src/reach/catalog.py

def load_skills(root: Path | str) -> list[Skill]:
    """Load and parse all skills under a directory root, sorted by skill name.

    Deduplicates skills sharing the same name by selecting the shortest path.

    Args:
        root: Directory path containing skill subdirectories or SKILL.md files.

    Returns:
        Sorted list of resident Skill objects found under the directory.

    Raises:
        NotADirectoryError: If the resolved path does not exist or is not a directory.
    """
    resolved = resolve_path(root)
    if not resolved.is_dir():
        msg = f"skill root does not exist: {resolved}"
        raise NotADirectoryError(msg)
    by_name: dict[str, Skill] = {}
    candidate_files = sorted(
        _skill_files(resolved),
        key=lambda p: (len(p.parts), str(p)),
    )
    for skill_file in candidate_files:
        skill = parse_frontmatter(skill_file.read_text(encoding="utf-8"), skill_file)
        if skill is not None and skill.name not in by_name:
            by_name[skill.name] = skill
    return sorted(by_name.values(), key=lambda s: s.name)

parse_frontmatter ¶

parse_frontmatter(text: str, path: Path) -> Skill | None

Parse a SKILL.md file's YAML frontmatter into a validated Skill model.

Parameters:

Name	Type	Description	Default
`text`	`str`	Raw markdown file contents including frontmatter block.	required
`path`	`Path`	Filesystem path to the SKILL.md file (used for fallback naming).	required

Returns:

Type	Description
`Skill \| None`	A validated Skill model instance, or None if frontmatter cannot be parsed.

Source code in src/reach/catalog.py

def parse_frontmatter(text: str, path: Path) -> Skill | None:
    """Parse a SKILL.md file's YAML frontmatter into a validated Skill model.

    Args:
        text: Raw markdown file contents including frontmatter block.
        path: Filesystem path to the SKILL.md file (used for fallback naming).

    Returns:
        A validated Skill model instance, or None if frontmatter cannot be parsed.
    """
    split = split_frontmatter(text)
    if split is None:
        return None
    frontmatter, _body = split
    try:
        loaded = yaml.safe_load(frontmatter)
    except yaml.YAMLError:
        return None
    if not isinstance(loaded, dict):
        return None
    try:
        parsed = _SkillFrontmatter.model_validate(loaded)
        name = parsed.name or path.parent.name
        raw_allowed = parsed.allowed_tools
        allowed = _extract_allowed_skills(raw_allowed)
        deps = _extract_declared_dependencies(loaded, allowed)
        manifest = find_skill_manifest(path)
        manifest_src = _resolve_manifest_source(name, manifest)

        return Skill(
            name=name,
            description=parsed.description,
            metadata=parsed.stringified_metadata(),
            path=path.parent,
            allowed_tools=allowed,
            declared_dependencies=deps,
            manifest_source=manifest_src,
            model_invocable=parsed.is_model_invocable(),
        )
    except (ValueError, ValidationError) as exc:
        logging.getLogger(__name__).warning("Skipping invalid SKILL.md at %s: %s", path, exc)
        return None

resident_skills ¶

resident_skills(
    catalog: Catalog, skills: Sequence[Skill]
) -> list[Skill]

Retrieve ordered Skill objects resident in the specified catalog.

Source code in src/reach/catalog.py

def resident_skills(catalog: Catalog, skills: Sequence[Skill]) -> list[Skill]:
    """Retrieve ordered Skill objects resident in the specified catalog."""
    by_name = {s.name: s for s in skills}
    missing = [name for name in catalog.skills if name not in by_name]
    if missing:
        msg = f"catalog {catalog.id!r} names skills not loaded: {missing}"
        raise KeyError(msg)
    return [by_name[name] for name in catalog.skills]

resolve_catalog ¶

resolve_catalog(
    catalogs: Sequence[Catalog], catalog_id: str
) -> Catalog

Retrieve a catalog by identifier from a sequence of catalogs.

Source code in src/reach/catalog.py

def resolve_catalog(catalogs: Sequence[Catalog], catalog_id: str) -> Catalog:
    """Retrieve a catalog by identifier from a sequence of catalogs."""
    for catalog in catalogs:
        if catalog.id == catalog_id:
            return catalog
    available = ", ".join(sorted(c.id for c in catalogs)[:8]) or "(none)"
    hint = ""
    if any(c.id.startswith("neighborhood:") for c in catalogs):
        hint = "; pass --catalog <name> --rescope to evaluate against an available catalog"
    msg = f"no catalog named {catalog_id!r}; available: {available}{hint}"
    raise KeyError(msg)

resolve_skill_target ¶

resolve_skill_target(
    target: str | Path | None,
    explicit_catalog: Path | str | None = None,
    *,
    command_name: str = "eval",
) -> ResolvedTarget | None

Resolve a skill name and catalog path from a name, directory, or SKILL.md file.

Parameters:

Name	Type	Description	Default
`target`	`str \| Path \| None`	Skill name, directory path, or SKILL.md file path.	required
`explicit_catalog`	`Path \| str \| None`	Explicit catalog path if specified by user flag (e.g. --skills).	`None`
`command_name`	`str`	CLI command name for formatting multi-skill error remedies.	`'eval'`

Returns:

Type	Description
`ResolvedTarget \| None`	ResolvedTarget with canonical skill_name and inferred catalog_path,
`ResolvedTarget \| None`	or None if target is None.

Raises:

Type	Description
`FileNotFoundError`	If target looks like a path but does not exist on disk.
`ValueError`	If target is a non-SKILL.md file, a directory containing skills, a directory with no SKILL.md, or a SKILL.md with invalid frontmatter.

Source code in src/reach/catalog.py

def resolve_skill_target(
    target: str | Path | None,
    explicit_catalog: Path | str | None = None,
    *,
    command_name: str = "eval",
) -> ResolvedTarget | None:
    """Resolve a skill name and catalog path from a name, directory, or SKILL.md file.

    Args:
        target: Skill name, directory path, or SKILL.md file path.
        explicit_catalog: Explicit catalog path if specified by user flag (e.g. --skills).
        command_name: CLI command name for formatting multi-skill error remedies.

    Returns:
        ResolvedTarget with canonical skill_name and inferred catalog_path,
        or None if target is None.

    Raises:
        FileNotFoundError: If target looks like a path but does not exist on disk.
        ValueError: If target is a non-SKILL.md file, a directory containing skills,
            a directory with no SKILL.md, or a SKILL.md with invalid frontmatter.
    """
    if not target or not (raw_str := str(target).strip()):
        return None

    catalog_path = resolve_path(explicit_catalog) if explicit_catalog else None

    looks_like_path = (
        isinstance(target, Path)
        or "/" in raw_str
        or "\\" in raw_str
        or raw_str.startswith(("~", "."))
    )

    # If an explicit catalog was provided and the target is a raw name, don't probe cwd
    if catalog_path is not None and not looks_like_path:
        return ResolvedTarget(skill_name=raw_str, catalog_path=catalog_path)

    named = resolve_path(target)

    # 1. Path does not exist
    if not named.exists():
        if looks_like_path:
            msg = f"skill path does not exist: '{target}'"
            raise FileNotFoundError(msg)
        return ResolvedTarget(skill_name=raw_str, catalog_path=catalog_path)

    # 2. Path is a file
    if named.is_file():
        return _resolve_manifest_file(
            named,
            catalog_path,
            looks_like_path=looks_like_path,
            target=target,
            raw_str=raw_str,
        )

    # 3. Path is a directory
    if named.is_dir():
        return _resolve_skill_directory(
            named,
            catalog_path,
            looks_like_path=looks_like_path,
            target=target,
            raw_str=raw_str,
            command_name=command_name,
        )

    return None

resolve_sweep_scales ¶

resolve_sweep_scales(
    total_skills: int,
    requested: Sequence[int] | None = None,
) -> tuple[int, ...]

Resolve and clamp catalog sweep scales against available corpus size.

Source code in src/reach/catalog.py

def resolve_sweep_scales(
    total_skills: int,
    requested: Sequence[int] | None = None,
) -> tuple[int, ...]:
    """Resolve and clamp catalog sweep scales against available corpus size."""
    if total_skills <= 0:
        msg = f"total_skills must be positive, got {total_skills}"
        raise ValueError(msg)

    if not requested:
        return generate_log_scales(total_skills)

    scales = sorted({s for s in requested if 1 <= s < total_skills})
    if total_skills not in scales:
        scales.append(total_skills)
    return tuple(scales)

split_frontmatter ¶

split_frontmatter(text: str) -> tuple[str, str] | None

Split raw markdown text into frontmatter YAML and markdown body content.

Parameters:

Name	Type	Description	Default
`text`	`str`	Raw content of a markdown skill file.	required

Returns:

Type	Description
`tuple[str, str] \| None`	A tuple of (frontmatter_yaml, markdown_body) if valid delimiter lines are found,
`tuple[str, str] \| None`	or None if the file lacks valid frontmatter delimiters.

Source code in src/reach/catalog.py

def split_frontmatter(text: str) -> tuple[str, str] | None:
    """Split raw markdown text into frontmatter YAML and markdown body content.

    Args:
        text: Raw content of a markdown skill file.

    Returns:
        A tuple of (frontmatter_yaml, markdown_body) if valid delimiter lines are found,
        or None if the file lacks valid frontmatter delimiters.
    """
    stripped = text.removeprefix(BOM)
    parts = _FRONTMATTER_PATTERN.split(stripped, maxsplit=2)
    if len(parts) < FRONTMATTER_SPLIT_PARTS or parts[0] != "":
        return None
    return parts[1], parts[2]

reach.catalog¶

DEFAULT_SWEEP_SCALES module-attribute ¶

CorpusScalingPlan ¶

create classmethod ¶

queries_for_scale ¶

ResolvedTarget ¶

build_catalogs ¶

build_corpus_scaling_catalogs ¶

build_corpus_scaling_queries ¶

build_corpus_scaling_sequence ¶

build_neighborhood_catalogs ¶

build_scaling_catalogs ¶

corpus_digest ¶

deduplicate_skills ¶

determine_min_scale ¶

find_cluster_medoids ¶

find_skill_manifest ¶

generate_log_scales ¶

load_registry_skills ¶

load_skills ¶

parse_frontmatter ¶

resident_skills ¶

resolve_catalog ¶

resolve_skill_target ¶

resolve_sweep_scales ¶

split_frontmatter ¶

`reach.catalog`¶

DEFAULT_SWEEP_SCALES `module-attribute` ¶

create `classmethod` ¶