cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(plugin): register spec-defined extension points in PluginManager #1217

Merged
freemo merged 1 commits from feature/plugin-extension-points into master 2026-04-02 17:40:19 +00:00
Conversation 8 Commits 1 Files Changed 7
+1738 -1
7 changed files with 1738 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/plugin_extension_points.feature
+229
View File
@@ -0,0 +1,229 @@
Feature: Plugin Extension Points Registration
The plugin infrastructure must register all 30 spec-defined extension
points with typed Protocol interfaces in the PluginManager.
Based on issue #939.
# -----------------------------------------------------------------------
# All 30 extension points are registered
# -----------------------------------------------------------------------
Scenario: All 30 extension points are registered via register_all_extension_points
Given a fresh PluginManager for extension point registration
When I register all spec-defined extension points
Then exactly 30 extension points should be registered
Scenario: Extension point definitions are available as a tuple
When I retrieve the extension point definitions
Then I should get exactly 30 extension point definitions
And each definition should have a non-empty name
And each definition should have a protocol_type
And each definition should have a non-empty description
# -----------------------------------------------------------------------
# Extension point lookup by name
# -----------------------------------------------------------------------
Scenario: Extension point lookup by name returns the correct point
Given a PluginManager with all extension points registered
When I look up extension point "context.strategy"
Then the looked-up extension point name should be "context.strategy"
Scenario: Extension point lookup for pipeline component by name
Given a PluginManager with all extension points registered
When I look up extension point "context.pipeline_component.scorer"
Then the looked-up extension point name should be "context.pipeline_component.scorer"
Scenario: Extension point lookup for safety guardrail by name
Given a PluginManager with all extension points registered
When I look up extension point "safety.guardrail"
Then the looked-up extension point name should be "safety.guardrail"
Scenario: Lookup of non-existent extension point returns None
Given a PluginManager with all extension points registered
When I look up extension point "nonexistent.point"
Then the looked-up extension point should be None
# -----------------------------------------------------------------------
# Extension point lookup by category
# -----------------------------------------------------------------------
Scenario: Extension points grouped by category have correct counts
When I retrieve extension points grouped by category
Then the "context" category should have 12 extension points
And the "output" category should have 3 extension points
And the "validation" category should have 2 extension points
And the "tool" category should have 2 extension points
And the "skill" category should have 2 extension points
And the "resource" category should have 2 extension points
And the "a2a" category should have 2 extension points
And the "event" category should have 2 extension points
And the "config" category should have 2 extension points
And the "safety" category should have 1 extension points
# -----------------------------------------------------------------------
# Protocol type correctness
# -----------------------------------------------------------------------
Scenario: Context strategy extension point has correct protocol type
Given a PluginManager with all extension points registered
When I look up extension point "context.strategy"
Then the protocol type name should be "ContextStrategyExtension"
Scenario: Output renderer extension point has correct protocol type
Given a PluginManager with all extension points registered
When I look up extension point "output.renderer"
Then the protocol type name should be "OutputRendererExtension"
Scenario: Safety guardrail extension point has correct protocol type
Given a PluginManager with all extension points registered
When I look up extension point "safety.guardrail"
Then the protocol type name should be "SafetyGuardrailExtension"
# -----------------------------------------------------------------------
# Plugin registration via entry points
# -----------------------------------------------------------------------
Scenario: Plugin registration via entry points discovers no plugins by default
Given a fresh PluginManager for extension point registration
When I discover extension point plugins from group "cleveragents.ext.test"
Then the entry point discovered list should be empty
# -----------------------------------------------------------------------
# TOTAL_EXTENSION_POINTS constant
# -----------------------------------------------------------------------
Scenario: TOTAL_EXTENSION_POINTS constant equals 30
Then the TOTAL_EXTENSION_POINTS constant should equal 30
# -----------------------------------------------------------------------
# Context extension points wired to ACMSPipeline
# -----------------------------------------------------------------------
Scenario: ACMSPipeline reports context extension points when plugin manager is wired
Given a PluginManager with all extension points registered
And an ACMSPipeline wired with the plugin manager
Then the pipeline should report 12 context extension points
Scenario: ACMSPipeline reports no context extension points when no plugin manager
Given an ACMSPipeline without a plugin manager
Then the pipeline should report 0 context extension points
# -----------------------------------------------------------------------
# All extension protocols are runtime-checkable
# -----------------------------------------------------------------------
Scenario: All extension protocol types are runtime-checkable
When I retrieve all extension protocol types
Then every protocol should be runtime-checkable
# -----------------------------------------------------------------------
# Idempotent registration
# -----------------------------------------------------------------------
Scenario: Registering extension points twice does not duplicate
Given a fresh PluginManager for extension point registration
When I register all spec-defined extension points
And I register all spec-defined extension points again
Then exactly 30 extension points should be registered
# -----------------------------------------------------------------------
# Extension point registry_key matches category
# -----------------------------------------------------------------------
Scenario: Each extension point registry_key matches its category
When I retrieve the extension point definitions
Then each definition registry_key should match its name category prefix
# -----------------------------------------------------------------------
# Protocol implementation conformance
# -----------------------------------------------------------------------
Scenario: A concrete context strategy satisfies ContextStrategyExtension
Given a concrete ContextStrategyExtension implementation
Then it should be recognized as a ContextStrategyExtension instance
And calling its protocol methods should succeed
Scenario: A concrete pipeline component satisfies ContextPipelineComponentExtension
Given a concrete ContextPipelineComponentExtension implementation
Then it should be recognized as a ContextPipelineComponentExtension instance
And calling its pipeline component methods should succeed
Scenario: A concrete storage backend satisfies ContextStorageBackendExtension
Given a concrete ContextStorageBackendExtension implementation
Then it should be recognized as a ContextStorageBackendExtension instance
And calling its storage methods should succeed
Scenario: A concrete output renderer satisfies OutputRendererExtension
Given a concrete OutputRendererExtension implementation
Then it should be recognized as an OutputRendererExtension instance
And calling its renderer methods should succeed
Scenario: A concrete output materializer satisfies OutputMaterializerExtension
Given a concrete OutputMaterializerExtension implementation
Then it should be recognized as an OutputMaterializerExtension instance
Scenario: A concrete output format satisfies OutputFormatExtension
Given a concrete OutputFormatExtension implementation
Then it should be recognized as an OutputFormatExtension instance
Scenario: A concrete validation runner satisfies ValidationRunnerExtension
Given a concrete ValidationRunnerExtension implementation
Then it should be recognized as a ValidationRunnerExtension instance
Scenario: A concrete validation rule provider satisfies ValidationRuleProviderExtension
Given a concrete ValidationRuleProviderExtension implementation
Then it should be recognized as a ValidationRuleProviderExtension instance
Scenario: A concrete tool provider satisfies ToolProviderExtension
Given a concrete ToolProviderExtension implementation
Then it should be recognized as a ToolProviderExtension instance
Scenario: A concrete tool middleware satisfies ToolMiddlewareExtension
Given a concrete ToolMiddlewareExtension implementation
Then it should be recognized as a ToolMiddlewareExtension instance
Scenario: A concrete skill provider satisfies SkillProviderExtension
Given a concrete SkillProviderExtension implementation
Then it should be recognized as a SkillProviderExtension instance
Scenario: A concrete skill template satisfies SkillTemplateExtension
Given a concrete SkillTemplateExtension implementation
Then it should be recognized as a SkillTemplateExtension instance
Scenario: A concrete resource resolver satisfies ResourceResolverExtension
Given a concrete ResourceResolverExtension implementation
Then it should be recognized as a ResourceResolverExtension instance
Scenario: A concrete resource type handler satisfies ResourceTypeHandlerExtension
Given a concrete ResourceTypeHandlerExtension implementation
Then it should be recognized as a ResourceTypeHandlerExtension instance
Scenario: A concrete A2A transport satisfies A2ATransportExtension
Given a concrete A2ATransportExtension implementation
Then it should be recognized as an A2ATransportExtension instance
Scenario: A concrete A2A extension method satisfies A2AExtensionMethodExtension
Given a concrete A2AExtensionMethodExtension implementation
Then it should be recognized as an A2AExtensionMethodExtension instance
Scenario: A concrete event handler satisfies EventHandlerExtension
Given a concrete EventHandlerExtension implementation
Then it should be recognized as an EventHandlerExtension instance
Scenario: A concrete event filter satisfies EventFilterExtension
Given a concrete EventFilterExtension implementation
Then it should be recognized as an EventFilterExtension instance
Scenario: A concrete config source satisfies ConfigSourceExtension
Given a concrete ConfigSourceExtension implementation
Then it should be recognized as a ConfigSourceExtension instance
Scenario: A concrete config validator satisfies ConfigValidatorExtension
Given a concrete ConfigValidatorExtension implementation
Then it should be recognized as a ConfigValidatorExtension instance
Scenario: A concrete safety guardrail satisfies SafetyGuardrailExtension
Given a concrete SafetyGuardrailExtension implementation
Then it should be recognized as a SafetyGuardrailExtension instance
And calling its guardrail methods should succeed
features/steps/plugin_extension_points_steps.py
+787
View File
@@ -0,0 +1,787 @@
"""Step definitions for plugin_extension_points.feature.
Tests all 30 spec-defined extension points: registration, lookup
by name, lookup by category, protocol type correctness, entry point
discovery, context subsystem wiring, and idempotent registration.
Based on issue #939.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.infrastructure.plugins.extension_catalog import (
TOTAL_EXTENSION_POINTS,
get_extension_point_definitions,
get_extension_points_by_category,
register_all_extension_points,
)
from cleveragents.infrastructure.plugins.extension_protocols import (
ALL_EXTENSION_PROTOCOLS,
A2AExtensionMethodExtension,
A2ATransportExtension,
ConfigSourceExtension,
ConfigValidatorExtension,
ContextPipelineComponentExtension,
ContextStorageBackendExtension,
ContextStrategyExtension,
EventFilterExtension,
EventHandlerExtension,
OutputFormatExtension,
OutputMaterializerExtension,
OutputRendererExtension,
ResourceResolverExtension,
ResourceTypeHandlerExtension,
SafetyGuardrailExtension,
SkillProviderExtension,
SkillTemplateExtension,
ToolMiddlewareExtension,
ToolProviderExtension,
ValidationRuleProviderExtension,
ValidationRunnerExtension,
)
from cleveragents.infrastructure.plugins.manager import PluginManager
from cleveragents.infrastructure.plugins.types import ExtensionPoint
# ---------------------------------------------------------------------------
# Fresh PluginManager
# ---------------------------------------------------------------------------
@given("a fresh PluginManager for extension point registration")
def step_fresh_pm(context: Context) -> None:
context.ep_manager = PluginManager()
@given("a PluginManager with all extension points registered")
def step_pm_with_all_eps(context: Context) -> None:
context.ep_manager = PluginManager()
register_all_extension_points(context.ep_manager)
# ---------------------------------------------------------------------------
# Register all extension points
# ---------------------------------------------------------------------------
@when("I register all spec-defined extension points")
def step_register_all(context: Context) -> None:
context.ep_registered_count = register_all_extension_points(
context.ep_manager,
)
@when("I register all spec-defined extension points again")
def step_register_all_again(context: Context) -> None:
register_all_extension_points(context.ep_manager)
@then("exactly {count:d} extension points should be registered")
def step_exactly_n_registered(context: Context, count: int) -> None:
eps = context.ep_manager.list_extension_points()
assert len(eps) == count, f"Expected {count}, got {len(eps)}"
# ---------------------------------------------------------------------------
# Extension point definitions
# ---------------------------------------------------------------------------
@when("I retrieve the extension point definitions")
def step_retrieve_defs(context: Context) -> None:
context.ep_definitions = get_extension_point_definitions()
@then("I should get exactly {count:d} extension point definitions")
def step_defs_count(context: Context, count: int) -> None:
assert len(context.ep_definitions) == count, (
f"Expected {count}, got {len(context.ep_definitions)}"
)
@then("each definition should have a non-empty name")
def step_each_def_has_name(context: Context) -> None:
for ep in context.ep_definitions:
assert ep.name, f"Extension point has empty name: {ep}"
@then("each definition should have a protocol_type")
def step_each_def_has_protocol(context: Context) -> None:
for ep in context.ep_definitions:
assert ep.protocol_type is not None, (
f"Extension point {ep.name} has no protocol_type"
)
@then("each definition should have a non-empty description")
def step_each_def_has_description(context: Context) -> None:
for ep in context.ep_definitions:
assert ep.description, (
f"Extension point {ep.name} has empty description"
)
# ---------------------------------------------------------------------------
# Lookup by name
# ---------------------------------------------------------------------------
@when('I look up extension point "{name}"')
def step_lookup_by_name(context: Context, name: str) -> None:
eps = context.ep_manager.list_extension_points()
matches = [ep for ep in eps if ep.name == name]
context.ep_lookup_result: ExtensionPoint | None = (
matches[0] if matches else None
)
@then('the looked-up extension point name should be "{name}"')
def step_lookup_name_is(context: Context, name: str) -> None:
assert context.ep_lookup_result is not None, (
f"Extension point '{name}' not found"
)
assert context.ep_lookup_result.name == name
@then("the looked-up extension point should be None")
def step_lookup_is_none(context: Context) -> None:
assert context.ep_lookup_result is None
# ---------------------------------------------------------------------------
# Lookup by category
# ---------------------------------------------------------------------------
@when("I retrieve extension points grouped by category")
def step_grouped_by_category(context: Context) -> None:
context.ep_categories = get_extension_points_by_category()
@then('the "{category}" category should have {count:d} extension points')
def step_category_count(context: Context, category: str, count: int) -> None:
actual = len(context.ep_categories.get(category, []))
assert actual == count, (
f"Category '{category}': expected {count}, got {actual}"
)
# ---------------------------------------------------------------------------
# Protocol type correctness
# ---------------------------------------------------------------------------
@then('the protocol type name should be "{proto_name}"')
def step_protocol_type_name(context: Context, proto_name: str) -> None:
assert context.ep_lookup_result is not None
actual = context.ep_lookup_result.protocol_type.__name__
assert actual == proto_name, f"Expected '{proto_name}', got '{actual}'"
# ---------------------------------------------------------------------------
# Entry point discovery
# ---------------------------------------------------------------------------
@when('I discover extension point plugins from group "{group}"')
def step_discover_ep_group(context: Context, group: str) -> None:
context.ep_discovered = context.ep_manager.discover(group=group)
@then("the entry point discovered list should be empty")
def step_ep_discovered_empty(context: Context) -> None:
assert len(context.ep_discovered) == 0
# ---------------------------------------------------------------------------
# TOTAL_EXTENSION_POINTS constant
# ---------------------------------------------------------------------------
@then("the TOTAL_EXTENSION_POINTS constant should equal {value:d}")
def step_total_constant(context: Context, value: int) -> None:
assert value == TOTAL_EXTENSION_POINTS, (
f"Expected {value}, got {TOTAL_EXTENSION_POINTS}"
)
# ---------------------------------------------------------------------------
# Context extension point wiring in ACMSPipeline
# ---------------------------------------------------------------------------
@given("an ACMSPipeline wired with the plugin manager")
def step_pipeline_with_pm(context: Context) -> None:
from cleveragents.application.services.acms_service import ACMSPipeline
context.ep_pipeline = ACMSPipeline(
plugin_manager=context.ep_manager,
)
@given("an ACMSPipeline without a plugin manager")
def step_pipeline_without_pm(context: Context) -> None:
from cleveragents.application.services.acms_service import ACMSPipeline
context.ep_pipeline = ACMSPipeline()
@then("the pipeline should report {count:d} context extension points")
def step_pipeline_context_eps(context: Context, count: int) -> None:
actual = len(context.ep_pipeline.context_extension_points)
assert actual == count, (
f"Expected {count} context extension points, got {actual}"
)
# ---------------------------------------------------------------------------
# Runtime-checkable protocols
# ---------------------------------------------------------------------------
@when("I retrieve all extension protocol types")
def step_retrieve_all_protocols(context: Context) -> None:
context.ep_protocol_map = ALL_EXTENSION_PROTOCOLS
@then("every protocol should be runtime-checkable")
def step_all_protocols_runtime_checkable(context: Context) -> None:
for name, proto in context.ep_protocol_map.items():
# runtime_checkable protocols have _is_runtime_protocol = True
is_rc: Any = getattr(proto, "_is_runtime_protocol", False)
assert is_rc, (
f"Protocol {name} ({proto.__name__}) is not runtime-checkable"
)
# ---------------------------------------------------------------------------
# registry_key matches category
# ---------------------------------------------------------------------------
@then("each definition registry_key should match its name category prefix")
def step_registry_key_matches(context: Context) -> None:
for ep in context.ep_definitions:
expected_category = ep.name.split(".")[0]
assert ep.registry_key == expected_category, (
f"Extension point '{ep.name}': registry_key '{ep.registry_key}' "
f"does not match expected category '{expected_category}'"
)
# ---------------------------------------------------------------------------
# Concrete protocol implementations for coverage
# ---------------------------------------------------------------------------
class _StubContextStrategy:
@property
def name(self) -> str:
return "stub"
def can_handle(self, request: Any) -> float:
return 1.0
def assemble(self, fragments: Any, budget: Any) -> list[Any]:
return []
class _StubPipelineComponent:
@property
def component_name(self) -> str:
return "stub_component"
def process(self, fragments: Any, context: Any) -> list[Any]:
return list(fragments)
class _StubStorageBackend:
def __init__(self) -> None:
self._store: dict[str, Any] = {}
@property
def backend_type(self) -> str:
return "memory"
def store(self, key: str, data: Any) -> None:
self._store[key] = data
def retrieve(self, key: str) -> Any:
return self._store.get(key)
class _StubOutputRenderer:
@property
def format_name(self) -> str:
return "text"
def render(self, data: Any) -> str:
return str(data)
class _StubOutputMaterializer:
@property
def target_type(self) -> str:
return "stdout"
def materialize(self, content: str, destination: str) -> None:
pass
class _StubOutputFormat:
@property
def format_id(self) -> str:
return "custom"
def serialize(self, data: Any) -> str:
return str(data)
def deserialize(self, raw: str) -> Any:
return raw
class _StubValidationRunner:
@property
def runner_name(self) -> str:
return "stub_runner"
def validate(self, target: Any) -> list[Any]:
return []
class _StubValidationRuleProvider:
@property
def provider_name(self) -> str:
return "stub_provider"
def get_rules(self) -> list[Any]:
return []
class _StubToolProvider:
@property
def provider_name(self) -> str:
return "stub_tools"
def list_tools(self) -> list[str]:
return ["tool1"]
def invoke(self, tool_name: str, arguments: Any) -> Any:
return None
class _StubToolMiddleware:
@property
def middleware_name(self) -> str:
return "stub_mw"
def before_invoke(self, tool_name: str, arguments: Any) -> dict[str, Any]:
return dict(arguments)
def after_invoke(self, tool_name: str, result: Any) -> Any:
return result
class _StubSkillProvider:
@property
def provider_name(self) -> str:
return "stub_skills"
def list_skills(self) -> list[str]:
return ["skill1"]
def get_skill(self, skill_name: str) -> Any:
return None
class _StubSkillTemplate:
@property
def template_name(self) -> str:
return "stub_template"
def render(self, context: Any) -> str:
return "rendered"
class _StubResourceResolver:
@property
def resolver_name(self) -> str:
return "stub_resolver"
def resolve(self, uri: str) -> Any:
return None
class _StubResourceTypeHandler:
@property
def handled_type(self) -> str:
return "file"
def can_handle(self, resource_type: str) -> bool:
return resource_type == "file"
def load(self, uri: str) -> Any:
return None
class _StubA2ATransport:
@property
def transport_name(self) -> str:
return "stub_transport"
def send(self, destination: str, payload: Any) -> None:
pass
def receive(self) -> Any:
return None
class _StubA2AExtensionMethod:
@property
def method_name(self) -> str:
return "stub_method"
def execute(self, params: Any) -> Any:
return None
class _StubEventHandler:
@property
def handler_name(self) -> str:
return "stub_handler"
def handle(self, event: Any) -> None:
pass
class _StubEventFilter:
@property
def filter_name(self) -> str:
return "stub_filter"
def should_propagate(self, event: Any) -> bool:
return True
class _StubConfigSource:
@property
def source_name(self) -> str:
return "stub_source"
def load(self) -> dict[str, Any]:
return {}
class _StubConfigValidator:
@property
def validator_name(self) -> str:
return "stub_validator"
def validate(self, config: Any) -> list[str]:
return []
class _StubSafetyGuardrail:
@property
def guardrail_name(self) -> str:
return "stub_guardrail"
def check(self, action: Any) -> bool:
return True
def explain(self) -> str:
return "Always allows"
# --- Context Strategy ---
@given("a concrete ContextStrategyExtension implementation")
def step_concrete_ctx_strategy(context: Context) -> None:
context.ep_impl = _StubContextStrategy()
@then("it should be recognized as a ContextStrategyExtension instance")
def step_isinstance_ctx_strategy(context: Context) -> None:
assert isinstance(context.ep_impl, ContextStrategyExtension)
@then("calling its protocol methods should succeed")
def step_call_ctx_strategy_methods(context: Context) -> None:
impl = context.ep_impl
assert impl.name == "stub"
assert impl.can_handle({}) == 1.0
assert impl.assemble([], None) == []
# --- Pipeline Component ---
@given("a concrete ContextPipelineComponentExtension implementation")
def step_concrete_pipeline(context: Context) -> None:
context.ep_impl = _StubPipelineComponent()
@then("it should be recognized as a ContextPipelineComponentExtension instance")
def step_isinstance_pipeline(context: Context) -> None:
assert isinstance(context.ep_impl, ContextPipelineComponentExtension)
@then("calling its pipeline component methods should succeed")
def step_call_pipeline_methods(context: Context) -> None:
impl = context.ep_impl
assert impl.component_name == "stub_component"
assert impl.process(["a", "b"], {}) == ["a", "b"]
# --- Storage Backend ---
@given("a concrete ContextStorageBackendExtension implementation")
def step_concrete_storage(context: Context) -> None:
context.ep_impl = _StubStorageBackend()
@then("it should be recognized as a ContextStorageBackendExtension instance")
def step_isinstance_storage(context: Context) -> None:
assert isinstance(context.ep_impl, ContextStorageBackendExtension)
@then("calling its storage methods should succeed")
def step_call_storage_methods(context: Context) -> None:
impl = context.ep_impl
assert impl.backend_type == "memory"
impl.store("k", "v")
assert impl.retrieve("k") == "v"
# --- Output Renderer ---
@given("a concrete OutputRendererExtension implementation")
def step_concrete_renderer(context: Context) -> None:
context.ep_impl = _StubOutputRenderer()
@then("it should be recognized as an OutputRendererExtension instance")
def step_isinstance_renderer(context: Context) -> None:
assert isinstance(context.ep_impl, OutputRendererExtension)
@then("calling its renderer methods should succeed")
def step_call_renderer_methods(context: Context) -> None:
impl = context.ep_impl
assert impl.format_name == "text"
assert impl.render(42) == "42"
# --- Output Materializer ---
@given("a concrete OutputMaterializerExtension implementation")
def step_concrete_materializer(context: Context) -> None:
context.ep_impl = _StubOutputMaterializer()
@then("it should be recognized as an OutputMaterializerExtension instance")
def step_isinstance_materializer(context: Context) -> None:
assert isinstance(context.ep_impl, OutputMaterializerExtension)
# --- Output Format ---
@given("a concrete OutputFormatExtension implementation")
def step_concrete_format(context: Context) -> None:
context.ep_impl = _StubOutputFormat()
@then("it should be recognized as an OutputFormatExtension instance")
def step_isinstance_format(context: Context) -> None:
assert isinstance(context.ep_impl, OutputFormatExtension)
# --- Validation Runner ---
@given("a concrete ValidationRunnerExtension implementation")
def step_concrete_val_runner(context: Context) -> None:
context.ep_impl = _StubValidationRunner()
@then("it should be recognized as a ValidationRunnerExtension instance")
def step_isinstance_val_runner(context: Context) -> None:
assert isinstance(context.ep_impl, ValidationRunnerExtension)
# --- Validation Rule Provider ---
@given("a concrete ValidationRuleProviderExtension implementation")
def step_concrete_val_rule(context: Context) -> None:
context.ep_impl = _StubValidationRuleProvider()
@then("it should be recognized as a ValidationRuleProviderExtension instance")
def step_isinstance_val_rule(context: Context) -> None:
assert isinstance(context.ep_impl, ValidationRuleProviderExtension)
# --- Tool Provider ---
@given("a concrete ToolProviderExtension implementation")
def step_concrete_tool_provider(context: Context) -> None:
context.ep_impl = _StubToolProvider()
@then("it should be recognized as a ToolProviderExtension instance")
def step_isinstance_tool_provider(context: Context) -> None:
assert isinstance(context.ep_impl, ToolProviderExtension)
# --- Tool Middleware ---
@given("a concrete ToolMiddlewareExtension implementation")
def step_concrete_tool_mw(context: Context) -> None:
context.ep_impl = _StubToolMiddleware()
@then("it should be recognized as a ToolMiddlewareExtension instance")
def step_isinstance_tool_mw(context: Context) -> None:
assert isinstance(context.ep_impl, ToolMiddlewareExtension)
# --- Skill Provider ---
@given("a concrete SkillProviderExtension implementation")
def step_concrete_skill_provider(context: Context) -> None:
context.ep_impl = _StubSkillProvider()
@then("it should be recognized as a SkillProviderExtension instance")
def step_isinstance_skill_provider(context: Context) -> None:
assert isinstance(context.ep_impl, SkillProviderExtension)
# --- Skill Template ---
@given("a concrete SkillTemplateExtension implementation")
def step_concrete_skill_template(context: Context) -> None:
context.ep_impl = _StubSkillTemplate()
@then("it should be recognized as a SkillTemplateExtension instance")
def step_isinstance_skill_template(context: Context) -> None:
assert isinstance(context.ep_impl, SkillTemplateExtension)
# --- Resource Resolver ---
@given("a concrete ResourceResolverExtension implementation")
def step_concrete_res_resolver(context: Context) -> None:
context.ep_impl = _StubResourceResolver()
@then("it should be recognized as a ResourceResolverExtension instance")
def step_isinstance_res_resolver(context: Context) -> None:
assert isinstance(context.ep_impl, ResourceResolverExtension)
# --- Resource Type Handler ---
@given("a concrete ResourceTypeHandlerExtension implementation")
def step_concrete_res_handler(context: Context) -> None:
context.ep_impl = _StubResourceTypeHandler()
@then("it should be recognized as a ResourceTypeHandlerExtension instance")
def step_isinstance_res_handler(context: Context) -> None:
assert isinstance(context.ep_impl, ResourceTypeHandlerExtension)
# --- A2A Transport ---
@given("a concrete A2ATransportExtension implementation")
def step_concrete_a2a_transport(context: Context) -> None:
context.ep_impl = _StubA2ATransport()
@then("it should be recognized as an A2ATransportExtension instance")
def step_isinstance_a2a_transport(context: Context) -> None:
assert isinstance(context.ep_impl, A2ATransportExtension)
# --- A2A Extension Method ---
@given("a concrete A2AExtensionMethodExtension implementation")
def step_concrete_a2a_method(context: Context) -> None:
context.ep_impl = _StubA2AExtensionMethod()
@then("it should be recognized as an A2AExtensionMethodExtension instance")
def step_isinstance_a2a_method(context: Context) -> None:
assert isinstance(context.ep_impl, A2AExtensionMethodExtension)
# --- Event Handler ---
@given("a concrete EventHandlerExtension implementation")
def step_concrete_event_handler(context: Context) -> None:
context.ep_impl = _StubEventHandler()
@then("it should be recognized as an EventHandlerExtension instance")
def step_isinstance_event_handler(context: Context) -> None:
assert isinstance(context.ep_impl, EventHandlerExtension)
# --- Event Filter ---
@given("a concrete EventFilterExtension implementation")
def step_concrete_event_filter(context: Context) -> None:
context.ep_impl = _StubEventFilter()
@then("it should be recognized as an EventFilterExtension instance")
def step_isinstance_event_filter(context: Context) -> None:
assert isinstance(context.ep_impl, EventFilterExtension)
# --- Config Source ---
@given("a concrete ConfigSourceExtension implementation")
def step_concrete_config_source(context: Context) -> None:
context.ep_impl = _StubConfigSource()
@then("it should be recognized as a ConfigSourceExtension instance")
def step_isinstance_config_source(context: Context) -> None:
assert isinstance(context.ep_impl, ConfigSourceExtension)
# --- Config Validator ---
@given("a concrete ConfigValidatorExtension implementation")
def step_concrete_config_validator(context: Context) -> None:
context.ep_impl = _StubConfigValidator()
@then("it should be recognized as a ConfigValidatorExtension instance")
def step_isinstance_config_validator(context: Context) -> None:
assert isinstance(context.ep_impl, ConfigValidatorExtension)
# --- Safety Guardrail ---
@given("a concrete SafetyGuardrailExtension implementation")
def step_concrete_safety(context: Context) -> None:
context.ep_impl = _StubSafetyGuardrail()
@then("it should be recognized as a SafetyGuardrailExtension instance")
def step_isinstance_safety(context: Context) -> None:
assert isinstance(context.ep_impl, SafetyGuardrailExtension)
@then("calling its guardrail methods should succeed")
def step_call_safety_methods(context: Context) -> None:
impl = context.ep_impl
assert impl.guardrail_name == "stub_guardrail"
assert impl.check("some_action") is True
assert impl.explain() == "Always allows"
src/cleveragents/application/container.py
+17
View File
@@ -103,6 +103,9 @@ from cleveragents.infrastructure.database.repositories import (
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
from cleveragents.infrastructure.events.reactive import ReactiveEventBus
from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
from cleveragents.infrastructure.plugins.extension_catalog import (
register_all_extension_points,
)
from cleveragents.infrastructure.plugins.manager import PluginManager
from cleveragents.providers.registry import ProviderRegistry, get_provider_registry
from cleveragents.shared.redaction import redact_value
@@ -771,6 +774,7 @@ class Container(containers.DeclarativeContainer):
ACMSPipeline,
settings=settings,
unit_of_work=unit_of_work,
plugin_manager=plugin_manager,
)
# ACMS ResourceFileWatcher — watches resource files for changes (#578).
@@ -856,6 +860,19 @@ def get_container() -> Container:
error_type=type(exc).__name__,
error_message=redact_value(str(exc)),
)
# Register all 30 spec-defined extension points on the
# PluginManager singleton so plugins can be discovered and
# validated against typed Protocol interfaces (#939).
try:
pm = _container.plugin_manager()
register_all_extension_points(pm)
except Exception as exc:
_logger.warning(
"extension_points_deferred",
reason="Failed to register extension points at startup.",
error_type=type(exc).__name__,
error_message=redact_value(str(exc)),
)
return _container
src/cleveragents/application/services/acms_service.py
+49
View File
@@ -33,6 +33,7 @@ import structlog
if TYPE_CHECKING:
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
from cleveragents.infrastructure.plugins.manager import PluginManager
from cleveragents.domain.models.acms.crp import (
ContextRequest,
@@ -626,6 +627,7 @@ class ACMSPipeline:
*,
settings: Settings | None = None,
unit_of_work: UnitOfWork | None = None,
plugin_manager: PluginManager | None = None,
# Phase 1 — Strategy Orchestration
strategy_selector: StrategySelector | None = None,
budget_allocator: BudgetAllocator | None = None,
@@ -672,6 +674,53 @@ class ACMSPipeline:
self._preamble_generator = preamble_generator or DefaultPreambleGenerator()
self._skeleton_compressor = skeleton_compressor or DefaultSkeletonCompressor()
# Plugin extension point integration (#939).
# Store the plugin manager reference so discovered context.*
# extension points can be queried at runtime.
self._plugin_manager = plugin_manager
self._discover_context_extensions()
# ------------------------------------------------------------------
# Plugin extension point discovery
# ------------------------------------------------------------------
def _discover_context_extensions(self) -> None:
"""Query context.* extension points from the PluginManager.
Logs discovered extension points for observability. Plugin
implementations are not auto-activated here — they are wired
on demand when a matching strategy or pipeline component is
requested.
"""
if self._plugin_manager is None:
return
context_eps = [
ep
for ep in self._plugin_manager.list_extension_points()
if ep.name.startswith("context.")
]
if context_eps:
self._logger.info(
"context_extensions_discovered",
count=len(context_eps),
names=[ep.name for ep in context_eps],
)
@property
def context_extension_points(self) -> list[str]:
"""Return names of registered context.* extension points.
Returns an empty list if no plugin manager is wired.
"""
if self._plugin_manager is None:
return []
return [
ep.name
for ep in self._plugin_manager.list_extension_points()
if ep.name.startswith("context.")
]
# ------------------------------------------------------------------
# Budget enforcement helper
# ------------------------------------------------------------------
src/cleveragents/infrastructure/plugins/__init__.py
+13 -1
View File
@@ -10,9 +10,11 @@ Key components:
- :class:`PluginManager` — lifecycle management (discover/activate/deactivate).
- :class:`PluginDescriptor` — immutable metadata for a discovered plugin.
- :class:`PluginState` — lifecycle state enum.
- :func:`register_all_extension_points` — registers all 30 spec-defined
extension points.
Based on ``docs/specification.md`` Extension Points Summary and
issue #585.
issue #585, #939.
ISSUES CLOSED: #585
"""
@@ -25,6 +27,12 @@ from cleveragents.infrastructure.plugins.exceptions import (
PluginNotFoundError,
ProtocolMismatchError,
)
from cleveragents.infrastructure.plugins.extension_catalog import (
TOTAL_EXTENSION_POINTS,
get_extension_point_definitions,
get_extension_points_by_category,
register_all_extension_points,
)
from cleveragents.infrastructure.plugins.loader import PluginLoader
from cleveragents.infrastructure.plugins.manager import PluginManager
from cleveragents.infrastructure.plugins.types import (
@@ -34,6 +42,7 @@ from cleveragents.infrastructure.plugins.types import (
)
__all__ = [
"TOTAL_EXTENSION_POINTS",
"ExtensionPoint",
"PluginDescriptor",
"PluginError",
@@ -43,4 +52,7 @@ __all__ = [
"PluginNotFoundError",
"PluginState",
"ProtocolMismatchError",
"get_extension_point_definitions",
"get_extension_points_by_category",
"register_all_extension_points",
]
src/cleveragents/infrastructure/plugins/extension_catalog.py
+306
View File
@@ -0,0 +1,306 @@
"""Extension point catalog for all 30 spec-defined extension points.
Provides :func:`register_all_extension_points` which registers every
extension point defined in ``docs/specification.md`` into the
:class:`PluginManager`.
Each extension point has a name, description, Protocol type, and
category derived from its dotted name prefix.
Based on ``docs/specification.md`` Extension Points Summary and
issue #939.
ISSUES CLOSED: #939
"""
from __future__ import annotations
from typing import Any, Protocol
from cleveragents.infrastructure.plugins.extension_protocols import (
A2AExtensionMethodExtension,
A2ATransportExtension,
ConfigSourceExtension,
ConfigValidatorExtension,
ContextPipelineComponentExtension,
ContextStorageBackendExtension,
ContextStrategyExtension,
EventFilterExtension,
EventHandlerExtension,
OutputFormatExtension,
OutputMaterializerExtension,
OutputRendererExtension,
ResourceResolverExtension,
ResourceTypeHandlerExtension,
SafetyGuardrailExtension,
SkillProviderExtension,
SkillTemplateExtension,
ToolMiddlewareExtension,
ToolProviderExtension,
ValidationRuleProviderExtension,
ValidationRunnerExtension,
)
from cleveragents.infrastructure.plugins.types import ExtensionPoint
# ---------------------------------------------------------------------------
# Structural protocol so the catalog is not tightly coupled to
# PluginManager — any object with ``register_extension_point`` works.
# ---------------------------------------------------------------------------
class ExtensionPointRegistrar(Protocol):
"""Structural type for :func:`register_all_extension_points`.
Any object with a ``register_extension_point`` method satisfies
this protocol (e.g. :class:`PluginManager`).
"""
def register_extension_point(self, extension_point: ExtensionPoint) -> None:
"""Register a single extension point."""
...
# ---------------------------------------------------------------------------
# Extension point definitions
# ---------------------------------------------------------------------------
class _ExtensionPointDef:
"""Internal descriptor for building ExtensionPoint instances."""
__slots__ = ("description", "name", "protocol_type")
def __init__(
self, name: str, protocol_type: type[Any], description: str
) -> None:
self.name = name
self.protocol_type = protocol_type
self.description = description
def _category_from_name(name: str) -> str:
"""Extract category prefix from a dotted extension point name.
Examples::
>>> _category_from_name("context.strategy")
'context'
>>> _category_from_name("context.pipeline_component.scorer")
'context'
"""
return name.split(".")[0]
# The authoritative list of all 30 spec-defined extension points.
_EXTENSION_POINT_DEFS: tuple[_ExtensionPointDef, ...] = (
# --- Context (12) ---
_ExtensionPointDef(
"context.strategy",
ContextStrategyExtension,
"Pluggable context assembly strategy",
),
_ExtensionPointDef(
"context.pipeline_component.strategy_selector",
ContextPipelineComponentExtension,
"Selects which context strategies to invoke",
),
_ExtensionPointDef(
"context.pipeline_component.budget_allocator",
ContextPipelineComponentExtension,
"Distributes token budget across strategies",
),
_ExtensionPointDef(
"context.pipeline_component.strategy_executor",
ContextPipelineComponentExtension,
"Executes selected strategies in parallel",
),
_ExtensionPointDef(
"context.pipeline_component.deduplicator",
ContextPipelineComponentExtension,
"Removes duplicate context fragments",
),
_ExtensionPointDef(
"context.pipeline_component.depth_resolver",
ContextPipelineComponentExtension,
"Resolves detail depth levels for fragments",
),
_ExtensionPointDef(
"context.pipeline_component.scorer",
ContextPipelineComponentExtension,
"Scores context fragments by relevance",
),
_ExtensionPointDef(
"context.pipeline_component.packer",
ContextPipelineComponentExtension,
"Packs scored fragments within token budget",
),
_ExtensionPointDef(
"context.pipeline_component.orderer",
ContextPipelineComponentExtension,
"Orders packed fragments for presentation",
),
_ExtensionPointDef(
"context.pipeline_component.preamble_generator",
ContextPipelineComponentExtension,
"Generates preamble text for assembled context",
),
_ExtensionPointDef(
"context.pipeline_component.skeleton_compressor",
ContextPipelineComponentExtension,
"Compresses code skeletons to save tokens",
),
_ExtensionPointDef(
"context.storage_backend",
ContextStorageBackendExtension,
"Pluggable storage backend for context data",
),
# --- Output (3) ---
_ExtensionPointDef(
"output.renderer",
OutputRendererExtension,
"Renders assembled output to a specific format",
),
_ExtensionPointDef(
"output.materializer",
OutputMaterializerExtension,
"Writes rendered output to a destination",
),
_ExtensionPointDef(
"output.format",
OutputFormatExtension,
"Custom output serialization format",
),
# --- Validation (2) ---
_ExtensionPointDef(
"validation.runner",
ValidationRunnerExtension,
"Pluggable validation execution engine",
),
_ExtensionPointDef(
"validation.rule_provider",
ValidationRuleProviderExtension,
"Provides validation rules from external sources",
),
# --- Tools (2) ---
_ExtensionPointDef(
"tool.provider",
ToolProviderExtension,
"Registers external tools for agent use",
),
_ExtensionPointDef(
"tool.middleware",
ToolMiddlewareExtension,
"Intercepts tool invocations for cross-cutting concerns",
),
# --- Skills (2) ---
_ExtensionPointDef(
"skill.provider",
SkillProviderExtension,
"Provides skill definitions from external sources",
),
_ExtensionPointDef(
"skill.template",
SkillTemplateExtension,
"Custom skill template renderer",
),
# --- Resources (2) ---
_ExtensionPointDef(
"resource.resolver",
ResourceResolverExtension,
"Resolves resource URIs to content",
),
_ExtensionPointDef(
"resource.type_handler",
ResourceTypeHandlerExtension,
"Handles loading of specific resource types",
),
# --- A2A (2) ---
_ExtensionPointDef(
"a2a.transport",
A2ATransportExtension,
"Agent-to-agent communication transport layer",
),
_ExtensionPointDef(
"a2a.extension_method",
A2AExtensionMethodExtension,
"Custom A2A protocol extension method",
),
# --- Events (2) ---
_ExtensionPointDef(
"event.handler",
EventHandlerExtension,
"Handles domain events from the event bus",
),
_ExtensionPointDef(
"event.filter",
EventFilterExtension,
"Filters events before propagation",
),
# --- Config (2) ---
_ExtensionPointDef(
"config.source",
ConfigSourceExtension,
"External configuration source provider",
),
_ExtensionPointDef(
"config.validator",
ConfigValidatorExtension,
"Validates configuration values against rules",
),
# --- Safety (1) ---
_ExtensionPointDef(
"safety.guardrail",
SafetyGuardrailExtension,
"Safety guardrail for agent action inspection",
),
)
# Public constant: number of spec-defined extension points.
TOTAL_EXTENSION_POINTS: int = 30
def get_extension_point_definitions() -> tuple[ExtensionPoint, ...]:
"""Return all 30 spec-defined extension points as ExtensionPoint models.
Each entry includes the point's name, Protocol type, description,
and a ``registry_key`` equal to its category prefix.
"""
return tuple(
ExtensionPoint(
name=defn.name,
protocol_type=defn.protocol_type,
description=defn.description,
registry_key=_category_from_name(defn.name),
)
for defn in _EXTENSION_POINT_DEFS
)
def get_extension_points_by_category() -> dict[str, list[ExtensionPoint]]:
"""Return extension points grouped by category.
Categories are derived from the first dotted segment of each
extension point name (e.g. ``"context"``, ``"output"``).
"""
grouped: dict[str, list[ExtensionPoint]] = {}
for ep in get_extension_point_definitions():
category = _category_from_name(ep.name)
grouped.setdefault(category, []).append(ep)
return grouped
def register_all_extension_points(manager: ExtensionPointRegistrar) -> int:
"""Register all 30 spec-defined extension points on *manager*.
Args:
manager: Object with a ``register_extension_point`` method
(typically :class:`PluginManager`).
Returns:
The number of extension points registered.
"""
count = 0
for ep in get_extension_point_definitions():
manager.register_extension_point(ep)
count += 1
return count
src/cleveragents/infrastructure/plugins/extension_protocols.py
+337
View File
@@ -0,0 +1,337 @@
"""Typed Protocol interfaces for all spec-defined extension points.
Defines ``@runtime_checkable`` Protocol classes for each of the 30
extension point categories. Each Protocol specifies the minimum
interface that a plugin implementation must satisfy.
Based on ``docs/specification.md`` Extension Points Summary and
issue #939.
ISSUES CLOSED: #939
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any, Protocol, runtime_checkable
# ---------------------------------------------------------------------------
# Context extension protocols (12 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class ContextStrategyExtension(Protocol):
"""Protocol for context strategy plugins."""
@property
def name(self) -> str: ...
def can_handle(self, request: Mapping[str, Any]) -> float: ...
def assemble(self, fragments: Sequence[Any], budget: Any) -> Sequence[Any]: ...
@runtime_checkable
class ContextPipelineComponentExtension(Protocol):
"""Protocol for pipeline stage components.
Covers all 10 pipeline phases: strategy_selector,
budget_allocator, strategy_executor, deduplicator,
depth_resolver, scorer, packer, orderer,
preamble_generator, skeleton_compressor.
"""
@property
def component_name(self) -> str: ...
def process(self, fragments: Sequence[Any], context: Mapping[str, Any]) -> Sequence[Any]: ... # noqa: E501
@runtime_checkable
class ContextStorageBackendExtension(Protocol):
"""Protocol for context storage backends (text, vector, graph)."""
@property
def backend_type(self) -> str: ...
def store(self, key: str, data: Any) -> None: ...
def retrieve(self, key: str) -> Any: ...
# ---------------------------------------------------------------------------
# Output extension protocols (3 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class OutputRendererExtension(Protocol):
"""Protocol for output renderer plugins."""
@property
def format_name(self) -> str: ...
def render(self, data: Any) -> str: ...
@runtime_checkable
class OutputMaterializerExtension(Protocol):
"""Protocol for output materializer plugins."""
@property
def target_type(self) -> str: ...
def materialize(self, content: str, destination: str) -> None: ...
@runtime_checkable
class OutputFormatExtension(Protocol):
"""Protocol for custom output format plugins."""
@property
def format_id(self) -> str: ...
def serialize(self, data: Any) -> str: ...
def deserialize(self, raw: str) -> Any: ...
# ---------------------------------------------------------------------------
# Validation extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class ValidationRunnerExtension(Protocol):
"""Protocol for validation runner plugins."""
@property
def runner_name(self) -> str: ...
def validate(self, target: Any) -> Sequence[Any]: ...
@runtime_checkable
class ValidationRuleProviderExtension(Protocol):
"""Protocol for validation rule provider plugins."""
@property
def provider_name(self) -> str: ...
def get_rules(self) -> Sequence[Any]: ...
# ---------------------------------------------------------------------------
# Tool extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class ToolProviderExtension(Protocol):
"""Protocol for tool provider plugins."""
@property
def provider_name(self) -> str: ...
def list_tools(self) -> Sequence[str]: ...
def invoke(self, tool_name: str, arguments: Mapping[str, Any]) -> Any: ...
@runtime_checkable
class ToolMiddlewareExtension(Protocol):
"""Protocol for tool middleware plugins."""
@property
def middleware_name(self) -> str: ...
def before_invoke(self, tool_name: str, arguments: Mapping[str, Any]) -> Mapping[str, Any]: ... # noqa: E501
def after_invoke(self, tool_name: str, result: Any) -> Any: ...
# ---------------------------------------------------------------------------
# Skill extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class SkillProviderExtension(Protocol):
"""Protocol for skill provider plugins."""
@property
def provider_name(self) -> str: ...
def list_skills(self) -> Sequence[str]: ...
def get_skill(self, skill_name: str) -> Any: ...
@runtime_checkable
class SkillTemplateExtension(Protocol):
"""Protocol for skill template plugins."""
@property
def template_name(self) -> str: ...
def render(self, context: Mapping[str, Any]) -> str: ...
# ---------------------------------------------------------------------------
# Resource extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class ResourceResolverExtension(Protocol):
"""Protocol for resource resolver plugins."""
@property
def resolver_name(self) -> str: ...
def resolve(self, uri: str) -> Any: ...
@runtime_checkable
class ResourceTypeHandlerExtension(Protocol):
"""Protocol for resource type handler plugins."""
@property
def handled_type(self) -> str: ...
def can_handle(self, resource_type: str) -> bool: ...
def load(self, uri: str) -> Any: ...
# ---------------------------------------------------------------------------
# A2A extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class A2ATransportExtension(Protocol):
"""Protocol for A2A transport plugins."""
@property
def transport_name(self) -> str: ...
def send(self, destination: str, payload: Any) -> None: ...
def receive(self) -> Any: ...
@runtime_checkable
class A2AExtensionMethodExtension(Protocol):
"""Protocol for A2A extension method plugins."""
@property
def method_name(self) -> str: ...
def execute(self, params: Mapping[str, Any]) -> Any: ...
# ---------------------------------------------------------------------------
# Event extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class EventHandlerExtension(Protocol):
"""Protocol for event handler plugins."""
@property
def handler_name(self) -> str: ...
def handle(self, event: Any) -> None: ...
@runtime_checkable
class EventFilterExtension(Protocol):
"""Protocol for event filter plugins."""
@property
def filter_name(self) -> str: ...
def should_propagate(self, event: Any) -> bool: ...
# ---------------------------------------------------------------------------
# Config extension protocols (2 extension points)
# ---------------------------------------------------------------------------
@runtime_checkable
class ConfigSourceExtension(Protocol):
"""Protocol for configuration source plugins."""
@property
def source_name(self) -> str: ...
def load(self) -> Mapping[str, Any]: ...
@runtime_checkable
class ConfigValidatorExtension(Protocol):
"""Protocol for configuration validator plugins."""
@property
def validator_name(self) -> str: ...
def validate(self, config: Mapping[str, Any]) -> Sequence[str]: ...
# ---------------------------------------------------------------------------
# Safety extension protocols (1 extension point)
# ---------------------------------------------------------------------------
@runtime_checkable
class SafetyGuardrailExtension(Protocol):
"""Protocol for safety guardrail plugins."""
@property
def guardrail_name(self) -> str: ...
def check(self, action: Any) -> bool: ...
def explain(self) -> str: ...
# ---------------------------------------------------------------------------
# Re-export mapping for programmatic access
# ---------------------------------------------------------------------------
ALL_EXTENSION_PROTOCOLS: dict[str, type[Any]] = {
"context.strategy": ContextStrategyExtension,
"context.pipeline_component.strategy_selector": ContextPipelineComponentExtension,
"context.pipeline_component.budget_allocator": ContextPipelineComponentExtension,
"context.pipeline_component.strategy_executor": ContextPipelineComponentExtension,
"context.pipeline_component.deduplicator": ContextPipelineComponentExtension,
"context.pipeline_component.depth_resolver": ContextPipelineComponentExtension,
"context.pipeline_component.scorer": ContextPipelineComponentExtension,
"context.pipeline_component.packer": ContextPipelineComponentExtension,
"context.pipeline_component.orderer": ContextPipelineComponentExtension,
"context.pipeline_component.preamble_generator": ContextPipelineComponentExtension,
"context.pipeline_component.skeleton_compressor": ContextPipelineComponentExtension,
"context.storage_backend": ContextStorageBackendExtension,
"output.renderer": OutputRendererExtension,
"output.materializer": OutputMaterializerExtension,
"output.format": OutputFormatExtension,
"validation.runner": ValidationRunnerExtension,
"validation.rule_provider": ValidationRuleProviderExtension,
"tool.provider": ToolProviderExtension,
"tool.middleware": ToolMiddlewareExtension,
"skill.provider": SkillProviderExtension,
"skill.template": SkillTemplateExtension,
"resource.resolver": ResourceResolverExtension,
"resource.type_handler": ResourceTypeHandlerExtension,
"a2a.transport": A2ATransportExtension,
"a2a.extension_method": A2AExtensionMethodExtension,
"event.handler": EventHandlerExtension,
"event.filter": EventFilterExtension,
"config.source": ConfigSourceExtension,
"config.validator": ConfigValidatorExtension,
"safety.guardrail": SafetyGuardrailExtension,
}