rfgen.registry

Plugin discovery and the metadata schema every plugin package declares. The registry is what turns “an installed package on the Python path” into “an emitter, channel, labeler, store, or command the framework can resolve from config.” A single concrete registry ships: EntryPointRegistry, which reads importlib.metadata.entry_points. An optional second primitive, pluggy, is gated behind the rfgen[plugin-hooks] extra for hook-based discovery.

Warning

Pre-implementation. Class signatures, field names, and class-attribute defaults are proposals. Once code lands, this page will be regenerated from docstrings via sphinx.ext.autodoc. The shape below matches what autodoc emits so the swap is mechanical.

Module summary

from rfgen.registry import EntryPointRegistry, PluginMetadata

emitters = EntryPointRegistry[BaseEmitter]("rfgen.emitters")
emitters.discover()              # records names; does not import any plugin module
print(emitters.available())      # ('cw_tone', 'fmcw', 'my_radar', 'pulsed_barker')
emitter = emitters.get("my_radar")  # this call triggers the lazy import

Discovery is two-phase by design: discover() walks the entry-point index and records the names; get(name) is the first call that imports the plugin module. Importing rfgen.registry itself never imports any third-party plugin.

Class index

Class	Kind	Notes
BaseRegistry	abc	Generic ABC; implements register, get, available, metadata_for.
EntryPointRegistry	concrete	Single shipped subclass; backed by importlib.metadata.entry_points.
PluginMetadata	datatype (Pydantic)	Frozen plugin descriptor.
BaseCommand	protocol	Dispatch contract for Layer 9 services callable from the CLI.

The constant EXIT_CODES_VALID = frozenset({0, 1, 2, 3, 130}) is also exported. It is the closed set of process exit codes a BaseCommand is allowed to return.

---

class rfgen.registry.BaseRegistry

from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Generic, TypeVar

T = TypeVar("T")


class BaseRegistry(ABC, Generic[T]):
    """Generic ABC for plugin registries."""

    def register(
        self,
        name: str,
        factory: Callable[..., T],
        *,
        source: str = "<unknown>",
        metadata: PluginMetadata | None = None,
    ) -> None: ...

    def get(self, name: str) -> T: ...

    def available(self) -> tuple[str, ...]: ...

    def metadata_for(self, name: str) -> PluginMetadata | None: ...

    @abstractmethod
    def discover(self) -> None: ...

A generic ABC parameterized by the produced plugin type T. Subclasses implement discover(); the shared registration and lookup API lives on the base class so subclasses do not duplicate it.

register(name, factory, *, source, metadata)

Register a factory under a name.

• name: canonical name for the plugin entry.

• factory: zero-or-more-argument callable returning a T.

• source: human-readable origin of the registration (entry-point module path, file path, …). Used in conflict error messages.

• metadata: optional PluginMetadata describing the plugin. If provided and metadata.requires lists an extra whose marker package is not installed, registration raises BackendUnavailableError.

Raises:

• RegistryError: if name is already registered. The message names both source modules.

• BackendUnavailableError: if metadata.requires lists an extra whose marker package is not present.

get(name)

Resolve name and return the constructed plugin instance. Raises PluginNotFoundError if name is not registered; the message lists the available names for this registry.

available()

Return the sorted tuple of registered names. On EntryPointRegistry, this includes pending entry-point names that have been discovered but not yet loaded.

metadata_for(name)

Return the PluginMetadata registered for name, or None if no metadata was registered.

discover()

Abstract. Concrete subclasses populate the registry from the underlying source. Must be idempotent.

---

class rfgen.registry.EntryPointRegistry

class EntryPointRegistry(BaseRegistry[T]):
    """A BaseRegistry backed by importlib.metadata.entry_points."""

    def __init__(self, group: str) -> None: ...

    @property
    def group(self) -> str: ...

    def discover(self) -> None: ...
    def get(self, name: str) -> T: ...
    def available(self) -> tuple[str, ...]: ...

The single concrete BaseRegistry subclass shipped with rfgen. It is constructed with a single entry-point group name. The framework instantiates one registry per documented group:

• rfgen.emitters

• rfgen.channels

• rfgen.time_placement

• rfgen.freq_placement

• rfgen.labelers

• rfgen.stores

• rfgen.annotators

• rfgen.llm_clients

• rfgen.executors

• rfgen.annotation_orchestrators

• rfgen.credentials

• rfgen.log_sinks

• rfgen.commands

Lazy discovery

discover() walks the entry-point index and records every entry in the configured group, but does not call EntryPoint.load(). The first get(name) call for a pending name pops the entry, calls load(), and registers the resulting object as the factory under source="entry_point:<group>:<name>".

