feat(extensibility): implement pluggable scope chain resolution

Introduce ComponentResolver with a deterministic 3-level scope chain
(plan > project > global) for resolving pluggable Protocol-based
components. The resolver supports caching, thread-safety, config.toml
extension loading, plan metadata loading, introspection APIs, and
security-restricted dynamic imports.

Includes 39 BDD scenarios, 9 Robot Framework integration tests,
ASV benchmarks, and vulture whitelist updates. 100% code coverage
on component_resolver.py; overall coverage at 97.07%.

ISSUES CLOSED: #552

benchmarks/component_resolver_bench.py

@@ -0,0 +1,150 @@
+"""ASV benchmarks for ComponentResolver scope chain resolution.
+
+Measures the performance of:
+- Global default resolution
+- Project override resolution
+- Plan override resolution (3-level chain)
+- Cache hit performance
+- Registration throughput
+- Introspection API latency
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+from typing import Any, Protocol, runtime_checkable
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.application.services.component_resolver import (  # noqa: E402
+    ComponentResolver,
+)
+
+
+# ---------------------------------------------------------------------------
+# Test protocol and implementations
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class _BenchProtocol(Protocol):
+    def run(self) -> str: ...
+
+
+class _BenchGlobal:
+    def run(self) -> str:
+        return "global"
+
+
+class _BenchProject:
+    def run(self) -> str:
+        return "project"
+
+
+class _BenchPlan:
+    def run(self) -> str:
+        return "plan"
+
+
+# ---------------------------------------------------------------------------
+# Timing suite
+# ---------------------------------------------------------------------------
+
+
+class ComponentResolverTimeSuite:
+    """Benchmark timing for component resolution across scope levels."""
+
+    def setup(self) -> None:
+        self._resolver = ComponentResolver()
+        self._resolver.register_global(_BenchProtocol, _BenchGlobal())
+        self._resolver.register_project("proj-1", _BenchProtocol, _BenchProject())
+        self._resolver.register_plan("plan-1", _BenchProtocol, _BenchPlan())
+
+    def time_resolve_global(self) -> None:
+        """Resolve from global scope (no plan/project context)."""
+        self._resolver.invalidate_cache()
+        self._resolver.resolve(_BenchProtocol)
+
+    def time_resolve_project(self) -> None:
+        """Resolve from project scope."""
+        self._resolver.invalidate_cache()
+        self._resolver.resolve(_BenchProtocol, project_id="proj-1")
+
+    def time_resolve_plan(self) -> None:
+        """Resolve from plan scope (full 3-level chain)."""
+        self._resolver.invalidate_cache()
+        self._resolver.resolve(_BenchProtocol, plan_id="plan-1", project_id="proj-1")
+
+    def time_resolve_cache_hit(self) -> None:
+        """Resolve with cache hit (pre-warmed)."""
+        # Prime cache
+        self._resolver.resolve(_BenchProtocol)
+        # Measure cache hit
+        self._resolver.resolve(_BenchProtocol)
+
+    def time_resolve_fallthrough(self) -> None:
+        """Resolve with fallthrough (unknown plan/project)."""
+        self._resolver.invalidate_cache()
+        self._resolver.resolve(_BenchProtocol, plan_id="unknown", project_id="unknown")
+
+    def time_register_global(self) -> None:
+        """Register a global implementation."""
+        resolver = ComponentResolver()
+        resolver.register_global(_BenchProtocol, _BenchGlobal())
+
+    def time_register_project(self) -> None:
+        """Register a project implementation."""
+        resolver = ComponentResolver()
+        resolver.register_project("p", _BenchProtocol, _BenchProject())
+
+    def time_register_plan(self) -> None:
+        """Register a plan implementation."""
+        resolver = ComponentResolver()
+        resolver.register_plan("p", _BenchProtocol, _BenchPlan())
+
+    def time_has_global(self) -> None:
+        """Check has_global."""
+        self._resolver.has_global(_BenchProtocol)
+
+    def time_list_global_types(self) -> None:
+        """List global type names."""
+        self._resolver.list_global_types()
+
+
+# ---------------------------------------------------------------------------
+# Memory suite
+# ---------------------------------------------------------------------------
+
+
+class ComponentResolverMemSuite:
+    """Benchmark memory consumption for component resolver operations."""
+
+    def setup(self) -> None:
+        self._resolver = ComponentResolver()
+        self._resolver.register_global(_BenchProtocol, _BenchGlobal())
+        for i in range(100):
+            self._resolver.register_project(
+                f"proj-{i}", _BenchProtocol, _BenchProject()
+            )
+            self._resolver.register_plan(f"plan-{i}", _BenchProtocol, _BenchPlan())
+
+    def mem_resolve_many(self) -> list[Any]:
+        """Memory for resolving 100 plan-level lookups."""
+        results = []
+        for i in range(100):
+            results.append(
+                self._resolver.resolve(
+                    _BenchProtocol, plan_id=f"plan-{i}", project_id=f"proj-{i}"
+                )
+            )
+        return results

features/component_resolver.feature

@@ -0,0 +1,348 @@
+@extensibility @component_resolver
+Feature: Pluggable Component Resolution with Scope Chain
+  As a CleverAgents developer
+  I want a ComponentResolver that resolves pluggable components through a
+  3-level scope chain (plan > project > global)
+  So that extension points can be consistently overridden at different scopes
+
+  # ---------------------------------------------------------------------------
+  # Global default resolution
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @global
+  Scenario: Global default resolution when no overrides exist
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol with no plan or project context
+    Then the resolved component should be the global default
+    And the resolution scope should be "global"
+
+  @scope_chain @global
+  Scenario: Global default registered at startup is available
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    Then has_global should return True for the test protocol
+
+  @scope_chain @global
+  Scenario: Resolving unregistered component raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to resolve an unregistered protocol
+    Then a ComponentNotFoundError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Project-level override
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @project
+  Scenario: Project override takes precedence over global
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    And a project-level override for project "my-project" for the test protocol
+    When I resolve the test protocol for project "my-project"
+    Then the resolved component should be the project override
+    And the resolution scope should be "project"
+
+  @scope_chain @project
+  Scenario: Project without override falls through to global
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol for project "other-project"
+    Then the resolved component should be the global default
+    And the resolution scope should be "global"
+
+  @scope_chain @project
+  Scenario: Project override registered via register_project
+    Given a fresh ComponentResolver instance
+    And a project-level override for project "my-project" for the test protocol
+    Then has_project should return True for "my-project" and the test protocol
+
+  @scope_chain @project
+  Scenario: Empty project_id raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to register a project override with empty project_id
+    Then a ComponentRegistrationError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Plan-level override (highest priority)
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @plan
+  Scenario: Plan override takes highest precedence
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    And a project-level override for project "my-project" for the test protocol
+    And a plan-level override for plan "plan-001" for the test protocol
+    When I resolve the test protocol for plan "plan-001" and project "my-project"
+    Then the resolved component should be the plan override
+    And the resolution scope should be "plan"
+
+  @scope_chain @plan
+  Scenario: Plan without override falls through to project then global
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    And a project-level override for project "my-project" for the test protocol
+    When I resolve the test protocol for plan "plan-002" and project "my-project"
+    Then the resolved component should be the project override
+    And the resolution scope should be "project"
+
+  @scope_chain @plan
+  Scenario: Empty plan_id raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to register a plan override with empty plan_id
+    Then a ComponentRegistrationError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Fallthrough on missing scope
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @fallthrough
+  Scenario: Missing plan and project falls through to global
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol for plan "unknown-plan" and project "unknown-project"
+    Then the resolved component should be the global default
+    And the resolution scope should be "global"
+
+  @scope_chain @fallthrough
+  Scenario: Only plan registered without global still resolves from plan
+    Given a fresh ComponentResolver instance
+    And a plan-level override for plan "plan-only" for the test protocol
+    When I resolve the test protocol for plan "plan-only"
+    Then the resolved component should be the plan override
+    And the resolution scope should be "plan"
+
+  # ---------------------------------------------------------------------------
+  # Caching
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @caching
+  Scenario: Resolution is cached per scope
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol twice with no context
+    Then the cache size should be 1
+    And both resolutions should return the same result
+
+  @scope_chain @caching
+  Scenario: Cache is invalidated on new registration
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol with no plan or project context
+    And I register a project-level override for project "p1" for the test protocol
+    Then the cache size should be 0
+
+  # ---------------------------------------------------------------------------
+  # Config.toml extension loading
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @config
+  Scenario: Load project extensions from config mapping
+    Given a fresh ComponentResolver instance
+    And an extensions config mapping with a known module path
+    When I load project extensions for project "ext-project"
+    Then the extension should be registered at the project scope
+
+  # ---------------------------------------------------------------------------
+  # Plan metadata extension loading
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @config
+  Scenario: Load plan extensions from metadata mapping
+    Given a fresh ComponentResolver instance
+    And a plan metadata mapping with a known module path
+    When I load plan extensions for plan "ext-plan"
+    Then the extension should be registered at the plan scope
+
+  # ---------------------------------------------------------------------------
+  # Removal
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @removal
+  Scenario: Remove global implementation
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I remove the global implementation for the test protocol
+    Then has_global should return False for the test protocol
+
+  @scope_chain @removal
+  Scenario: Remove project implementation
+    Given a fresh ComponentResolver instance
+    And a project-level override for project "rm-project" for the test protocol
+    When I remove the project implementation for "rm-project" and the test protocol
+    Then has_project should return False for "rm-project" and the test protocol
+
+  @scope_chain @removal
+  Scenario: Remove plan implementation
+    Given a fresh ComponentResolver instance
+    And a plan-level override for plan "rm-plan" for the test protocol
+    When I remove the plan implementation for "rm-plan" and the test protocol
+    Then has_plan should return False for "rm-plan" and the test protocol
+
+  # ---------------------------------------------------------------------------
+  # Introspection
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @introspection
+  Scenario: List global types
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    Then list_global_types should include the test protocol name
+
+  @scope_chain @introspection
+  Scenario: List project types
+    Given a fresh ComponentResolver instance
+    And a project-level override for project "intro-project" for the test protocol
+    Then list_project_types for "intro-project" should include the test protocol name
+
+  @scope_chain @introspection
+  Scenario: List plan types
+    Given a fresh ComponentResolver instance
+    And a plan-level override for plan "intro-plan" for the test protocol
+    Then list_plan_types for "intro-plan" should include the test protocol name
+
+  # ---------------------------------------------------------------------------
+  # Clear
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @clear
+  Scenario: Clear removes all registrations
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    And a project-level override for project "c-project" for the test protocol
+    And a plan-level override for plan "c-plan" for the test protocol
+    When I clear the resolver
+    Then has_global should return False for the test protocol
+    And has_project should return False for "c-project" and the test protocol
+    And has_plan should return False for "c-plan" and the test protocol
+
+  # ---------------------------------------------------------------------------
+  # Extension point catalog
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @catalog
+  Scenario: Register and list extension points
+    Given a fresh ComponentResolver instance
+    When I register an extension point for the test protocol
+    Then list_extension_points should include the test protocol
+
+  # ---------------------------------------------------------------------------
+  # None implementation rejected
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @validation
+  Scenario: Registering None as global raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to register None as a global implementation
+    Then a ComponentRegistrationError should be raised
+
+  @scope_chain @validation
+  Scenario: Registering None as project implementation raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to register None as a project implementation
+    Then a ComponentRegistrationError should be raised
+
+  @scope_chain @validation
+  Scenario: Registering None as plan implementation raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to register None as a plan implementation
+    Then a ComponentRegistrationError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Security: Module prefix restriction
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @security
+  Scenario: Import from disallowed module prefix raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to import a component from "evil_module:Hacker"
+    Then a ComponentRegistrationError should be raised
+
+  @scope_chain @security
+  Scenario: Import with invalid format raises error
+    Given a fresh ComponentResolver instance
+    When I attempt to import a component from "no_colon_here"
+    Then a ComponentRegistrationError should be raised
+
+  # ---------------------------------------------------------------------------
+  # Edge cases: unknown extension names and failed imports
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @config @edge
+  Scenario: Load project extensions with unknown type name skips gracefully
+    Given a fresh ComponentResolver instance
+    And an extensions config with an unknown extension type name
+    When I load project extensions with unknown type for project "edge-proj"
+    Then the loaded extensions list should be empty
+
+  @scope_chain @config @edge
+  Scenario: Load project extensions with bad module path logs warning
+    Given a fresh ComponentResolver instance
+    And an extensions config with a bad module path
+    When I load project extensions with bad path for project "bad-proj"
+    Then the loaded extensions list should be empty
+
+  @scope_chain @config @edge
+  Scenario: Load plan extensions with unknown type name skips gracefully
+    Given a fresh ComponentResolver instance
+    And a plan metadata with an unknown extension type name
+    When I load plan extensions with unknown type for plan "edge-plan"
+    Then the loaded extensions list should be empty
+
+  @scope_chain @config @edge
+  Scenario: Load plan extensions with bad module path logs warning
+    Given a fresh ComponentResolver instance
+    And a plan metadata with a bad module path
+    When I load plan extensions with bad path for plan "bad-plan"
+    Then the loaded extensions list should be empty
+
+  # ---------------------------------------------------------------------------
+  # Edge cases: remove from non-existent scope
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @removal @edge
+  Scenario: Remove project from non-existent project returns False
+    Given a fresh ComponentResolver instance
+    When I remove the project implementation for "nonexistent" and the test protocol
+    Then the component resolver removal result should be False
+
+  @scope_chain @removal @edge
+  Scenario: Remove plan from non-existent plan returns False
+    Given a fresh ComponentResolver instance
+    When I remove the plan implementation for "nonexistent" and the test protocol
+    Then the component resolver removal result should be False
+
+  # ---------------------------------------------------------------------------
+  # Edge cases: list types for non-existent scopes
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @introspection @edge
+  Scenario: List project types for non-existent project returns empty
+    Given a fresh ComponentResolver instance
+    Then list_project_types for "nonexistent" should be empty
+
+  @scope_chain @introspection @edge
+  Scenario: List plan types for non-existent plan returns empty
+    Given a fresh ComponentResolver instance
+    Then list_plan_types for "nonexistent" should be empty
+
+  # ---------------------------------------------------------------------------
+  # Edge case: _ScopeRegistry.remove returns False for missing type
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @removal @edge
+  Scenario: Remove non-existent type from global returns False
+    Given a fresh ComponentResolver instance
+    When I remove the global implementation for the test protocol
+    Then the component resolver removal result should be False
+
+  # ---------------------------------------------------------------------------
+  # Edge case: invalidate_cache explicitly
+  # ---------------------------------------------------------------------------
+
+  @scope_chain @caching @edge
+  Scenario: Explicit invalidate_cache clears all cached entries
+    Given a fresh ComponentResolver instance
+    And a global default implementation for a test protocol
+    When I resolve the test protocol with no plan or project context
+    And I explicitly invalidate the cache
+    Then the cache size should be 0

features/steps/component_resolver_steps.py

@@ -0,0 +1,516 @@
+"""Step definitions for component_resolver.feature.
+
+Tests the ComponentResolver's 3-level scope chain resolution:
+plan > project > global.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, runtime_checkable
+
+from behave import given, then, when  # type: ignore[import-untyped]
+
+from cleveragents.application.services.component_resolver import (
+    ComponentNotFoundError,
+    ComponentRegistrationError,
+    ComponentResolver,
+    ScopeLevel,
+)
+
+# ---------------------------------------------------------------------------
+# Test Protocol and implementations
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class _TestProtocol(Protocol):
+    """A simple test protocol for scope chain testing."""
+
+    def execute(self) -> str: ...
+
+
+class _GlobalDefault:
+    """Global default implementation."""
+
+    def execute(self) -> str:
+        return "global"
+
+
+class _ProjectOverride:
+    """Project-level override implementation."""
+
+    def execute(self) -> str:
+        return "project"
+
+
+class _PlanOverride:
+    """Plan-level override implementation."""
+
+    def execute(self) -> str:
+        return "plan"
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh ComponentResolver instance")
+def step_fresh_resolver(context: Any) -> None:
+    context.resolver = ComponentResolver()
+    context.test_protocol = _TestProtocol
+    context.global_impl = _GlobalDefault()
+    context.project_impl = _ProjectOverride()
+    context.plan_impl = _PlanOverride()
+    context.error = None
+    context.resolution_result = None
+    context.resolution_result_2 = None
+
+
+@given("a global default implementation for a test protocol")
+def step_register_global_default(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.register_global(context.test_protocol, context.global_impl)
+
+
+@given('a project-level override for project "{project_id}" for the test protocol')
+def step_register_project_override(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.register_project(project_id, context.test_protocol, context.project_impl)
+
+
+@given('a plan-level override for plan "{plan_id}" for the test protocol')
+def step_register_plan_override(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.register_plan(plan_id, context.test_protocol, context.plan_impl)
+
+
+@given("an extensions config mapping with a known module path")
+def step_extensions_config(context: Any) -> None:
+    # Use a real module path that exists in the codebase
+    context.extensions_config = {
+        "strategy_selector": (
+            "cleveragents.application.services.acms_pipeline:ConfidenceWeightedSelector"
+        ),
+    }
+    from cleveragents.application.services.acms_service import StrategySelector
+
+    context.type_registry = {
+        "strategy_selector": StrategySelector,
+    }
+    context.ext_component_type = StrategySelector
+
+
+@given("a plan metadata mapping with a known module path")
+def step_plan_metadata_config(context: Any) -> None:
+    context.plan_metadata = {
+        "strategy_selector": (
+            "cleveragents.application.services.acms_pipeline:ConfidenceWeightedSelector"
+        ),
+    }
+    from cleveragents.application.services.acms_service import StrategySelector
+
+    context.type_registry = {
+        "strategy_selector": StrategySelector,
+    }
+    context.ext_component_type = StrategySelector
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I resolve the test protocol with no plan or project context")
+def step_resolve_no_context(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.resolution_result = resolver.resolve(context.test_protocol)
+
+
+@when("I attempt to resolve an unregistered protocol")
+def step_resolve_unregistered(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.resolve(context.test_protocol)
+    except ComponentNotFoundError as exc:
+        context.error = exc
+
+
+@when('I resolve the test protocol for project "{project_id}"')
+def step_resolve_for_project(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.resolution_result = resolver.resolve(
+        context.test_protocol, project_id=project_id
+    )
+
+
+@when('I resolve the test protocol for plan "{plan_id}" and project "{project_id}"')
+def step_resolve_for_plan_and_project(
+    context: Any, plan_id: str, project_id: str
+) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.resolution_result = resolver.resolve(
+        context.test_protocol, plan_id=plan_id, project_id=project_id
+    )
+
+
+@when('I resolve the test protocol for plan "{plan_id}"')
+def step_resolve_for_plan(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.resolution_result = resolver.resolve(context.test_protocol, plan_id=plan_id)
+
+
+@when("I resolve the test protocol twice with no context")
+def step_resolve_twice(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.resolution_result = resolver.resolve(context.test_protocol)
+    context.resolution_result_2 = resolver.resolve(context.test_protocol)
+
+
+@when('I remove the project implementation for "{project_id}" and the test protocol')
+def step_remove_project(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.removal_result = resolver.remove_project(project_id, context.test_protocol)
+
+
+@when('I remove the plan implementation for "{plan_id}" and the test protocol')
+def step_remove_plan(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.removal_result = resolver.remove_plan(plan_id, context.test_protocol)
+
+
+@when("I remove the global implementation for the test protocol")
+def step_remove_global(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.removal_result = resolver.remove_global(context.test_protocol)
+
+
+@when("I attempt to register a project override with empty project_id")
+def step_register_empty_project_id(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.register_project("", context.test_protocol, context.global_impl)
+    except ComponentRegistrationError as exc:
+        context.error = exc
+
+
+@when("I attempt to register a plan override with empty plan_id")
+def step_register_empty_plan_id(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.register_plan("", context.test_protocol, context.global_impl)
+    except ComponentRegistrationError as exc:
+        context.error = exc
+
+
+@when(
+    'I register a project-level override for project "{project_id}" for the test protocol'
+)
+def step_when_register_project_override(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.register_project(project_id, context.test_protocol, context.project_impl)
+
+
+@when('I load project extensions for project "{project_id}"')
+def step_load_project_extensions(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.ext_project_id = project_id
+    resolver.load_project_extensions(
+        project_id, context.extensions_config, context.type_registry
+    )
+
+
+@when('I load plan extensions for plan "{plan_id}"')
+def step_load_plan_extensions(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.ext_plan_id = plan_id
+    resolver.load_plan_extensions(plan_id, context.plan_metadata, context.type_registry)
+
+
+@when("I clear the resolver")
+def step_clear_resolver(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.clear()
+
+
+@when("I register an extension point for the test protocol")
+def step_register_extension_point(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.register_extension_point(
+        context.test_protocol,
+        description="A test extension point",
+        category="test",
+    )
+
+
+@when("I attempt to register None as a global implementation")
+def step_register_none_global(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.register_global(context.test_protocol, None)  # type: ignore[arg-type]
+    except ComponentRegistrationError as exc:
+        context.error = exc
+
+
+@when("I attempt to register None as a project implementation")
+def step_register_none_project(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.register_project("p", context.test_protocol, None)  # type: ignore[arg-type]
+    except ComponentRegistrationError as exc:
+        context.error = exc
+
+
+@when("I attempt to register None as a plan implementation")
+def step_register_none_plan(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver.register_plan("p", context.test_protocol, None)  # type: ignore[arg-type]
+    except ComponentRegistrationError as exc:
+        context.error = exc
+
+
+@when('I attempt to import a component from "{module_path}"')
+def step_import_disallowed(context: Any, module_path: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    try:
+        resolver._import_component(module_path)
+    except (ComponentRegistrationError, ImportError, AttributeError) as exc:
+        context.error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the resolved component should be the global default")
+def step_assert_global_default(context: Any) -> None:
+    assert context.resolution_result is not None
+    assert context.resolution_result.component is context.global_impl
+
+
+@then("the resolved component should be the project override")
+def step_assert_project_override(context: Any) -> None:
+    assert context.resolution_result is not None
+    assert context.resolution_result.component is context.project_impl
+
+
+@then("the resolved component should be the plan override")
+def step_assert_plan_override(context: Any) -> None:
+    assert context.resolution_result is not None
+    assert context.resolution_result.component is context.plan_impl
+
+
+@then('the resolution scope should be "{scope}"')
+def step_assert_scope(context: Any, scope: str) -> None:
+    assert context.resolution_result is not None
+    expected = ScopeLevel(scope)
+    assert context.resolution_result.scope == expected, (
+        f"Expected scope {expected}, got {context.resolution_result.scope}"
+    )
+
+
+@then("a ComponentNotFoundError should be raised")
+def step_assert_not_found_error(context: Any) -> None:
+    assert isinstance(context.error, ComponentNotFoundError), (
+        f"Expected ComponentNotFoundError, got {type(context.error)}"
+    )
+
+
+@then("a ComponentRegistrationError should be raised")
+def step_assert_registration_error(context: Any) -> None:
+    assert isinstance(context.error, ComponentRegistrationError), (
+        f"Expected ComponentRegistrationError, got {type(context.error)}"
+    )
+
+
+@then("has_global should return True for the test protocol")
+def step_assert_has_global_true(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_global(context.test_protocol) is True
+
+
+@then("has_global should return False for the test protocol")
+def step_assert_has_global_false(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_global(context.test_protocol) is False
+
+
+@then('has_project should return True for "{project_id}" and the test protocol')
+def step_assert_has_project_true(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_project(project_id, context.test_protocol) is True
+
+
+@then('has_project should return False for "{project_id}" and the test protocol')
+def step_assert_has_project_false(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_project(project_id, context.test_protocol) is False
+
+
+@then('has_plan should return True for "{plan_id}" and the test protocol')
+def step_assert_has_plan_true(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_plan(plan_id, context.test_protocol) is True
+
+
+@then('has_plan should return False for "{plan_id}" and the test protocol')
+def step_assert_has_plan_false(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_plan(plan_id, context.test_protocol) is False
+
+
+@then("the cache size should be {size:d}")
+def step_assert_cache_size(context: Any, size: int) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.cache_size() == size, (
+        f"Expected cache size {size}, got {resolver.cache_size()}"
+    )
+
+
+@then("both resolutions should return the same result")
+def step_assert_same_result(context: Any) -> None:
+    assert context.resolution_result is context.resolution_result_2
+
+
+@then("the extension should be registered at the project scope")
+def step_assert_extension_at_project(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_project(context.ext_project_id, context.ext_component_type)
+
+
+@then("the extension should be registered at the plan scope")
+def step_assert_extension_at_plan(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    assert resolver.has_plan(context.ext_plan_id, context.ext_component_type)
+
+
+@then("list_global_types should include the test protocol name")
+def step_assert_list_global(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    names = resolver.list_global_types()
+    assert "_TestProtocol" in names, f"Expected _TestProtocol in {names}"
+
+
+@then('list_project_types for "{project_id}" should include the test protocol name')
+def step_assert_list_project(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    names = resolver.list_project_types(project_id)
+    assert "_TestProtocol" in names, f"Expected _TestProtocol in {names}"
+
+
+@then('list_plan_types for "{plan_id}" should include the test protocol name')
+def step_assert_list_plan(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    names = resolver.list_plan_types(plan_id)
+    assert "_TestProtocol" in names, f"Expected _TestProtocol in {names}"
+
+
+@then("list_extension_points should include the test protocol")
+def step_assert_list_extension_points(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    points = resolver.list_extension_points()
+    assert len(points) == 1
+    assert points[0].component_type is context.test_protocol
+
+
+# ---------------------------------------------------------------------------
+# Edge-case steps for coverage improvement
+# ---------------------------------------------------------------------------
+
+
+@given("an extensions config with an unknown extension type name")
+def step_ext_config_unknown_type(context: Any) -> None:
+    context.edge_extensions_config = {"nonexistent_extension": "some:Module"}
+    context.edge_type_registry = {}
+
+
+@given("an extensions config with a bad module path")
+def step_ext_config_bad_path(context: Any) -> None:
+    from cleveragents.application.services.acms_service import StrategySelector
+
+    context.edge_extensions_config = {
+        "strategy_selector": "cleveragents.nonexistent_module:Foo"
+    }
+    context.edge_type_registry = {"strategy_selector": StrategySelector}
+
+
+@given("a plan metadata with an unknown extension type name")
+def step_plan_meta_unknown_type(context: Any) -> None:
+    context.edge_plan_metadata = {"nonexistent_extension": "some:Module"}
+    context.edge_type_registry = {}
+
+
+@given("a plan metadata with a bad module path")
+def step_plan_meta_bad_path(context: Any) -> None:
+    from cleveragents.application.services.acms_service import StrategySelector
+
+    context.edge_plan_metadata = {
+        "strategy_selector": "cleveragents.nonexistent_module:Foo"
+    }
+    context.edge_type_registry = {"strategy_selector": StrategySelector}
+
+
+@when('I load project extensions with unknown type for project "{project_id}"')
+def step_load_proj_ext_unknown(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.loaded = resolver.load_project_extensions(
+        project_id, context.edge_extensions_config, context.edge_type_registry
+    )
+
+
+@when('I load project extensions with bad path for project "{project_id}"')
+def step_load_proj_ext_bad_path(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.loaded = resolver.load_project_extensions(
+        project_id, context.edge_extensions_config, context.edge_type_registry
+    )
+
+
+@when('I load plan extensions with unknown type for plan "{plan_id}"')
+def step_load_plan_ext_unknown(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.loaded = resolver.load_plan_extensions(
+        plan_id, context.edge_plan_metadata, context.edge_type_registry
+    )
+
+
+@when('I load plan extensions with bad path for plan "{plan_id}"')
+def step_load_plan_ext_bad_path(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.loaded = resolver.load_plan_extensions(
+        plan_id, context.edge_plan_metadata, context.edge_type_registry
+    )
+
+
+@when("I explicitly invalidate the cache")
+def step_explicit_invalidate(context: Any) -> None:
+    resolver: ComponentResolver = context.resolver
+    resolver.invalidate_cache()
+
+
+@then("the loaded extensions list should be empty")
+def step_assert_loaded_empty(context: Any) -> None:
+    assert context.loaded == [], f"Expected empty, got {context.loaded}"
+
+
+@then("the component resolver removal result should be False")
+def step_assert_removal_false(context: Any) -> None:
+    assert context.removal_result is False, (
+        f"Expected removal_result to be False, got {context.removal_result!r}"
+    )
+
+
+@then('list_project_types for "{project_id}" should be empty')
+def step_assert_project_types_empty(context: Any, project_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    names = resolver.list_project_types(project_id)
+    assert names == [], f"Expected empty, got {names}"
+
+
+@then('list_plan_types for "{plan_id}" should be empty')
+def step_assert_plan_types_empty(context: Any, plan_id: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    names = resolver.list_plan_types(plan_id)
+    assert names == [], f"Expected empty, got {names}"

robot/component_resolver.robot

@@ -0,0 +1,81 @@
+*** Settings ***
+Documentation    End-to-end tests for ComponentResolver with 3-level scope chain
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_component_resolver.py
+
+*** Test Cases ***
+ComponentResolver Global Default Resolution
+    [Documentation]    Verify global default is resolved when no overrides exist
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    global-default    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-global-ok
+
+ComponentResolver Project Override
+    [Documentation]    Verify project-level override takes precedence over global
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    project-override    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-project-override-ok
+
+ComponentResolver Plan Override
+    [Documentation]    Verify plan-level override takes highest precedence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    plan-override    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-plan-override-ok
+
+ComponentResolver Fallthrough To Global
+    [Documentation]    Verify fallthrough when no plan or project override matches
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    fallthrough    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-fallthrough-ok
+
+ComponentResolver Missing Component Error
+    [Documentation]    Verify ComponentNotFoundError for unregistered components
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    not-found    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-not-found-ok
+
+ComponentResolver Caching Behavior
+    [Documentation]    Verify resolution results are cached
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    caching    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-caching-ok
+
+ComponentResolver Security Module Prefix
+    [Documentation]    Verify disallowed module prefixes are rejected
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    security    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-security-ok
+
+ComponentResolver Introspection APIs
+    [Documentation]    Verify list_global_types, list_project_types, list_plan_types
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    introspection    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-introspection-ok
+
+ComponentResolver Clear All
+    [Documentation]    Verify clear removes all registrations
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    clear    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    component-resolver-clear-ok

robot/helper_component_resolver.py

@@ -0,0 +1,204 @@
+"""Helper script for component_resolver.robot end-to-end tests.
+
+Each subcommand is a self-contained check that prints a sentinel on success.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any, Protocol, runtime_checkable
+
+# Ensure local source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.component_resolver import (  # noqa: E402
+    ComponentNotFoundError,
+    ComponentRegistrationError,
+    ComponentResolver,
+    ScopeLevel,
+)
+
+# ---------------------------------------------------------------------------
+# Test protocol and implementations
+# ---------------------------------------------------------------------------
+
+
+@runtime_checkable
+class _Greeter(Protocol):
+    def greet(self) -> str: ...
+
+
+class _DefaultGreeter:
+    def greet(self) -> str:
+        return "hello-global"
+
+
+class _ProjectGreeter:
+    def greet(self) -> str:
+        return "hello-project"
+
+
+class _PlanGreeter:
+    def greet(self) -> str:
+        return "hello-plan"
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def global_default_resolution() -> None:
+    """Verify global default resolution."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    result = resolver.resolve(_Greeter)
+    if result.scope == ScopeLevel.GLOBAL and result.component.greet() == "hello-global":
+        print("component-resolver-global-ok")
+    else:
+        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
+        sys.exit(1)
+
+
+def project_override() -> None:
+    """Verify project override takes precedence over global."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    resolver.register_project("proj-1", _Greeter, _ProjectGreeter())
+    result = resolver.resolve(_Greeter, project_id="proj-1")
+    if (
+        result.scope == ScopeLevel.PROJECT
+        and result.component.greet() == "hello-project"
+    ):
+        print("component-resolver-project-override-ok")
+    else:
+        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
+        sys.exit(1)
+
+
+def plan_override() -> None:
+    """Verify plan override takes highest precedence."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    resolver.register_project("proj-1", _Greeter, _ProjectGreeter())
+    resolver.register_plan("plan-1", _Greeter, _PlanGreeter())
+    result = resolver.resolve(_Greeter, plan_id="plan-1", project_id="proj-1")
+    if result.scope == ScopeLevel.PLAN and result.component.greet() == "hello-plan":
+        print("component-resolver-plan-override-ok")
+    else:
+        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
+        sys.exit(1)
+
+
+def fallthrough_to_global() -> None:
+    """Verify fallthrough when no plan or project override exists."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    result = resolver.resolve(_Greeter, plan_id="no-plan", project_id="no-project")
+    if result.scope == ScopeLevel.GLOBAL and result.component.greet() == "hello-global":
+        print("component-resolver-fallthrough-ok")
+    else:
+        print(f"FAIL: {result.scope} / {result.component.greet()}", file=sys.stderr)
+        sys.exit(1)
+
+
+def missing_component_error() -> None:
+    """Verify ComponentNotFoundError for unregistered components."""
+    resolver = ComponentResolver()
+    try:
+        resolver.resolve(_Greeter)
+        print("FAIL: expected ComponentNotFoundError", file=sys.stderr)
+        sys.exit(1)
+    except ComponentNotFoundError:
+        print("component-resolver-not-found-ok")
+
+
+def caching_behavior() -> None:
+    """Verify resolution caching."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    r1 = resolver.resolve(_Greeter)
+    r2 = resolver.resolve(_Greeter)
+    if r1 is r2 and resolver.cache_size() == 1:
+        print("component-resolver-caching-ok")
+    else:
+        print(f"FAIL: same={r1 is r2}, cache={resolver.cache_size()}", file=sys.stderr)
+        sys.exit(1)
+
+
+def security_module_prefix() -> None:
+    """Verify module prefix restriction."""
+    resolver = ComponentResolver()
+    try:
+        resolver._import_component("evil_module:Hacker")
+        print("FAIL: expected error", file=sys.stderr)
+        sys.exit(1)
+    except ComponentRegistrationError:
+        print("component-resolver-security-ok")
+
+
+def introspection_apis() -> None:
+    """Verify introspection methods."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    resolver.register_project("p1", _Greeter, _ProjectGreeter())
+    resolver.register_plan("pl1", _Greeter, _PlanGreeter())
+    g = resolver.list_global_types()
+    p = resolver.list_project_types("p1")
+    pl = resolver.list_plan_types("pl1")
+    if "_Greeter" in g and "_Greeter" in p and "_Greeter" in pl:
+        print("component-resolver-introspection-ok")
+    else:
+        print(f"FAIL: global={g}, project={p}, plan={pl}", file=sys.stderr)
+        sys.exit(1)
+
+
+def clear_all() -> None:
+    """Verify clear removes all registrations."""
+    resolver = ComponentResolver()
+    resolver.register_global(_Greeter, _DefaultGreeter())
+    resolver.register_project("p1", _Greeter, _ProjectGreeter())
+    resolver.register_plan("pl1", _Greeter, _PlanGreeter())
+    resolver.clear()
+    ok = (
+        not resolver.has_global(_Greeter)
+        and not resolver.has_project("p1", _Greeter)
+        and not resolver.has_plan("pl1", _Greeter)
+    )
+    if ok:
+        print("component-resolver-clear-ok")
+    else:
+        print("FAIL: clear did not remove all", file=sys.stderr)
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Main dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Any] = {
+    "global-default": global_default_resolution,
+    "project-override": project_override,
+    "plan-override": plan_override,
+    "fallthrough": fallthrough_to_global,
+    "not-found": missing_component_error,
+    "caching": caching_behavior,
+    "security": security_module_prefix,
+    "introspection": introspection_apis,
+    "clear": clear_all,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        print(f"Commands: {', '.join(sorted(_COMMANDS))}", file=sys.stderr)
+        sys.exit(1)
+    cmd = sys.argv[1]
+    fn = _COMMANDS.get(cmd)
+    if fn is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        sys.exit(1)
+    fn()

src/cleveragents/application/services/component_resolver.py

@@ -0,0 +1,717 @@
+"""Pluggable component resolution with 3-level scope chain.
+
+Implements a ``ComponentResolver`` that resolves pluggable components
+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
+across ACMS pipeline components, context strategies, sandbox strategies,
+UKO vocabulary/analyzers, and validation implementations — all of which
+can be resolved through this scope chain.
+
+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
+application startup.
+
+Resolution is deterministic and cached per scope.  Missing
+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.
+
+ISSUES CLOSED: #552
+"""
+
+from __future__ import annotations
+
+import importlib
+import threading
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, TypeVar
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+T = TypeVar("T")
+
+
+# ---------------------------------------------------------------------------
+# Public types
+# ---------------------------------------------------------------------------
+
+
+class ScopeLevel(Enum):
+    """Scope levels from highest to lowest priority."""
+
+    PLAN = "plan"
+    PROJECT = "project"
+    GLOBAL = "global"
+
+
+@dataclass(frozen=True, slots=True)
+class ResolutionResult:
+    """Result of resolving a component through the scope chain.
+
+    Attributes:
+        component: The resolved component instance.
+        scope: The scope level that provided the component.
+        component_type_name: Human-readable name of the component type.
+    """
+
+    component: object
+    scope: ScopeLevel
+    component_type_name: str
+
+
+@dataclass(frozen=True, slots=True)
+class ExtensionPointEntry:
+    """Metadata for a registered extension point.
+
+    Attributes:
+        component_type: The Protocol type for this extension point.
+        description: Human-readable description of the extension point.
+        category: Category grouping (e.g. ``acms``, ``context``, ``sandbox``).
+    """
+
+    component_type: type[Any]
+    description: str
+    category: str
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+
+class ComponentNotFoundError(KeyError):
+    """Raised when no component implementation is found at any scope level."""
+
+
+class ComponentRegistrationError(ValueError):
+    """Raised when component registration fails validation."""
+
+
+# ---------------------------------------------------------------------------
+# Module-level allowed prefixes for dynamic imports (security)
+# ---------------------------------------------------------------------------
+
+_ALLOWED_MODULE_PREFIXES: tuple[str, ...] = ("cleveragents.",)
+
+
+# ---------------------------------------------------------------------------
+# ComponentResolver
+# ---------------------------------------------------------------------------
+
+
+@dataclass(slots=True)
+class _ScopeRegistry:
+    """Internal registry for a single scope level.
+
+    Maps ``type[T]`` to the implementing instance.
+    """
+
+    implementations: dict[type[Any], object] = field(default_factory=dict)
+
+    def get(self, component_type: type[Any]) -> object | None:
+        """Return the implementation for *component_type*, or ``None``."""
+        return self.implementations.get(component_type)
+
+    def set(self, component_type: type[Any], implementation: object) -> None:
+        """Register *implementation* for *component_type*."""
+        self.implementations[component_type] = implementation
+
+    def remove(self, component_type: type[Any]) -> bool:
+        """Remove the implementation for *component_type*.
+
+        Returns ``True`` if an entry was removed, ``False`` otherwise.
+        """
+        if component_type in self.implementations:
+            del self.implementations[component_type]
+            return True
+        return False
+
+    def has(self, component_type: type[Any]) -> bool:
+        """Return ``True`` if *component_type* is registered."""
+        return component_type in self.implementations
+
+    def list_types(self) -> list[type[Any]]:
+        """Return all registered component types."""
+        return list(self.implementations.keys())
+
+
+class ComponentResolver:
+    """Resolves pluggable components through a 3-level scope chain.
+
+    The scope chain determines which implementation of a pluggable
+    component is used:
+
+    1. **Plan-level override** (highest priority) — plan metadata
+       specifies a custom implementation.
+    2. **Project-level override** — ``config.toml`` ``[extensions]``
+       section or programmatic registration.
+    3. **Global default** — built-in default registered at startup.
+
+    Resolution is deterministic and cached per scope.  Missing
+    implementations at a scope level transparently fall through to
+    the next scope.
+
+    Example::
+
+        resolver = ComponentResolver()
+        resolver.register_global(StrategySelector, ConfidenceWeightedSelector())
+        resolver.register_project("my-project", StrategySelector, custom_selector)
+        result = resolver.resolve(StrategySelector, project_id="my-project")
+        assert result.scope == ScopeLevel.PROJECT
+
+    Thread-safety: all public methods are thread-safe.
+
+    Based on ``docs/specification.md`` §§ 44369-44396 and issue #552.
+    """
+
+    # Module prefixes allowed for dynamic import (security).
+    ALLOWED_MODULE_PREFIXES: tuple[str, ...] = _ALLOWED_MODULE_PREFIXES
+
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._global = _ScopeRegistry()
+        self._projects: dict[str, _ScopeRegistry] = {}
+        self._plans: dict[str, _ScopeRegistry] = {}
+        self._cache: dict[
+            tuple[type[Any], str | None, str | None], ResolutionResult
+        ] = {}
+        self._extension_points: dict[type[Any], ExtensionPointEntry] = {}
+        self._logger = logger.bind(service="component_resolver")
+
+    # ------------------------------------------------------------------
+    # Extension point catalog
+    # ------------------------------------------------------------------
+
+    def register_extension_point(
+        self,
+        component_type: type[Any],
+        *,
+        description: str = "",
+        category: str = "",
+    ) -> None:
+        """Register a known extension point for documentation/validation.
+
+        Args:
+            component_type: The Protocol type for this extension point.
+            description: Human-readable description.
+            category: Category grouping.
+        """
+        with self._lock:
+            self._extension_points[component_type] = ExtensionPointEntry(
+                component_type=component_type,
+                description=description,
+                category=category,
+            )
+
+    def list_extension_points(self) -> list[ExtensionPointEntry]:
+        """Return all registered extension points."""
+        with self._lock:
+            return list(self._extension_points.values())
+
+    # ------------------------------------------------------------------
+    # Registration — Global scope
+    # ------------------------------------------------------------------
+
+    def register_global(
+        self,
+        component_type: type[Any],
+        implementation: object,
+    ) -> None:
+        """Register a global default implementation.
+
+        Args:
+            component_type: The Protocol type to register against.
+            implementation: The implementing instance.
+
+        Raises:
+            ComponentRegistrationError: If *implementation* is ``None``.
+        """
+        if implementation is None:
+            msg = (
+                f"Cannot register None as global implementation "
+                f"for {component_type.__name__}"
+            )
+            raise ComponentRegistrationError(msg)
+
+        with self._lock:
+            self._global.set(component_type, implementation)
+            self._invalidate_cache_for_type(component_type)
+            self._logger.debug(
+                "component.registered",
+                scope="global",
+                component_type=component_type.__name__,
+            )
+
+    # ------------------------------------------------------------------
+    # Registration — Project scope
+    # ------------------------------------------------------------------
+
+    def register_project(
+        self,
+        project_id: str,
+        component_type: type[Any],
+        implementation: object,
+    ) -> None:
+        """Register a project-level override.
+
+        Args:
+            project_id: Unique project identifier.
+            component_type: The Protocol type to register against.
+            implementation: The implementing instance.
+
+        Raises:
+            ComponentRegistrationError: If *project_id* is empty or
+                *implementation* is ``None``.
+        """
+        if not project_id:
+            msg = "project_id must not be empty"
+            raise ComponentRegistrationError(msg)
+        if implementation is None:
+            msg = (
+                f"Cannot register None as project implementation "
+                f"for {component_type.__name__}"
+            )
+            raise ComponentRegistrationError(msg)
+
+        with self._lock:
+            if project_id not in self._projects:
+                self._projects[project_id] = _ScopeRegistry()
+            self._projects[project_id].set(component_type, implementation)
+            self._invalidate_cache_for_type(component_type)
+            self._logger.debug(
+                "component.registered",
+                scope="project",
+                project_id=project_id,
+                component_type=component_type.__name__,
+            )
+
+    # ------------------------------------------------------------------
+    # Registration — Plan scope
+    # ------------------------------------------------------------------
+
+    def register_plan(
+        self,
+        plan_id: str,
+        component_type: type[Any],
+        implementation: object,
+    ) -> None:
+        """Register a plan-level override.
+
+        Args:
+            plan_id: Unique plan identifier.
+            component_type: The Protocol type to register against.
+            implementation: The implementing instance.
+
+        Raises:
+            ComponentRegistrationError: If *plan_id* is empty or
+                *implementation* is ``None``.
+        """
+        if not plan_id:
+            msg = "plan_id must not be empty"
+            raise ComponentRegistrationError(msg)
+        if implementation is None:
+            msg = (
+                f"Cannot register None as plan implementation "
+                f"for {component_type.__name__}"
+            )
+            raise ComponentRegistrationError(msg)
+
+        with self._lock:
+            if plan_id not in self._plans:
+                self._plans[plan_id] = _ScopeRegistry()
+            self._plans[plan_id].set(component_type, implementation)
+            self._invalidate_cache_for_type(component_type)
+            self._logger.debug(
+                "component.registered",
+                scope="plan",
+                plan_id=plan_id,
+                component_type=component_type.__name__,
+            )
+
+    # ------------------------------------------------------------------
+    # Resolution
+    # ------------------------------------------------------------------
+
+    def resolve(
+        self,
+        component_type: type[Any],
+        *,
+        plan_id: str | None = None,
+        project_id: str | None = None,
+    ) -> ResolutionResult:
+        """Resolve a component through the scope chain: plan > project > global.
+
+        Args:
+            component_type: The Protocol type to resolve.
+            plan_id: Optional plan identifier for plan-level lookup.
+            project_id: Optional project identifier for project-level lookup.
+
+        Returns:
+            A ``ResolutionResult`` with the component instance and the
+            scope level that provided it.
+
+        Raises:
+            ComponentNotFoundError: If no implementation is found at
+                any scope level.
+        """
+        cache_key = (component_type, plan_id, project_id)
+
+        with self._lock:
+            cached = self._cache.get(cache_key)
+            if cached is not None:
+                return cached
+
+            result = self._resolve_uncached(component_type, plan_id, project_id)
+            self._cache[cache_key] = result
+            return result
+
+    def _resolve_uncached(
+        self,
+        component_type: type[Any],
+        plan_id: str | None,
+        project_id: str | None,
+    ) -> ResolutionResult:
+        """Internal resolution without cache (must hold lock)."""
+        type_name = component_type.__name__
+
+        # Level 1: Plan scope
+        if plan_id:
+            plan_registry = self._plans.get(plan_id)
+            if plan_registry is not None:
+                impl = plan_registry.get(component_type)
+                if impl is not None:
+                    self._logger.debug(
+                        "component.resolved",
+                        scope="plan",
+                        plan_id=plan_id,
+                        component_type=type_name,
+                    )
+                    return ResolutionResult(
+                        component=impl,
+                        scope=ScopeLevel.PLAN,
+                        component_type_name=type_name,
+                    )
+
+        # Level 2: Project scope
+        if project_id:
+            project_registry = self._projects.get(project_id)
+            if project_registry is not None:
+                impl = project_registry.get(component_type)
+                if impl is not None:
+                    self._logger.debug(
+                        "component.resolved",
+                        scope="project",
+                        project_id=project_id,
+                        component_type=type_name,
+                    )
+                    return ResolutionResult(
+                        component=impl,
+                        scope=ScopeLevel.PROJECT,
+                        component_type_name=type_name,
+                    )
+
+        # Level 3: Global scope
+        impl = self._global.get(component_type)
+        if impl is not None:
+            self._logger.debug(
+                "component.resolved",
+                scope="global",
+                component_type=type_name,
+            )
+            return ResolutionResult(
+                component=impl,
+                scope=ScopeLevel.GLOBAL,
+                component_type_name=type_name,
+            )
+
+        msg = (
+            f"No implementation found for {type_name} at any scope level. "
+            f"Checked: plan={plan_id!r}, project={project_id!r}, global."
+        )
+        raise ComponentNotFoundError(msg)
+
+    # ------------------------------------------------------------------
+    # Config.toml extension loading
+    # ------------------------------------------------------------------
+
+    def load_project_extensions(
+        self,
+        project_id: str,
+        extensions_config: dict[str, str],
+        type_registry: dict[str, type[Any]],
+    ) -> list[str]:
+        """Load project-level overrides from ``config.toml`` ``[extensions]``.
+
+        Parses a mapping of extension point names to ``"module:ClassName"``
+        strings and registers them at the project scope.
+
+        Args:
+            project_id: The project to register overrides for.
+            extensions_config: Mapping of extension point name to
+                ``"module:ClassName"`` import path.
+            type_registry: Mapping of extension point name to the
+                Protocol type class.
+
+        Returns:
+            List of successfully loaded extension point names.
+
+        Example config.toml::
+
+            [extensions]
+            strategy_selector = "my_module:CustomSelector"
+            budget_allocator = "my_module:CustomAllocator"
+        """
+        loaded: list[str] = []
+        for ext_name, module_path in extensions_config.items():
+            component_type = type_registry.get(ext_name)
+            if component_type is None:
+                self._logger.warning(
+                    "extension.unknown_type",
+                    extension=ext_name,
+                    project_id=project_id,
+                )
+                continue
+
+            try:
+                instance = self._import_component(module_path)
+                self.register_project(project_id, component_type, instance)
+                loaded.append(ext_name)
+                self._logger.info(
+                    "extension.loaded",
+                    extension=ext_name,
+                    module_path=module_path,
+                    project_id=project_id,
+                )
+            except (ComponentRegistrationError, ImportError, AttributeError) as exc:
+                self._logger.warning(
+                    "extension.load_failed",
+                    extension=ext_name,
+                    module_path=module_path,
+                    project_id=project_id,
+                    error=str(exc),
+                )
+
+        return loaded
+
+    def load_plan_extensions(
+        self,
+        plan_id: str,
+        plan_metadata: dict[str, str],
+        type_registry: dict[str, type[Any]],
+    ) -> list[str]:
+        """Load plan-level overrides from plan metadata.
+
+        Args:
+            plan_id: The plan to register overrides for.
+            plan_metadata: Mapping of extension point name to
+                ``"module:ClassName"`` import path from plan metadata.
+            type_registry: Mapping of extension point name to the
+                Protocol type class.
+
+        Returns:
+            List of successfully loaded extension point names.
+        """
+        loaded: list[str] = []
+        for ext_name, module_path in plan_metadata.items():
+            component_type = type_registry.get(ext_name)
+            if component_type is None:
+                self._logger.warning(
+                    "extension.unknown_type",
+                    extension=ext_name,
+                    plan_id=plan_id,
+                )
+                continue
+
+            try:
+                instance = self._import_component(module_path)
+                self.register_plan(plan_id, component_type, instance)
+                loaded.append(ext_name)
+                self._logger.info(
+                    "extension.loaded",
+                    extension=ext_name,
+                    module_path=module_path,
+                    plan_id=plan_id,
+                )
+            except (ComponentRegistrationError, ImportError, AttributeError) as exc:
+                self._logger.warning(
+                    "extension.load_failed",
+                    extension=ext_name,
+                    module_path=module_path,
+                    plan_id=plan_id,
+                    error=str(exc),
+                )
+
+        return loaded
+
+    # ------------------------------------------------------------------
+    # Introspection
+    # ------------------------------------------------------------------
+
+    def has_global(self, component_type: type[Any]) -> bool:
+        """Return ``True`` if a global implementation is registered."""
+        with self._lock:
+            return self._global.has(component_type)
+
+    def has_project(self, project_id: str, component_type: type[Any]) -> bool:
+        """Return ``True`` if a project-level override is registered."""
+        with self._lock:
+            registry = self._projects.get(project_id)
+            if registry is None:
+                return False
+            return registry.has(component_type)
+
+    def has_plan(self, plan_id: str, component_type: type[Any]) -> bool:
+        """Return ``True`` if a plan-level override is registered."""
+        with self._lock:
+            registry = self._plans.get(plan_id)
+            if registry is None:
+                return False
+            return registry.has(component_type)
+
+    def list_global_types(self) -> list[str]:
+        """Return names of all globally registered component types."""
+        with self._lock:
+            return [t.__name__ for t in self._global.list_types()]
+
+    def list_project_types(self, project_id: str) -> list[str]:
+        """Return names of component types overridden for a project."""
+        with self._lock:
+            registry = self._projects.get(project_id)
+            if registry is None:
+                return []
+            return [t.__name__ for t in registry.list_types()]
+
+    def list_plan_types(self, plan_id: str) -> list[str]:
+        """Return names of component types overridden for a plan."""
+        with self._lock:
+            registry = self._plans.get(plan_id)
+            if registry is None:
+                return []
+            return [t.__name__ for t in registry.list_types()]
+
+    # ------------------------------------------------------------------
+    # Removal
+    # ------------------------------------------------------------------
+
+    def remove_global(self, component_type: type[Any]) -> bool:
+        """Remove a global implementation.
+
+        Returns ``True`` if an entry was removed.
+        """
+        with self._lock:
+            removed = self._global.remove(component_type)
+            if removed:
+                self._invalidate_cache_for_type(component_type)
+            return removed
+
+    def remove_project(self, project_id: str, component_type: type[Any]) -> bool:
+        """Remove a project-level override.
+
+        Returns ``True`` if an entry was removed.
+        """
+        with self._lock:
+            registry = self._projects.get(project_id)
+            if registry is None:
+                return False
+            removed = registry.remove(component_type)
+            if removed:
+                self._invalidate_cache_for_type(component_type)
+            return removed
+
+    def remove_plan(self, plan_id: str, component_type: type[Any]) -> bool:
+        """Remove a plan-level override.
+
+        Returns ``True`` if an entry was removed.
+        """
+        with self._lock:
+            registry = self._plans.get(plan_id)
+            if registry is None:
+                return False
+            removed = registry.remove(component_type)
+            if removed:
+                self._invalidate_cache_for_type(component_type)
+            return removed
+
+    def clear(self) -> None:
+        """Remove all registrations and clear the cache."""
+        with self._lock:
+            self._global = _ScopeRegistry()
+            self._projects.clear()
+            self._plans.clear()
+            self._cache.clear()
+
+    # ------------------------------------------------------------------
+    # Cache management
+    # ------------------------------------------------------------------
+
+    def invalidate_cache(self) -> None:
+        """Clear the entire resolution cache.
+
+        Useful after bulk registration changes.
+        """
+        with self._lock:
+            self._cache.clear()
+
+    def cache_size(self) -> int:
+        """Return the number of cached resolution results."""
+        with self._lock:
+            return len(self._cache)
+
+    def _invalidate_cache_for_type(self, component_type: type[Any]) -> None:
+        """Remove cache entries involving *component_type* (must hold lock)."""
+        to_remove = [key for key in self._cache if key[0] is component_type]
+        for key in to_remove:
+            del self._cache[key]
+
+    # ------------------------------------------------------------------
+    # Dynamic import helper
+    # ------------------------------------------------------------------
+
+    def _import_component(self, module_path: str) -> object:
+        """Import and instantiate a component from ``"module:ClassName"``.
+
+        Args:
+            module_path: Import path in ``"module:ClassName"`` format.
+
+        Returns:
+            An instance of the imported class.
+
+        Raises:
+            ComponentRegistrationError: If the format is invalid or the
+                module prefix is not allowed.
+            ImportError: If the module cannot be imported.
+            AttributeError: If the class is not found in the module.
+        """
+        if ":" not in module_path:
+            msg = (
+                f"Invalid module_path '{module_path}': "
+                f"expected 'module:ClassName' format"
+            )
+            raise ComponentRegistrationError(msg)
+
+        module_name, class_name = module_path.rsplit(":", 1)
+
+        # Security: restrict to allowed prefixes
+        if self.ALLOWED_MODULE_PREFIXES and not any(
+            module_name.startswith(prefix) for prefix in self.ALLOWED_MODULE_PREFIXES
+        ):
+            msg = (
+                f"Module '{module_name}' is not in the allowed prefix list: "
+                f"{self.ALLOWED_MODULE_PREFIXES}. Only modules under these "
+                f"prefixes may be dynamically imported."
+            )
+            raise ComponentRegistrationError(msg)
+
+        module = importlib.import_module(module_name)
+        cls = getattr(module, class_name)
+        return cls()

vulture_whitelist.py

@@ -870,3 +870,30 @@ PlanDecisionContextStrategy  # noqa: B018, F821
 max_iterations  # noqa: B018, F821
 _temporal_score  # noqa: B018, F821
 _extract_node_prefix  # noqa: B018, F821
+
+# ComponentResolver — pluggable scope chain resolution (#552)
+ComponentResolver  # noqa: B018, F821
+ComponentNotFoundError  # noqa: B018, F821
+ComponentRegistrationError  # noqa: B018, F821
+ScopeLevel  # noqa: B018, F821
+ResolutionResult  # noqa: B018, F821
+ExtensionPointEntry  # noqa: B018, F821
+register_global  # noqa: B018, F821
+register_project  # noqa: B018, F821
+register_plan  # noqa: B018, F821
+resolve  # noqa: B018, F821
+load_project_extensions  # noqa: B018, F821
+load_plan_extensions  # noqa: B018, F821
+has_global  # noqa: B018, F821
+has_project  # noqa: B018, F821
+has_plan  # noqa: B018, F821
+list_global_types  # noqa: B018, F821
+list_project_types  # noqa: B018, F821
+list_plan_types  # noqa: B018, F821
+remove_global  # noqa: B018, F821
+remove_project  # noqa: B018, F821
+remove_plan  # noqa: B018, F821
+invalidate_cache  # noqa: B018, F821
+cache_size  # noqa: B018, F821
+register_extension_point  # noqa: B018, F821
+list_extension_points  # noqa: B018, F821