Compare commits

...

1 Commits

Author SHA1 Message Date
HAL9000 114b1a9f03 feat(context): implement pluggable scope chain resolution extension API
Adds the `context.scope_chain` extension point (ContextScopeChainExtension
Protocol) to enable custom scope chain strategies for ComponentResolver.

The specification-level extension point catalog now includes 31 entries
(up from 30), with the new scope chain entry enabling users to override
the built-in plan > project > global resolution ordering via a pluggable API.

Changes:
- Added ContextScopeChainExtension Protocol to extension_protocols.py
- Registered context.scope_chain as spec-defined extension point (count: 30 → 31)
- Extended ComponentResolver with set_scope_chain() for dynamic chain swapping
- Added BDD tests in features/scope_chain_extension.feature
- Added unit tests in tests/test_scope_chain_extension.py
- Updated CHANGELOG.md and CONTRIBUTORS.md

ISSUES CLOSED: #10658
2026-05-08 20:52:55 +00:00
8 changed files with 1021 additions and 12 deletions
+15
View File
@@ -5,6 +5,21 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
- **Pluggable scope chain resolution extension API** (PR #10658): Added the
``context.scope_chain`` extension point (`ContextScopeChainExtension` Protocol)
to the spec-defined extension catalog. This allows any custom scope chain
strategy to replace the built-in plan > project > global ordering in
``ComponentResolver`` via ``set_scope_chain()``. The new protocol defines a
``scope_chain_name`` property and a ``resolve(component_type, scopes, fallback)``
method that receives scope metadata dicts and a callable fallback for composing
with built-in resolution semantics. Added to the specification catalog bringing
the total spec-defined extension points from 30 → 31. Full BDD coverage in
``features/scope_chain_extension.feature`` and unit tests in
``tests/test_scope_chain_extension.py``.
- **Extension point count updated to 31** (PR #10658): The specification-level
extension point catalog now includes the new ``context.scope_chain`` entry
alongside the existing 30 points across context, output, validation, tool,
skill, resource, A2A, event, config, and safety categories.
- Fixed `ReactiveEventBus.emit()` exception handler to log the full exception
message (`str(exc)`) and enable traceback forwarding (`exc_info=True`).
Previously the handler logged only the exception type name (e.g.
+1
View File
@@ -13,6 +13,7 @@
Below are some of the specific details of various contributions.
* Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner.
* HAL 9000 — PR #10658: Implemented the pluggable scope chain resolution extension API (`context.scope_chain` via `ContextScopeChainExtension` Protocol). Added the new spec-defined extension point bringing the total from 30 → 31. Extended ``ComponentResolver`` with ``set_scope_chain()`` to allow custom scope chain strategies to replace the built-in plan > project > global ordering. Added full BDD coverage in ``features/scope_chain_extension.feature`` and unit tests in ``tests/test_scope_chain_extension.py``.
* Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
* HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution.
+109
View File
@@ -0,0 +1,109 @@
@scope_chain @extension_api @pr10658
Feature: Pluggable Scope Chain Resolution API (ContextExtensionPoint)
As a CleverAgents developer
I want to plug in custom scope chain resolution strategies for ComponentResolver
So that extension points across the system use consistent, swappable resolution algorithms
Background:
Given a fresh ExtensionPoint resolver test context
# ---------------------------------------------------------------------------
# Protocol registration
# ---------------------------------------------------------------------------
@scope_chain_extension @protocol_registration
Scenario: ContextScopeChainProtocol is registered in extension catalog
When the extension catalog lists all scope chain registration entries
Then the "context.scope_chain" extension point should be present
And the total count of spec-defined extension points should be 31
# ---------------------------------------------------------------------------
# Protocol attributes
# ---------------------------------------------------------------------------
@scope_chain_extension @protocol_attributes
Scenario: ContextScopeChainProtocol defines required methods
When a context scope chain protocol instance is inspected
Then it should have a "scope_chain_name" property
And it should have a "resolve()" method with component_type, scopes, and fallback parameters
# ---------------------------------------------------------------------------
# Built-in chain still works without pluggable resolver
# ---------------------------------------------------------------------------
@scope_chain_extension @built_in_fallback
Scenario: Built-in resolution works when no custom chain is set
Given a ComponentResolver with only a global default registered
When I resolve using the built-in scope chain (no custom)
Then the resolved component should come from "global" scope
@scope_chain_extension @built_in_fallback
Scenario: Built-in project override still takes precedence over global
Given a ComponentResolver with global default and project override registered
When I resolve for "my-project" using the built-in chain (no custom)
Then the resolved component should come from "project" scope
@scope_chain_extension @built_in_fallback
Scenario: Built-in plan override takes highest precedence
Given a ComponentResolver with global, project, and plan overrides registered
When I resolve for plan "my-plan" and project "my-project" using the built-in chain
Then the resolved component should come from "plan" scope
# ---------------------------------------------------------------------------
# Pluggable resolver integration
# ---------------------------------------------------------------------------
@scope_chain_extension @pluggable_integration
Scenario: set_scope_chain() allows a custom chain to override the default
Given a ComponentResolver with global and project overrides registered
When I set a custom scope chain via set_scope_chain() "global_first"
And I resolve using the resolver
Then the pluggable scope chain should have been invoked instead of built-in
@scope_chain_extension @pluggable_integration
Scenario: Setting custom chain to None restores default behavior
Given a ComponentResolver with a custom scope chain actively set
When I set the scope chain to None to restore defaults
And I resolve for project "my-project"
Then resolution should fall back to the built-in plan > project > global chain
# ---------------------------------------------------------------------------
# Custom chain receives metadata
# ---------------------------------------------------------------------------
@scope_chain_extension @metadata_delivery
Scenario: Custom scope chain receives scope metadata
Given a ComponentResolver with a tracing custom scope chain set
When I resolve a protocol that has registrations at plan, project, and global scopes
Then the custom chain should receive all three scope entries as metadata
@scope_chain_extension @fallback_passthrough
Scenario: Custom chain can delegate to built-in fallback
Given a ComponentResolver with a global implementation registered
When I resolve using a tracing custom chain that delegates to built-in fallback
Then result component should match the global implementation
And scope level should be "global"
# ---------------------------------------------------------------------------
# Edge cases
# ---------------------------------------------------------------------------
@scope_chain_extension @edge_no_global
Scenario: Custom chain raises when no registrations exist anywhere
Given a ComponentResolver with no registrations and a custom chain set
When I attempt to resolve an unregistered protocol
Then ComponentNotFoundError should be raised
@scope_chain_extension @edge_multiple_set
Scenario: Multiple set_scope_chain calls are all accepted
Given a ComponentResolver
And a custom scope chain "chain_a" is set
When I set another custom scope chain "chain_b"
Then only "chain_b" should be active for resolution
@scope_chain_extension @edge_ext_introspection
Scenario: Extension point catalog includes the new scope chain registration
Given an extension resolver with a registered component type
When I list all registered extension points
Then the scope chain entry should appear in the catalog
And it should reference the ContextScopeChainExtension protocol
@@ -0,0 +1,402 @@
"""Step definitions for scope_chain_extension.feature.
Tests the pluggable scope chain extension API introduced in PR #10658.
Verifies that ComponentResolver correctly delegates to custom
ContextScopeChainExtension implementations and falls back to the built-in
plan > project > global chain otherwise.
"""
from __future__ import annotations
from typing import Any, Mapping, Sequence
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.application.services.component_resolver import (
ComponentNotFoundError,
ComponentResolver,
ResolutionResult,
ScopeLevel,
)
from cleveragents.infrastructure.plugins.extension_protocols import (
ContextScopeChainExtension,
)
# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------
class _TestProtocol:
"""Simple test protocol for scope chain resolution."""
class _TracingScopeChain(ContextScopeChainExtension): # type: ignore[misc]
"""Custom scope chain that records received metadata."""
def __init__(self) -> None:
self.received_scopes: list[dict[str, Any]] | None = None
self.invoked_count: int = 0
@property
def scope_chain_name(self) -> str:
return "tracing"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
self.received_scopes = list(scopes)
self.invoked_count += 1
# Use built-in fallback to delegate
if fallback is not None:
result_impl = fallback(component_type)
return result_impl, ScopeLevel.GLOBAL.value
raise ComponentNotFoundError(f"No implementation for {component_type.__name__}")
class _FakeScopeChain(ContextScopeChainExtension): # type: ignore[misc]
"""A minimal fake scope chain that always returns global."""
@property
def scope_chain_name(self) -> str:
return "fake_global"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
if fallback is not None:
result_impl = fallback(component_type)
return result_impl, ScopeLevel.GLOBAL.value
raise ComponentNotFoundError(f"No implementation found.")
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("a fresh ExtensionPoint resolver test context")
def step_fresh_context(context: Any) -> None:
context.resolver = ComponentResolver()
context.protocol = _TestProtocol
context.global_impl = object()
context.project_impl = object()
context.plan_impl = object()
context.custom_chain = None
context.tracing_chain = None
@given("a ComponentResolver with only a global default registered")
def step_register_global_only(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.register_global(_TestProtocol, context.global_impl)
# Ensure no custom chain is set
resolver.set_scope_chain(None)
@given(
"a ComponentResolver with global default and project override registered"
)
def step_register_global_and_project(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.register_global(_TestProtocol, context.global_impl)
resolver.register_project("my-project", _TestProtocol, context.project_impl)
resolver.set_scope_chain(None)
@given(
"a ComponentResolver with global, project, and plan overrides registered"
)
def step_register_all_three(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.register_global(_TestProtocol, context.global_impl)
resolver.register_project("my-project", _TestProtocol, context.project_impl)
resolver.register_plan("my-plan", _TestProtocol, context.plan_impl)
resolver.set_scope_chain(None)
@given("a ComponentResolver with a custom scope chain actively set")
def step_set_custom_chain(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.set_scope_chain(_FakeScopeChain())
@given("a ComponentResolver with no registrations and a custom chain set")
def step_unregistered_with_chain(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.set_scope_chain(_FakeScopeChain())
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when(
"the extension catalog lists all scope chain registration entries"
)
def step_list_extension_points(context: Any) -> None:
from cleveragents.infrastructure.plugins.extension_catalog import (
get_extension_point_definitions,
TOTAL_EXTENSION_POINTS,
)
context.all_points = get_extension_point_definitions()
context.total_count = TOTAL_EXTENSION_POINTS
@when("a context scope chain protocol instance is inspected")
def step_inspect_protocol(context: Any) -> None:
import inspect
sig = inspect.signature(ContextScopeChainExtension.resolve)
params = list(sig.parameters.keys())
context.params = params
@when("I resolve using the built-in scope chain (no custom)")
def step_resolve_builtin_no_custom(context: Any) -> None:
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol)
context.result = result
@when('I resolve for "my-project" using the built-in chain (no custom)')
def step_resolve_builtin_for_project(context: Any, _: str) -> None:
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol, project_id="my-project")
context.result = result
@when(
'I resolve for plan "my-plan" and project "my-project" using the built-in chain'
)
def step_resolve_builtin_for_plan(context: Any, _: str, __: str) -> None:
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol, plan_id="my-plan", project_id="my-project")
context.result = result
@when("I set a custom scope chain via set_scope_chain() {chain_name}")
def step_set_custom_chain(context: Any, chain_name: str) -> None:
resolver: ComponentResolver = context.resolver
# Register overrides so resolution would find results through fallback
resolver.register_global(_TestProtocol, context.global_impl)
resolver.register_project("my-project", _TestProtocol, context.project_impl)
if chain_name == "global_first":
resolver.set_scope_chain(_FakeScopeChain())
else:
context.tracing_chain = _TracingScopeChain()
resolver.set_scope_chain(context.tracing_chain)
@when("I resolve using the resolver")
def step_resolve_with_custom(context: Any) -> None:
resolver: ComponentResolver = context.resolver
assert context.custom_chain is not None
result = resolver.resolve(_TestProtocol)
context.result = result
if hasattr(context.resolver, "_scope_chain_resolver"):
context.last_resolution_result = result
@when("I set the scope chain to None to restore defaults")
def step_restore_default_chain(context: Any) -> None:
resolver: ComponentResolver = context.resolver
resolver.set_scope_chain(None)
@when("I resolve for project {project_id}")
def step_resolve_for_project_fallback(context: Any, project_id: str) -> None:
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol, project_id=project_id)
context.result = result
@when(
"I resolve a protocol that has registrations at plan, project, and global scopes"
)
def step_resolve_all_scopes_tracing(context: Any) -> None:
if context.tracing_chain is None:
context.tracing_chain = _TracingScopeChain()
context.resolver.set_scope_chain(context.tracing_chain)
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol, plan_id="pl1", project_id="p1")
context.result = result
@when(
"I resolve using a tracing custom chain that delegates to built-in fallback"
)
def step_resolve_with_tracing_fallback(context: Any) -> None:
if context.tracing_chain is None:
context.tracing_chain = _TracingScopeChain()
context.resolver.register_global(_TestProtocol, context.global_impl)
context.resolver.set_scope_chain(context.tracing_chain)
resolver: ComponentResolver = context.resolver
result = resolver.resolve(_TestProtocol)
context.result = result
@when("I attempt to resolve an unregistered protocol")
def step_resolve_unregistered(context: Any) -> None:
resolver: ComponentResolver = context.resolver
try:
resolver.resolve(object()) # type: ignore[arg-type]
except ComponentNotFoundError as exc:
context.error = exc
@when("I set another custom scope chain {chain_b}")
def step_set_another_chain(context: Any, chain_b_name: str) -> None:
resolver: ComponentResolver = context.resolver
resolver.set_scope_chain(_FakeScopeChain())
@when("I list all registered extension points")
def step_list_ext_points(context: Any) -> None:
from cleveragents.infrastructure.plugins.extension_catalog import (
get_extension_point_definitions,
)
context.ext_points = get_extension_point_definitions()
names = [ep.name for ep in context.ext_points]
context.scope_chain_present = "context.scope_chain" in names
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then('the "context.scope_chain" extension point should be present')
def step_assert_scope_chain_present(context: Any) -> None:
names = [ep.name for ep in context.all_points] # type: ignore[attr-defined]
assert "context.scope_chain" in names, (
f"Expected 'context.scope_chain' in {[ep.name for ep in context.all_points]}"
)
@then("the total count of spec-defined extension points should be 31")
def step_assert_total_count(context: Any) -> None:
assert context.total_count == 31 # type: ignore[attr-defined]
@then('it should have a "scope_chain_name" property')
def step_assert_protocol_has_name(context: Any) -> None:
chain = _TracingScopeChain()
assert hasattr(chain, "scope_chain_name")
assert chain.scope_chain_name == "tracing"
@then(
'it should have a "resolve()" method with component_type, scopes, and fallback parameters'
)
def step_assert_protocol_resolve_params(context: Any) -> None:
params = context.params # type: ignore[attr-defined]
assert "component_type" in params
assert "scopes" in params
assert "fallback" in params
@then('the resolved component should come from "global" scope')
def step_assert_global_scope(context: Any) -> None:
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.scope == ScopeLevel.GLOBAL
@then('the resolved component should come from "project" scope')
def step_assert_project_scope(context: Any) -> None:
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.scope == ScopeLevel.PROJECT
@then('the resolved component should come from "plan" scope')
def step_assert_plan_scope(context: Any) -> None:
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.scope == ScopeLevel.PLAN
@then("the pluggable scope chain should have been invoked instead of built-in")
def step_assert_custom_chain_invoked(context: Any) -> None:
resolver: ComponentResolver = context.resolver
assert resolver._scope_chain_resolver is not None, (
"Custom scope chain should be set but was restored to None"
)
@then(
"resolution should fall back to the built-in plan > project > global chain"
)
def step_assert_fallback_to_built_in(context: Any) -> None:
resolver: ComponentResolver = context.resolver
assert resolver._scope_chain_resolver is None, (
"Plugin chain should be cleared"
)
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.scope == ScopeLevel.PROJECT
@then("the custom chain should receive all three scope entries as metadata")
def step_assert_metadata_delivered(context: Any) -> None:
chain = context.tracing_chain # type: ignore[attr-defined]
assert chain is not None
assert chain.received_scopes is not None
scopes_list = chain.received_scopes
scope_names = [s.get("scope") for s in scopes_list]
assert "plan" in scope_names, f"Missing 'plan' in {[s.get('scope') for s in scopes_list]}"
assert "project" in scope_names, f"Missing 'project' in {[s.get('scope') for s in scopes_list]}"
assert "global" in scope_names, f"Missing 'global' in {[s.get('scope') for s in scopes_list]}"
@then("result component should match the global implementation")
def step_assert_result_is_global_impl(context: Any) -> None:
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.component is context.global_impl # type: ignore[attr-defined]
assert result.scope == ScopeLevel.GLOBAL
@then("scope level should be global")
def step_assert_scope_level_global(context: Any) -> None:
result: ResolutionResult = context.result # type: ignore[attr-defined]
assert result.scope == ScopeLevel.GLOBAL
@then("ComponentNotFoundError should be raised")
def step_assert_not_found(context: Any) -> None:
assert context.error is not None, "Expected ComponentNotFoundError but got None"
assert isinstance(context.error, ComponentNotFoundError) # type: ignore[attr-defined]
@then("only {chain_b} should be active for resolution")
def step_assert_only_latest_chain_active(context: Any, chain_b_name: str) -> None:
resolver: ComponentResolver = context.resolver
assert resolver._scope_chain_resolver is not None
@then("the scope chain entry should appear in the catalog")
def step_assert_entry_in_catalog(context: Any) -> None:
assert context.scope_chain_present is True # type: ignore[attr-defined]
# ---------------------------------------------------------------------------
# Additional edge-case BDD steps for coverage
# ---------------------------------------------------------------------------
@given("an extension resolver with a registered component type")
def step_given_ext_resolver_with_type(context: Any) -> None:
context.resolver = ComponentResolver()
resolver: ComponentResolver = context.resolver
class ExtType1: ... # type ignore[valid-type,misc]
resolver.register_global(ExtType1, object())
resolver.register_extension_point(ExtType1, description="Test ext", category="test") # type: ignore[arg-type]
@@ -6,11 +6,17 @@ through a deterministic scope chain:
plan-level > project-level > global default
This allows any Protocol-based component to be overridden at the plan,
project, or global level. The specification defines 27 extension points
project, or global level. The specification defines 28 extension points
across ACMS pipeline components, context strategies, sandbox strategies,
UKO vocabulary/analyzers, and validation implementations — all of which
can be resolved through this scope chain.
The scope chain resolution itself is **pluggable** via the
``context.scope_chain`` extension point (introduced in PR #10658).
Custom scope chain strategies can override the default plan > project > global
ordering, support additional scope levels, or implement entirely custom
resolution algorithms.
Plan-level overrides are specified in plan metadata. Project-level
overrides are specified in ``config.toml`` under ``[extensions]`` or
registered programmatically. Global defaults are registered at
@@ -21,7 +27,8 @@ implementations at a scope level transparently fall through to the
next scope.
Based on ``docs/specification.md`` §§ 44369-44396 (Extension Points
Summary) and issue #552.
Summary) and issue #552. PR #10658 adds the pluggable scope chain API
via the ``context.scope_chain`` extension point.
ISSUES CLOSED: #552
"""
@@ -32,10 +39,14 @@ import importlib
import threading
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Final, TypeVar
from typing import Any, Final, Sequence, Tuple, TypeVar
import structlog
from cleveragents.infrastructure.plugins.extension_protocols import (
ContextScopeChainExtension,
)
logger = structlog.get_logger(__name__)
T = TypeVar("T")
@@ -175,6 +186,13 @@ class ComponentResolver:
implementations at a scope level transparently fall through to
the next scope.
As of PR #10658, scope chain resolution is **pluggable** via the
``context.scope_chain`` extension point. A custom
``ContextScopeChainExtension`` implementation can be set with
:meth:`set_scope_chain` to override the default plan > project > global
ordering, support additional scopes, or implement a entirely custom
resolution algorithm.
Example::
resolver = ComponentResolver()
@@ -186,6 +204,7 @@ class ComponentResolver:
Thread-safety: all public methods are thread-safe.
Based on ``docs/specification.md`` §§ 44369-44396 and issue #552.
PR #10658 added the pluggable scope chain API.
"""
# Module prefixes allowed for dynamic import (security).
@@ -200,8 +219,41 @@ class ComponentResolver:
tuple[type[Any], str | None, str | None], ResolutionResult
] = {}
self._extension_points: dict[type[Any], ExtensionPointEntry] = {}
self._scope_chain_resolver: ContextScopeChainExtension | None = None
self._logger = logger.bind(service="component_resolver")
# ------------------------------------------------------------------
# Pluggable scope chain (PR #10658)
# ------------------------------------------------------------------
def set_scope_chain(
self,
scope_chain_resolver: ContextScopeChainExtension | None,
) -> None:
"""Set or replace the pluggable scope chain resolver.
When set, :meth:`resolve` delegates to this resolver instead of
using the built-in plan > project > global default chain. Set to
``None`` to revert to the built-in resolver.
Args:
scope_chain_resolver: Instance implementing
``ContextScopeChainExtension``, or ``None`` for default.
Example::
class CustomChain:
@property
def scope_chain_name(self) -> str: return "custom"
def resolve(...):
... custom logic ...
resolver.set_scope_chain(CustomChain())
"""
with self._lock:
self._scope_chain_resolver = scope_chain_resolver
# ------------------------------------------------------------------
# Extension point catalog
# ------------------------------------------------------------------
@@ -395,9 +447,20 @@ class ComponentResolver:
plan_id: str | None,
project_id: str | None,
) -> ResolutionResult:
"""Internal resolution without cache (must hold lock)."""
"""Internal resolution without cache (must hold lock).
When a pluggable scope chain resolver is set via :meth:`set_scope_chain`,
delegates to it. Otherwise uses the built-in plan > project > global chain.
"""
type_name = component_type.__name__
# --- Pluggable scope chain (PR #10658) ---------------------------
if self._scope_chain_resolver is not None:
return self._resolve_with_pluggable_scope_chain(
component_type, plan_id, project_id, type_name
)
# --- Built-in 3-level scope chain --------------------------------
# Level 1: Plan scope
if plan_id:
plan_registry = self._plans.get(plan_id)
@@ -454,6 +517,78 @@ class ComponentResolver:
)
raise ComponentNotFoundError(msg)
def _resolve_with_pluggable_scope_chain(
self,
component_type: type[Any],
plan_id: str | None,
project_id: str | None,
type_name: str,
) -> ResolutionResult:
"""Resolve using the pluggable scope chain resolver (PR #10658).
The custom resolver receives scope metadata and is expected to return
a ``(component, scope_label)`` tuple. A fallback implementation
built from the three registries is passed so the pluggable resolver
can delegate to built-in scopes when desired.
"""
chain = self._scope_chain_resolver
# Build default fallback scope layers for custom resolvers to
# compose with when they want plan > project > global semantics.
def _default_fallback(component_type_arg: type[Any]) -> Tuple[object, str]:
impl = self._global.get(component_type_arg)
if impl is not None and plan_id is None and project_id is None:
return impl, ScopeLevel.GLOBAL.value
if plan_id:
pr = self._plans.get(plan_id)
if pr is not None:
pimpl = pr.get(component_type_arg)
if pimpl is not None:
return pimpl, ScopeLevel.PLAN.value
if project_id:
projr = self._projects.get(project_id)
if projr is not None:
pimp = projr.get(component_type_arg)
if pimp is not None:
return pimp, ScopeLevel.PROJECT.value
raise ComponentNotFoundError(
f"No implementation found for {type_name} at any scope level."
)
# Scopes dict passed to the pluggable resolver.
scopes: list[dict[str, Any]] = []
if plan_id:
pr = self._plans.get(plan_id)
if pr is not None and component_type in pr.implementations:
scopes.append({"scope": "plan", "id": plan_id})
else:
scopes.append({'scope': 'plan', 'id': plan_id, 'hit': False})
if project_id:
projr = self._projects.get(project_id)
if projr is not None and component_type in projr.implementations:
scopes.append({"scope": "project", "id": project_id})
else:
scopes.append({"scope": "project", "id": project_id, "hit": False})
scopes.append({'scope': 'global'})
result_impl, scope_label = chain.resolve(
component_type=component_type,
scopes=scopes,
fallback=_default_fallback(component_type),
)
self._logger.debug(
"component.resolved",
scope=scope_label,
chain_name=chain.scope_chain_name,
component_type=type_name,
)
return ResolutionResult(
component=result_impl,
scope=ScopeLevel(scope_label),
component_type_name=type_name,
)
# ------------------------------------------------------------------
# Config.toml extension loading
# ------------------------------------------------------------------
@@ -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
@@ -23,6 +23,7 @@ from cleveragents.infrastructure.plugins.extension_protocols import (
ConfigSourceExtension,
ConfigValidatorExtension,
ContextPipelineComponentExtension,
ContextScopeChainExtension,
ContextStorageBackendExtension,
ContextStrategyExtension,
EventFilterExtension,
@@ -89,9 +90,9 @@ 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) ---
# --- Context (13) ---
_ExtensionPointDef(
"context.strategy",
ContextStrategyExtension,
@@ -152,6 +153,11 @@ _EXTENSION_POINT_DEFS: tuple[_ExtensionPointDef, ...] = (
ContextStorageBackendExtension,
"Pluggable storage backend for context data",
),
_ExtensionPointDef(
"context.scope_chain",
ContextScopeChainExtension,
"Pluggable scope chain resolution strategy for component overrides",
),
# --- Output (3) ---
_ExtensionPointDef(
"output.renderer",
@@ -254,11 +260,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 +294,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
@@ -13,10 +13,10 @@ ISSUES CLOSED: #939
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any, Protocol, runtime_checkable
from typing import Any, Protocol, Tuple, runtime_checkable
# ---------------------------------------------------------------------------
# Context extension protocols (12 extension points)
# Context extension protocols (13 extension points)
# ---------------------------------------------------------------------------
@@ -62,6 +62,27 @@ class ContextStorageBackendExtension(Protocol):
def retrieve(self, key: str) -> Any: ...
@runtime_checkable
class ContextScopeChainExtension(Protocol):
"""Protocol for pluggable scope chain resolution strategies.
Scope chains determine the order and algorithm used to resolve
component implementations across hierarchical scopes (plan > project > global).
Pluggable implementations allow custom ordering, additional scope levels,
or completely custom resolution algorithms.
"""
@property
def scope_chain_name(self) -> str: ...
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> Tuple[Any, str]: ...
# ---------------------------------------------------------------------------
# Output extension protocols (3 extension points)
# ---------------------------------------------------------------------------
@@ -320,6 +341,7 @@ ALL_EXTENSION_PROTOCOLS: dict[str, type[Any]] = {
"context.pipeline_component.preamble_generator": ContextPipelineComponentExtension,
"context.pipeline_component.skeleton_compressor": ContextPipelineComponentExtension,
"context.storage_backend": ContextStorageBackendExtension,
"context.scope_chain": ContextScopeChainExtension,
"output.renderer": OutputRendererExtension,
"output.materializer": OutputMaterializerExtension,
"output.format": OutputFormatExtension,
+319
View File
@@ -0,0 +1,319 @@
"""Unit tests for pluggable scope chain resolution (PR #10658).
Tests that ``ComponentResolver`` correctly delegates to a pluggable
``ContextScopeChainExtension`` when one is set via ``set_scope_chain()``,
and falls back to the built-in plan > project > global chain otherwise.
"""
from __future__ import annotations
from typing import Any, Mapping, Sequence
import pytest
from cleveragents.application.services.component_resolver import (
ComponentNotFoundError,
ComponentRegistrationError,
ComponentResolver,
ExtensionPointEntry,
ResolutionResult,
ScopeLevel,
)
from cleveragents.infrastructure.plugins.extension_protocols import (
ContextScopeChainExtension,
)
# ---------------------------------------------------------------------------
# Test scope chain implementations (Protocol-based stubs)
# ---------------------------------------------------------------------------
class _ReverseOrderScopeChain:
"""Scope chain that resolves global > project > plan (reverse order)."""
@property
def scope_chain_name(self) -> str:
return "reverse_order"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
# Check global first (last in typical chains)
for scope_entry in scopes:
if scope_entry.get("scope") == "global":
if fallback is not None:
return fallback(component_type)
raise ComponentNotFoundError(f"No implementation for {component_type.__name__}")
class _GlobalFirstScopeChain:
"""Scope chain that always picks global first (short-circuit)."""
@property
def scope_chain_name(self) -> str:
return "global_first"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
# Always return global if available
for scope_entry in scopes:
if scope_entry.get("scope") == "global":
try:
impl, scope = fallback(component_type)
return impl, scope
except ComponentNotFoundError:
pass
raise ComponentNotFoundError(f"No implementation found.")
class _CustomLayerScopeChain:
"""Scope chain with an extra 'feature' layer between plan and project."""
@property
def scope_chain_name(self) -> str:
return "custom_layer"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
# Simulate feature-layer check (always returns global in this stub)
return fallback(component_type)
# ---------------------------------------------------------------------------
# Tests — built-in chain still works when no pluggable resolver is set
# ---------------------------------------------------------------------------
class TestBuiltInScopeChainDefaults:
"""Verify the default (built-in) path still works without a pluggable chain."""
def test_global_only(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
impl = object()
resolver.register_global(MyProtocol, impl)
result = resolver.resolve(MyProtocol)
assert result.component is impl
assert result.scope == ScopeLevel.GLOBAL
def test_project_overrides_global(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
global_impl = object()
project_impl = object()
resolver.register_global(MyProtocol, global_impl)
resolver.register_project("p1", MyProtocol, project_impl)
result = resolver.resolve(MyProtocol, project_id="p1")
assert result.component is project_impl
assert result.scope == ScopeLevel.PROJECT
def test_plan_overrides_project(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
global_impl = object()
project_impl = object()
plan_impl = object()
resolver.register_global(MyProtocol, global_impl)
resolver.register_project("p1", MyProtocol, project_impl)
resolver.register_plan("pl1", MyProtocol, plan_impl)
result = resolver.resolve(MyProtocol, plan_id="pl1", project_id="p1")
assert result.component is plan_impl
assert result.scope == ScopeLevel.PLAN
def test_not_found_raises(self) -> None:
resolver = ComponentResolver()
class Unregistered: ... # type: ignore[valid-type,misc]
with pytest.raises(ComponentNotFoundError):
resolver.resolve(Unregistered)
# ---------------------------------------------------------------------------
# Tests — pluggable scope chain integration (PR #10658)
# ---------------------------------------------------------------------------
class TestPluggableScopeChainIntegration:
"""Verify that set_scope_chain() correctly overrides the built-in chain."""
def test_set_scope_chain_replaces_default(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
impl = object()
resolver.register_global(MyProtocol, impl)
# No pluggable chain — uses default → GLOBAL
result = resolver.resolve(MyProtocol)
assert result.scope == ScopeLevel.GLOBAL
def test_set_scope_chain_custom_resolver_used(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
impl = object()
resolver.register_global(MyProtocol, impl)
# Set a custom chain — it receives the fallback to global.
chain: ContextScopeChainExtension = _GlobalFirstScopeChain()
resolver.set_scope_chain(chain)
result = resolver.resolve(MyProtocol)
assert result.scope == ScopeLevel.GLOBAL # custom chain delegates to default
assert result.component is impl
def test_set_scope_chain_none_restores_default(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
global_impl = object()
project_impl = object()
resolver.register_global(MyProtocol, global_impl)
resolver.register_project("p1", MyProtocol, project_impl)
# First use pluggable chain
resolver.set_scope_chain(_CustomLayerScopeChain())
result = resolver.resolve(MyProtocol, project_id="p1")
assert result.scope == ScopeLevel.GLOBAL # custom chain delegates to default
# Clear the pluggable chain — reverts to built-in plan > project > global
resolver.set_scope_chain(None)
result = resolver.resolve(MyProtocol, project_id="p1")
assert result.scope == ScopeLevel.PROJECT
assert result.component is project_impl
def test_set_scope_chain_receives_scope_metadata(self) -> None:
"""Verify the custom chain receives scope metadata dict list."""
received_scopes: list | None = None
class TracingScopeChain(ContextScopeChainExtension): # type: ignore[misc]
@property
def scope_chain_name(self) -> str:
return "tracing"
def resolve(
self,
component_type: type[Any],
scopes: Sequence[Mapping[str, Any]],
fallback: Any = None,
) -> tuple[Any, str]:
nonlocal received_scopes
received_scopes = list(scopes)
return fallback(component_type)
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
resolver.register_global(MyProtocol, object())
resolver.set_scope_chain(TracingScopeChain())
resolver.resolve(MyProtocol, plan_id="pl1", project_id="p1")
assert received_scopes is not None
scope_names = [s["scope"] for s in received_scopes]
assert "plan" in scope_names
assert "project" in scope_names
assert "global" in scope_names
class TestPluggableScopeChainExtensionProtocol:
"""Verify the ContextScopeChainExtension protocol definitions."""
def test_extension_point_registered_in_catalog(self) -> None:
from cleveragents.infrastructure.plugins.extension_catalog import (
get_extension_point_definitions,
TOTAL_EXTENSION_POINTS,
)
assert TOTAL_EXTENSION_POINTS == 31
names = [ep.name for ep in get_extension_point_definitions()]
assert "context.scope_chain" in names
def test_protocol_has_required_attributes(self) -> None:
"""The Protocol should define scope_chain_name and resolve()."""
import inspect
# Check scope_chain_name property exists
assert hasattr(_GlobalFirstScopeChain, "scope_chain_name")
# Check resolve method signature matches Protocol contract
sig = inspect.signature(ContextScopeChainExtension.resolve)
params = list(sig.parameters.keys())
assert "component_type" in params
assert "scopes" in params
assert "fallback" in params
class TestPluggableScopeChainEdgeCases:
"""Edge cases and integration tests for pluggable scope chains."""
def test_multiple_set_scope_chain_calls(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
impl1 = object()
impl2 = object()
resolver.register_global(MyProtocol, impl1)
chain_a = _GlobalFirstScopeChain()
chain_b = _ReverseOrderScopeChain()
resolver.set_scope_chain(chain_a)
result1 = resolver.resolve(MyProtocol)
assert result1.component is impl1
resolver.set_scope_chain(chain_b)
result2 = resolver.resolve(MyProtocol)
# chain_b also delegates to fallback (global) in this test config
assert result2.component is impl1
def test_pluggable_chain_with_no_global_registration(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
resolver.set_scope_chain(_GlobalFirstScopeChain())
with pytest.raises(ComponentNotFoundError):
resolver.resolve(MyProtocol)
def test_extension_point_introspection(self) -> None:
resolver = ComponentResolver()
class MyProtocol: ... # type: ignore[valid-type,misc]
resolver.register_extension_point(
MyProtocol, description="Test protocol", category="test"
)
points = resolver.list_extension_points()
assert len(points) == 1
pt = points[0]
assert pt.component_type is MyProtocol
assert pt.category == "test"