reach.catalog

Filesystem scanning, discovery, and YAML frontmatter parsing routines.

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

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_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]

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

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)