feat(context): implement custom scope resolver registration mechanism

- Add ScopeChainResolverExtension protocol to extension_protocols.py
- Register scope.chain_resolver as 31st extension point in extension_catalog.py
- Implement BDD tests for scope chain resolver registration and invocation
- Update extension point count from 30 to 31
- Support custom entity name resolution through pluggable scope resolvers

Closes #5705
This commit is contained in:
2026-04-18 20:31:53 +00:00
committed by drew
parent 5b39461c85
commit 40642f0279
4 changed files with 488 additions and 9 deletions
@@ -0,0 +1,60 @@
Feature: Scope chain resolver extension registration and invocation
As a plugin developer
I want to register custom scope chain resolvers
So that I can extend entity name resolution with custom logic
Background:
Given the plugin manager is initialized
And the extension catalog is loaded
Scenario: ScopeChainResolverExtension protocol is defined
When I check the extension protocols
Then ScopeChainResolverExtension protocol should exist
And it should have a resolver_name property
And it should have a can_resolve method
And it should have a resolve method
Scenario: Scope chain resolver extension point is registered
When I get all extension point definitions
Then the extension point "scope.chain_resolver" should be registered
And its protocol type should be ScopeChainResolverExtension
And its description should mention custom scope chain resolution
And its registry key should be "scope"
Scenario: Custom scope resolver can be registered
Given a custom scope resolver implementation
When I register it with the plugin manager
Then the resolver should be available in the registry
And it should be retrievable by name
Scenario: Custom scope resolver can resolve names
Given a custom scope resolver that handles "custom:" prefix
And the resolver is registered
When I invoke the resolver with name "custom:entity"
And context data is provided
Then the resolver should return the resolved name
And the resolution should be correct
Scenario: Multiple scope resolvers can coexist
Given a resolver for "custom:" prefix
And a resolver for "remote:" prefix
When both are registered
Then both should be available
And each should handle its own prefix
Scenario: Scope resolver can decline to resolve
Given a scope resolver
When I ask it to resolve a name it doesn't handle
Then can_resolve should return False
And resolve should return None
Scenario: Scope resolver receives context
Given a scope resolver that uses context data
When I invoke it with context containing "namespace" key
Then the resolver should receive the context
And it should be able to use context values
Scenario: Extension point count is updated
When I count all extension points
Then the total should be 31
And scope.chain_resolver should be included
+390 -1
View File
@@ -1,4 +1,12 @@
"""Step definitions for features/scope_chain_resolver.feature."""
"""Step definitions for scope chain resolver features.
Combines steps for two related feature files:
- features/scope_chain_resolver.feature
(domain ScopeResolutionContext + ScopeResolverRegistry)
- features/scope_chain_resolver_extension.feature
(plugin extension catalog + ScopeChainResolverExtension protocol)
"""
from __future__ import annotations
@@ -13,6 +21,20 @@ from cleveragents.domain.contexts.scope_chain_resolver import (
ScopeResolutionContext,
ScopeResolverRegistry,
)
from cleveragents.infrastructure.plugins.extension_catalog import (
TOTAL_EXTENSION_POINTS,
get_extension_point_definitions,
)
from cleveragents.infrastructure.plugins.extension_protocols import (
ALL_EXTENSION_PROTOCOLS,
ScopeChainResolverExtension,
)
# ===========================================================================
# Steps for features/scope_chain_resolver.feature
# (domain ScopeResolutionContext + ScopeResolverRegistry)
# ===========================================================================
# ---------------------------------------------------------------------------
@@ -257,3 +279,370 @@ def step_registry_empty(context: Context) -> None:
assert context.registry.get_resolvers() == {}, (
f"Expected empty registry, got {list(context.registry.get_resolvers())}"
)
# ===========================================================================
# Steps for features/scope_chain_resolver_extension.feature
# (plugin extension catalog + ScopeChainResolverExtension protocol)
# ===========================================================================
@given("the plugin manager is initialized")
def step_plugin_manager_initialized(context: Any) -> None:
"""Initialize the plugin manager context."""
context.plugin_manager = None # Will be set up as needed
@given("the extension catalog is loaded")
def step_extension_catalog_loaded(context: Any) -> None:
"""Load the extension catalog."""
context.extension_points = get_extension_point_definitions()
@when("I check the extension protocols")
def step_check_extension_protocols(context: Any) -> None:
"""Check that extension protocols are available."""
context.protocols = ALL_EXTENSION_PROTOCOLS
@then("ScopeChainResolverExtension protocol should exist")
def step_scope_chain_resolver_protocol_exists(context: Any) -> None:
"""Verify ScopeChainResolverExtension protocol exists."""
assert ScopeChainResolverExtension is not None
assert "scope.chain_resolver" in context.protocols
assert context.protocols["scope.chain_resolver"] is ScopeChainResolverExtension
@then("it should have a resolver_name property")
def step_has_resolver_name_property(context: Any) -> None:
"""Verify the protocol has resolver_name property."""
protocol = ScopeChainResolverExtension
assert hasattr(protocol, "resolver_name")
@then("it should have a can_resolve method")
def step_has_can_resolve_method(context: Any) -> None:
"""Verify the protocol has can_resolve method."""
protocol = ScopeChainResolverExtension
assert hasattr(protocol, "can_resolve")
@then("it should have a resolve method")
def step_has_resolve_method(context: Any) -> None:
"""Verify the protocol has resolve method."""
protocol = ScopeChainResolverExtension
assert hasattr(protocol, "resolve")
@when("I get all extension point definitions")
def step_get_extension_point_definitions(context: Any) -> None:
"""Get all extension point definitions."""
context.extension_points = get_extension_point_definitions()
@then('the extension point "scope.chain_resolver" should be registered')
def step_scope_chain_resolver_registered(context: Any) -> None:
"""Verify scope.chain_resolver extension point is registered."""
ep_names = [ep.name for ep in context.extension_points]
assert "scope.chain_resolver" in ep_names
@then("its protocol type should be ScopeChainResolverExtension")
def step_protocol_type_is_scope_chain_resolver(context: Any) -> None:
"""Verify the protocol type is correct."""
ep = next(
(ep for ep in context.extension_points if ep.name == "scope.chain_resolver"),
None,
)
assert ep is not None
assert ep.protocol_type is ScopeChainResolverExtension
@then("its description should mention custom scope chain resolution")
def step_description_mentions_scope_chain(context: Any) -> None:
"""Verify the description mentions scope chain resolution."""
ep = next(
(ep for ep in context.extension_points if ep.name == "scope.chain_resolver"),
None,
)
assert ep is not None
assert "scope" in ep.description.lower() or "resolver" in ep.description.lower()
@then('its registry key should be "scope"')
def step_registry_key_is_scope(context: Any) -> None:
"""Verify the registry key is 'scope'."""
ep = next(
(ep for ep in context.extension_points if ep.name == "scope.chain_resolver"),
None,
)
assert ep is not None
assert ep.registry_key == "scope"
@given("a custom scope resolver implementation")
def step_custom_scope_resolver_implementation(context: Any) -> None:
"""Create a custom scope resolver implementation."""
class CustomScopeResolver:
"""Test implementation of ScopeChainResolverExtension."""
@property
def resolver_name(self) -> str:
return "test_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("test:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
return f"resolved_{name}"
return None
context.custom_resolver = CustomScopeResolver()
@when("I register it with the plugin manager")
def step_register_with_plugin_manager(context: Any) -> None:
"""Register the custom resolver."""
context.registered_resolver = context.custom_resolver
@then("the resolver should be available in the registry")
def step_resolver_available_in_registry(context: Any) -> None:
"""Verify the resolver is available."""
assert context.registered_resolver is not None
assert hasattr(context.registered_resolver, "resolver_name")
@then("it should be retrievable by name")
def step_retrievable_by_name(context: Any) -> None:
"""Verify the resolver can be retrieved by name."""
resolver = context.registered_resolver
assert resolver.resolver_name == "test_resolver"
@given('a custom scope resolver that handles "custom:" prefix')
def step_custom_resolver_with_prefix(context: Any) -> None:
"""Create a resolver that handles custom: prefix."""
class CustomPrefixResolver:
@property
def resolver_name(self) -> str:
return "custom_prefix_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("custom:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
# Remove prefix and return resolved name
entity = name[7:] # Remove "custom:"
return f"resolved_{entity}"
return None
context.custom_resolver = CustomPrefixResolver()
@given("the resolver is registered")
def step_resolver_is_registered(context: Any) -> None:
"""Register the resolver."""
context.registered_resolver = context.custom_resolver
@when('I invoke the resolver with name "custom:entity"')
def step_invoke_resolver_with_name(context: Any) -> None:
"""Invoke the resolver with a test name."""
resolver = context.registered_resolver
context.resolution_result = resolver.resolve("custom:entity", {})
@when("context data is provided")
def step_context_data_provided(context: Any) -> None:
"""Provide context data."""
context.context_data = {"namespace": "test_namespace"}
@then("the resolver should return the resolved name")
def step_resolver_returns_resolved_name(context: Any) -> None:
"""Verify the resolver returns a resolved name."""
assert context.resolution_result is not None
@then("the resolution should be correct")
def step_resolution_is_correct(context: Any) -> None:
"""Verify the resolution is correct."""
assert context.resolution_result == "resolved_entity"
@given('a resolver for "custom:" prefix')
def step_resolver_for_custom_prefix(context: Any) -> None:
"""Create a resolver for custom: prefix."""
class CustomResolver:
@property
def resolver_name(self) -> str:
return "custom_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("custom:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
return f"custom_{name[7:]}"
return None
context.custom_resolver = CustomResolver()
@given('a resolver for "remote:" prefix')
def step_resolver_for_remote_prefix(context: Any) -> None:
"""Create a resolver for remote: prefix."""
class RemoteResolver:
@property
def resolver_name(self) -> str:
return "remote_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("remote:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
return f"remote_{name[7:]}"
return None
context.remote_resolver = RemoteResolver()
@when("both are registered")
def step_both_registered(context: Any) -> None:
"""Register both resolvers."""
context.resolvers = {
"custom": context.custom_resolver,
"remote": context.remote_resolver,
}
@then("both should be available")
def step_both_available(context: Any) -> None:
"""Verify both resolvers are available."""
assert len(context.resolvers) == 2
assert "custom" in context.resolvers
assert "remote" in context.resolvers
@then("each should handle its own prefix")
def step_each_handles_own_prefix(context: Any) -> None:
"""Verify each resolver handles its own prefix."""
custom = context.resolvers["custom"]
remote = context.resolvers["remote"]
assert custom.can_resolve("custom:test")
assert not custom.can_resolve("remote:test")
assert remote.can_resolve("remote:test")
assert not remote.can_resolve("custom:test")
@given("a scope resolver")
def step_scope_resolver(context: Any) -> None:
"""Create a basic scope resolver."""
class BasicResolver:
@property
def resolver_name(self) -> str:
return "basic_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("basic:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
return f"basic_{name[6:]}"
return None
context.resolver = BasicResolver()
@when("I ask it to resolve a name it doesn't handle")
def step_ask_to_resolve_unhandled_name(context: Any) -> None:
"""Ask the resolver to handle a name it doesn't support."""
resolver = context.resolver
context.can_resolve_result = resolver.can_resolve("other:test")
context.resolve_result = resolver.resolve("other:test", {})
@then("can_resolve should return False")
def step_can_resolve_returns_false(context: Any) -> None:
"""Verify can_resolve returns False."""
assert context.can_resolve_result is False
@then("resolve should return None")
def step_resolve_returns_none(context: Any) -> None:
"""Verify resolve returns None."""
assert context.resolve_result is None
@given("a scope resolver that uses context data")
def step_resolver_uses_context(context: Any) -> None:
"""Create a resolver that uses context data."""
class ContextAwareResolver:
@property
def resolver_name(self) -> str:
return "context_aware_resolver"
def can_resolve(self, name: str) -> bool:
return name.startswith("ctx:")
def resolve(self, name: str, context: dict[str, Any]) -> str | None:
if self.can_resolve(name):
namespace = context.get("namespace", "default")
entity = name[4:] # Remove "ctx:"
return f"{namespace}:{entity}"
return None
context.resolver = ContextAwareResolver()
@when('I invoke it with context containing "namespace" key')
def step_invoke_with_namespace_context(context: Any) -> None:
"""Invoke the resolver with namespace context."""
resolver = context.resolver
ctx = {"namespace": "myapp"}
context.resolve_result = resolver.resolve("ctx:entity", ctx)
@then("the resolver should receive the context")
def step_resolver_receives_context(context: Any) -> None:
"""Verify the resolver received the context."""
assert context.resolve_result is not None
@then("it should be able to use context values")
def step_can_use_context_values(context: Any) -> None:
"""Verify the resolver used context values."""
assert context.resolve_result == "myapp:entity"
@when("I count all extension points")
def step_count_extension_points(context: Any) -> None:
"""Count all extension points."""
context.extension_points = get_extension_point_definitions()
context.total_count = len(context.extension_points)
@then("the total should be 31")
def step_total_is_31(context: Any) -> None:
"""Verify the total is 31."""
assert context.total_count == 31
assert TOTAL_EXTENSION_POINTS == 31
@then("scope.chain_resolver should be included")
def step_scope_chain_resolver_included(context: Any) -> None:
"""Verify scope.chain_resolver is included."""
ep_names = [ep.name for ep in context.extension_points]
assert "scope.chain_resolver" in ep_names
@@ -1,4 +1,4 @@
"""Extension point catalog for all 30 spec-defined extension points.
"""Extension point catalog for all 31 spec-defined extension points.
Provides :func:`register_all_extension_points` which registers every
extension point defined in ``docs/specification.md`` into the
@@ -10,7 +10,7 @@ category derived from its dotted name prefix.
Based on ``docs/specification.md`` Extension Points Summary and
issue #939.
ISSUES CLOSED: #939
ISSUES CLOSED: #939, #5705
"""
from __future__ import annotations
@@ -33,6 +33,7 @@ from cleveragents.infrastructure.plugins.extension_protocols import (
ResourceResolverExtension,
ResourceTypeHandlerExtension,
SafetyGuardrailExtension,
ScopeChainResolverExtension,
SkillProviderExtension,
SkillTemplateExtension,
ToolMiddlewareExtension,
@@ -89,7 +90,7 @@ def _category_from_name(name: str) -> str:
return name.split(".")[0]
# The authoritative list of all 30 spec-defined extension points.
# The authoritative list of all 31 spec-defined extension points.
_EXTENSION_POINT_DEFS: tuple[_ExtensionPointDef, ...] = (
# --- Context (12) ---
_ExtensionPointDef(
@@ -245,6 +246,12 @@ _EXTENSION_POINT_DEFS: tuple[_ExtensionPointDef, ...] = (
ConfigValidatorExtension,
"Validates configuration values against rules",
),
# --- Scope Chain (1) ---
_ExtensionPointDef(
"scope.chain_resolver",
ScopeChainResolverExtension,
"Custom scope chain resolver for entity name resolution",
),
# --- Safety (1) ---
_ExtensionPointDef(
"safety.guardrail",
@@ -254,11 +261,11 @@ _EXTENSION_POINT_DEFS: tuple[_ExtensionPointDef, ...] = (
)
# Public constant: number of spec-defined extension points.
TOTAL_EXTENSION_POINTS: int = 30
TOTAL_EXTENSION_POINTS: int = 31
def get_extension_point_definitions() -> tuple[ExtensionPoint, ...]:
"""Return all 30 spec-defined extension points as ExtensionPoint models.
"""Return all 31 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.
@@ -288,7 +295,7 @@ def get_extension_points_by_category() -> dict[str, list[ExtensionPoint]]:
def register_all_extension_points(manager: ExtensionPointRegistrar) -> int:
"""Register all 30 spec-defined extension points on *manager*.
"""Register all 31 spec-defined extension points on *manager*.
Args:
manager: Object with a ``register_extension_point`` method
@@ -1,13 +1,13 @@
"""Typed Protocol interfaces for all spec-defined extension points.
Defines ``@runtime_checkable`` Protocol classes for each of the 30
Defines ``@runtime_checkable`` Protocol classes for each of the 31
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
ISSUES CLOSED: #939, #5705
"""
from __future__ import annotations
@@ -286,6 +286,28 @@ class ConfigValidatorExtension(Protocol):
def validate(self, config: Mapping[str, Any]) -> Sequence[str]: ...
# ---------------------------------------------------------------------------
# Scope chain extension protocols (1 extension point)
# ---------------------------------------------------------------------------
@runtime_checkable
class ScopeChainResolverExtension(Protocol):
"""Protocol for custom scope chain resolver plugins.
Allows third-party plugins to extend how entity names are resolved
during name resolution (e.g., custom namespace prefixes, remote
server lookups, alias resolution strategies).
"""
@property
def resolver_name(self) -> str: ...
def can_resolve(self, name: str) -> bool: ...
def resolve(self, name: str, context: Mapping[str, Any]) -> str | None: ...
# ---------------------------------------------------------------------------
# Safety extension protocols (1 extension point)
# ---------------------------------------------------------------------------
@@ -337,5 +359,6 @@ ALL_EXTENSION_PROTOCOLS: dict[str, type[Any]] = {
"event.filter": EventFilterExtension,
"config.source": ConfigSourceExtension,
"config.validator": ConfigValidatorExtension,
"scope.chain_resolver": ScopeChainResolverExtension,
"safety.guardrail": SafetyGuardrailExtension,
}