The full entry-point index is fetched once per process via a private _all_entry_points() helper wrapped in functools.lru_cache(maxsize=1). Entry points are statically declared in installed distribution metadata and do not change at runtime, so a single process-wide scan is safe. Across rfgen’s thirteen registries, this avoids thirteen full metadata scans.

Same-name conflict

Two installed packages declaring the same name in the same entry-point group raise RegistryError at registration time; the message names both source modules.

Missing-extra conflict

If a plugin’s PluginMetadata.requires lists an extra whose marker package is not installed, registration raises BackendUnavailableError, naming the missing extra and the pip install rfgen[<extra>] invocation that would fix it.

---

class rfgen.registry.PluginMetadata

from pydantic import BaseModel, ConfigDict, Field, field_validator
import packaging.version


class PluginMetadata(BaseModel):
    """Standardized descriptor for a plugin package."""

    model_config = ConfigDict(frozen=True)

    name: str = Field(min_length=1, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$")
    version: str
    family: str | None = None
    summary: str = Field(default="", max_length=500)
    requires: list[str] = Field(default_factory=list)
    homepage: str | None = None
    license: str = "Apache-2.0"

A frozen Pydantic v2 model. Validates on construction and is the single source of truth for what a plugin claims to provide and what extras it needs. Frozen instances are read-only after construction; this matches the frozen-by-default posture of the Layer 1 dataclasses and prevents silent corruption of registry state by code that holds a shared reference.

Fields

Field	Type	Required	Default	Description
name	str	yes	–	PyPI distribution name, e.g. "rfgen-myradar". Must match the documented identifier pattern.
version	str	yes	–	SemVer / PEP 440 version of the plugin package. Validated via packaging.version.Version.
family	str | None	no	None	Family tag (emitter family, channel backend tag, …).
summary	str	no	""	Short human-readable description, max 500 chars.
requires	list[str]	no	[]	List of pip extras that must be installed for this plugin to operate. Registration fails with BackendUnavailableError if any required extra is missing.
homepage	str | None	no	None	Project URL, typically the GitHub repo.
license	str	no	"Apache-2.0"	SPDX license identifier.

Validation

• name is rejected unless it matches ^[A-Za-z][A-Za-z0-9_-]*$.

• version is rejected if packaging.version.Version(v) raises InvalidVersion.

• summary longer than 500 characters is rejected.

• The model is frozen; assigning to a field after construction raises pydantic.ValidationError.

Example

from rfgen.registry import PluginMetadata

PLUGIN = PluginMetadata(
    name="rfgen-myradar",
    version="0.2.1",
    family="radar",
    summary="High-fidelity polyphase pulsed radar emitter for rfgen.",
    requires=["torchsig"],
    homepage="https://github.com/myorg/rfgen-myradar",
    license="Apache-2.0",
)

---

class rfgen.registry.BaseCommand

from typing import Protocol, runtime_checkable
from pydantic import BaseModel


@runtime_checkable
class BaseCommand(Protocol):
    """Dispatch contract for Layer 9 services callable from the CLI."""

    name: str
    summary: str
    params_schema: type[BaseModel]

    def run(
        self,
        *,
        params: BaseModel,
        registry: BaseRegistry[object],
        log_sink: object,
    ) -> int: ...

A runtime-checkable Protocol. Each Layer 10 service owns and registers its BaseCommand implementation under entry-point group rfgen.commands; the CLI dispatches to them by name through the registry, never through build-time imports. Generation is owned by rfgen.executors, annotation by rfgen.annotation_orchestrators, and dataset inspection by rfgen.audit. rfgen.commands contains compatibility reexports only.

Attributes

• name: canonical command name (used by the CLI for dispatch and listings).

• summary: one-line human description.

• params_schema: Pydantic model describing the parameter surface.

run(*, params, registry, log_sink)

• params: validated parameters; an instance of params_schema.

• registry: the plugin registry the command should resolve names through.

• log_sink: structured-log sink, typed as object here. The Layer 1 registry module cannot import the Layer 2 BaseLogSink type without introducing a forward layer dependency that would break parallel-build of Layer 1 and Layer 2; the type tightens to BaseLogSink in the Layer 9 CLI dispatcher signature where the import is already in scope.

Returns a process exit code from the closed set EXIT_CODES_VALID = frozenset({0, 1, 2, 3, 130}) (success, config error, runtime error, validation failure, keyboard interrupt). Any other return value is a contract violation; the contract test in validation-and-audit asserts every shipped command returns a member of EXIT_CODES_VALID.

---

End-to-end: shipping a third-party emitter plugin

This is the canonical path from “I wrote an emitter” to “rfgen finds and loads it.” No registry code in rfgen needs to change to support a new plugin; only the plugin author’s pyproject.toml does.

1. The plugin author writes a class

# rfgen_myradar/emitter.py
from rfgen.emitters import BaseEmitter

class MyRadarEmitter(BaseEmitter):
    """A polyphase pulsed radar emitter shipped as a third-party package."""
    ...

2. They declare an entry point in their pyproject.toml

[project]
name = "rfgen-myradar"
version = "0.2.1"

[project.entry-points."rfgen.emitters"]
my_radar = "rfgen_myradar.emitter:MyRadarEmitter"

The entry-point group rfgen.emitters is one of the thirteen documented groups. The key (my_radar) is the canonical name the framework will resolve through; the value is the dotted path to the factory.

3. The user installs both packages

pip install rfgen rfgen-myradar

4. rfgen discovers it via EntryPointRegistry

from rfgen.registry import EntryPointRegistry
from rfgen.emitters import BaseEmitter

emitters: EntryPointRegistry[BaseEmitter] = EntryPointRegistry("rfgen.emitters")
emitters.discover()                  # records 'my_radar' in the pending table
assert "my_radar" in emitters.available()

instance = emitters.get("my_radar")  # this is the call that imports rfgen_myradar.emitter
assert isinstance(instance, BaseEmitter)

The discover() call walks the cached entry-point index, filters by group, and records my_radar as pending. Importing rfgen_myradar.emitter does not happen until get("my_radar") is called.

5. End users hit this path through config, not directly

In normal use, a user writes a YAML config that names the emitter as a string:

emitters:
  - name: my_radar
    params: {...}

Hydra parses the YAML, the Pydantic emitter config validates the string at instantiation time, and the framework resolves it through the same EntryPointRegistry instance. The user never calls get directly.

Lazy-import contract

import rfgen.registry does not import any plugin module. Even registry.discover() does not. Only get(name) triggers the import for that specific entry. This keeps cold-start cost bounded by the number of plugins the user actually resolves, not the number installed on the system.

---

Optional hook-based discovery via pluggy

pluggy is the optional second primitive, gated behind the rfgen[plugin-hooks] extra. It is imported defensively in rfgen.registry:

try:
    import pluggy as _pluggy
    _PLUGGY_AVAILABLE = True
except ImportError:
    _PLUGGY_AVAILABLE = False

pluggy is the layered hook-mechanism behind pytest, tox, and devpi. It complements EntryPointRegistry for discovery patterns that go beyond “one name, one factory”, for example, a hook that lets multiple plugins register schema fragments under the same config key, or one that lets plugins observe pipeline events. Layer 8’s plugin-hub design uses pluggy for the “observe annotation events” hook surface; entry points remain the primary registration channel.

The base rfgen install does not require pluggy. Code that wants the hook surface checks _PLUGGY_AVAILABLE and degrades cleanly when the extra is not installed.

---

Why JsonManifestRegistry is intentionally omitted

Earlier drafts of this page proposed three additional concrete registries: JsonManifestRegistry, PrivateIndexRegistry, and CompositeRegistry. None of them ship.

The user-confirmed decision (recorded in .agent-state/rfgen-plan.log.md, pass-6 decisions log, 2026-06-21) is that importlib.metadata.entry_points plus pluggy already cover wheel-time and layered hook-based discovery. A manifest reader would grow maintenance surface (a JSON schema, a fetcher, a cache, an offline-mode flag, a validation pass on remote records) without adding capability the existing two primitives do not already provide. Private corporate registries are covered by configuring the local Python environment to install from a private index (the standard pip workflow), at which point EntryPointRegistry finds the plugin like any other installed package.

If a future need arises (e.g., shipping a frozen registry inside a single-file wheel where entry-point metadata is not available, or a hub that distributes plugin descriptors without distributing the wheels), the question can be re-opened in a new pass. The implementation in src/rfgen/registry.py carries an inline source comment naming the chosen primitives and pointing at this decision so the omission is not silent.

---

See Also

• Reference / Plugin Metadata: declaration site of PLUGIN: PluginMetadata on plugin packages, plus the PLUGIN_CARD.md convention.

• Reference / Project layout: the thirteen entry-point group names and the package directories that consume them.

• BackendUnavailableError, PluginNotFoundError, RegistryError: the error surface this module raises.

• How-to / Add a custom emitter: end-to-end package layout including the pyproject.toml entry-point declaration.

---

Legacy class names

Legacy: JsonManifestRegistry

Intentionally not shipped. See the “Why JsonManifestRegistry is intentionally omitted” section above for the pass-6 user-confirmed decision. Existing cross-references (e.g., from docs/background/open-questions.md) resolve to this anchor so the docs render; the class itself does not exist in rfgen.registry.