Use capability policies¶
Capability policies govern three moments: when two plugins claim the same
capability, when multiple providers exist and one must be chosen, and when no
provider exists. Each moment has its own field on Core.
Understand the three policy fields¶
Core accepts three keyword arguments that define capability resolution.
| Field | Controls | Values |
|---|---|---|
capability_collision |
What happens when a second plugin registers a capability already provided | "last_wins_with_warning" (default), "error_on_conflict", "first_wins" |
capability_selection |
Which provider is returned when multiple exist | "last_registered" (default), "first_registered" |
capability_missing |
What get_capability() does when no provider exists |
"raise" (default), "return_none" |
The defaults allow duplicate registrations and warn, select the newest provider,
and raise CapabilityError when nothing is found.
capability_access is a separate, fourth axis — security posture, not resolution policy. It controls which plugins may call get_capability() for a capability at all, independent of how the kernel selects among providers. See Lock down capability access and the secure capability access explanation.
Enforce strict uniqueness in production¶
-
Create a
Corewithcapability_collision="error_on_conflict". -
Register the first provider — this succeeds.
-
Attempt to register a second plugin that claims the same capability.
try: await core.register_plugin(StoragePluginB()) except PluginError as exc: print(exc) # "Capability 'storage' is already provided by: storage_plugin_a # (capability_collision policy is 'error_on_conflict')"PluginErroris raised beforeStoragePluginBis added to the registry; the check is atomic — if one capability collides, none from that plugin register.
Allow override during development¶
-
Create a
Corewithcapability_collision="last_wins_with_warning"andcapability_selection="last_registered". These are the defaults, so explicit kwargs are optional but make the intent readable. -
Register two providers for the same capability.
await core.start() await core.register_plugin(StoragePluginA()) await core.register_plugin(StoragePluginB())Both registrations succeed. The second logs a
WARNINGnaming both plugins. -
Call
get_capability()— the last-registered provider is returned.This replaces the real provider with a stub during development without touching the dependent plugin.
Freeze the first provider silently¶
-
Create a
Corewithcapability_collision="first_wins". -
Register two providers.
await core.start() await core.register_plugin(StoragePluginA()) await core.register_plugin(StoragePluginB())The second registration is silently ignored. A
DEBUG-level message is logged. -
Call
get_capability()— the first-registered provider is returned.Use this to pre-register a test stub; later registrations silently defer to it.
Handle optional capabilities gracefully¶
-
Create a
Corewithcapability_missing="return_none". -
Call
get_capability()for a capability that may not exist.async def process(self) -> None: cache = await self.get_capability("cache") if cache is not None: return await cache.get("key") return self.compute_slowly()With the default
"raise"policy, the call raisesCapabilityErrornaming every registered capability. Use"raise"when absence indicates a configuration mistake; use"return_none"when the capability is optional and a fallback exists.
Combine policies¶
The three fields are independent. Mix them to match each deployment's needs.
core = Core(
capability_collision="error_on_conflict", # fail fast on duplicates
capability_selection="first_registered", # deterministic when multiple exist
capability_missing="raise", # no silent degradation
)
Core validates all three fields in CoreConfig.__post_init__(), raising
ValueError for unrecognised values before any plugin is registered.
Related pages¶
- Work with capabilities — declare
providesandrequires, and callget_capability() - Use tags for provider selection — narrow a pool of providers at lookup time without changing global policy