features/scope_chain_extension_api.feature

@@ -0,0 +1,130 @@
+Feature: Pluggable Scope Chain Resolution Extension API
+  As a developer
+  I want to register custom scope chain resolvers
+  So that I can extend scope resolution with custom context sources
+
+  Background:
+    Given a fresh scope chain registry
+
+  Scenario: Register a custom scope chain resolver
+    Given a custom resolver named "test_resolver" with version "1.0.0"
+    When I register the resolver
+    Then the resolver should be registered
+    And the registry should list the resolver
+
+  Scenario: Unregister a custom resolver
+    Given a custom resolver named "test_resolver" with version "1.0.0"
+    And the resolver is registered
+    When I unregister the resolver
+    Then the resolver should not be registered
+
+  Scenario: Reject duplicate resolver registration
+    Given a custom resolver named "test_resolver" with version "1.0.0"
+    And the resolver is registered
+    When I try to register another resolver with the same name
+    Then registration should fail with ValueError
+
+  Scenario: Reject resolver with empty name
+    Given a custom resolver with empty name
+    When I try to register the resolver
+    Then registration should fail with ValueError
+
+  Scenario: Reject non-protocol resolver
+    Given an object that does not implement ScopeChainResolver
+    When I try to register it as a resolver
+    Then registration should fail with TypeError
+
+  Scenario: Select resolver based on context
+    Given a custom resolver that supports context with source="custom"
+    And the resolver is registered
+    When I select a resolver with context source="custom"
+    Then the custom resolver should be selected
+
+  Scenario: Fallback to default resolver when no match
+    Given a custom resolver that supports context with source="custom"
+    And the resolver is registered
+    When I select a resolver with context source="other"
+    Then the default resolver should be selected
+
+  Scenario: Default resolver always supports any context
+    When I select a resolver with no context
+    Then the default resolver should be selected
+
+  Scenario: List all registered resolvers
+    Given a custom resolver named "resolver1" with version "1.0.0"
+    And a custom resolver named "resolver2" with version "2.0.0"
+    And both resolvers are registered
+    When I list all registered resolvers
+    Then the list should contain both resolvers with their versions
+
+  Scenario: Resolve scope using selected resolver
+    Given a custom resolver that delegates to default resolver
+    And the resolver is registered
+    And a project with linked resources
+    When I resolve scope using the registry
+    Then a ResourceScope should be returned
+    And the scope should contain the project's resources
+
+  Scenario: Load resolvers from plugin directory
+    Given a plugin directory with a resolver module
+    And the resolver module exports a valid resolver
+    When I load plugins from the directory
+    Then the resolver should be registered
+    And the loader should report success
+
+  Scenario: Handle missing plugin directory gracefully
+    Given a non-existent plugin directory path
+    When I try to load plugins from the directory
+    Then loading should fail with ValueError
+
+  Scenario: Skip invalid plugin modules
+    Given a plugin directory with an invalid module
+    When I load plugins from the directory
+    Then loading should continue
+    And invalid modules should be logged as warnings
+
+  Scenario: Filesystem resolver supports scope config files
+    Given a project directory with .scope-chain.yaml
+    And a filesystem resolver
+    When I check if the resolver supports the project context
+    Then the resolver should support the context
+
+  Scenario: Filesystem resolver does not support missing config
+    Given a project directory without scope config
+    And a filesystem resolver
+    When I check if the resolver supports the project context
+    Then the resolver should not support the context
+
+  Scenario: Filesystem resolver delegates to default resolver
+    Given a filesystem resolver
+    And a project with linked resources
+    When I resolve scope using the filesystem resolver
+    Then a ResourceScope should be returned
+    And the scope should contain the project's resources
+
+  Scenario: Thread-safe resolver registration
+    Given a registry with multiple threads
+    When threads concurrently register resolvers
+    Then all resolvers should be registered
+    And no race conditions should occur
+
+  Scenario: Thread-safe resolver selection
+    Given a registry with registered resolvers
+    When threads concurrently select resolvers
+    Then all selections should succeed
+    And no race conditions should occur
+
+  Scenario: Resolver version tracking
+    Given a custom resolver with version "2.5.3"
+    And the resolver is registered
+    When I retrieve the resolver
+    Then the resolver version should be "2.5.3"
+
+  Scenario: Multiple resolvers with different contexts
+    Given a resolver that supports context with type="database"
+    And a resolver that supports context with type="wiki"
+    And both resolvers are registered
+    When I select a resolver with context type="database"
+    Then the database resolver should be selected
+    When I select a resolver with context type="wiki"
+    Then the wiki resolver should be selected

features/steps/scope_chain_extension_api_steps.py

@@ -0,0 +1,577 @@
+"""Step definitions for scope chain extension API tests."""
+
+from __future__ import annotations
+
+import tempfile
+import threading
+from pathlib import Path
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+
+from cleveragents.domain.models.acms.scope_chain_extension import (
+    ScopeChainResolver,
+)
+from cleveragents.domain.models.acms.scope_chain_registry import (
+    ScopeChainRegistry,
+)
+from cleveragents.domain.models.acms.scope_resolution import (
+    resolve_resource_scope,
+)
+from cleveragents.infrastructure.plugins.extensions.filesystem_scope_resolver import (
+    FilesystemScopeResolver,
+)
+from cleveragents.infrastructure.plugins.scope_chain_loader import (
+    ScopeChainPluginLoader,
+)
+
+if TYPE_CHECKING:
+    pass
+
+
+@given("a fresh scope chain registry")
+def step_fresh_registry(context):
+    """Create a fresh registry for testing."""
+    context.registry = ScopeChainRegistry()
+    context.resolvers = {}
+    context.selected_resolver = None
+    context.error = None
+
+
+@given('a custom resolver named "{name}" with version "{version}"')
+def step_custom_resolver(context, name, version):
+    """Create a custom resolver with given name and version."""
+    resolver = _create_mock_resolver(name, version)
+    context.resolvers[name] = resolver
+
+
+@given('a custom resolver with version "{version}"')
+def step_custom_resolver_version_only(context, version):
+    """Create a custom resolver with given version and a default name."""
+    resolver = _create_mock_resolver("versioned_resolver", version)
+    context.resolvers["versioned_resolver"] = resolver
+
+
+@given("a custom resolver with empty name")
+def step_resolver_empty_name(context):
+    """Create a resolver with empty name."""
+    resolver = _create_mock_resolver("", "1.0.0")
+    context.resolvers["empty"] = resolver
+
+
+@given("an object that does not implement ScopeChainResolver")
+def step_non_resolver_object(context):
+    """Create an object that doesn't implement the protocol."""
+    context.non_resolver = {"not": "a resolver"}
+
+
+@given('a custom resolver that supports context with source="{source}"')
+def step_resolver_with_context_support(context, source):
+    """Create a resolver that supports specific context."""
+    resolver = _create_mock_resolver(f"resolver_{source}", "1.0.0")
+    resolver.supports = lambda ctx: ctx.get("source") == source
+    context.resolvers[f"resolver_{source}"] = resolver
+
+
+@given("a custom resolver that delegates to default resolver")
+def step_resolver_delegates_default(context):
+    """Create a resolver that delegates to default resolver."""
+    resolver = _create_delegating_resolver()
+    context.resolvers["delegating"] = resolver
+
+
+@given("the resolver is registered")
+def step_register_resolver(context):
+    """Register the most recently created resolver."""
+    for name, resolver in context.resolvers.items():
+        if name not in getattr(context, "registered", set()):
+            context.registry.register(resolver)
+            if not hasattr(context, "registered"):
+                context.registered = set()
+            context.registered.add(name)
+
+
+@given("both resolvers are registered")
+def step_register_both_resolvers(context):
+    """Register all created resolvers."""
+    for _name, resolver in context.resolvers.items():
+        context.registry.register(resolver)
+
+
+@given("a project with linked resources")
+def step_project_with_resources(context):
+    """Create a mock project with linked resources."""
+    context.project = _create_mock_project()
+
+
+@given("a plugin directory with a resolver module")
+def step_plugin_directory_with_resolver(context):
+    """Create a temporary plugin directory with a resolver module."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    plugin_dir = Path(context.temp_dir.name)
+
+    # Create a resolver module
+    resolver_code = '''
+from cleveragents.domain.models.acms.scope_resolution import resolve_resource_scope
+
+class TestPluginResolver:
+    @property
+    def name(self):
+        return "test_plugin"
+
+    @property
+    def version(self):
+        return "1.0.0"
+
+    def supports(self, context):
+        return context.get("source") == "plugin"
+
+    def resolve(self, projects, **kwargs):
+        return resolve_resource_scope(projects, **kwargs)
+'''
+    (plugin_dir / "test_resolver.py").write_text(resolver_code)
+    context.plugin_dir = plugin_dir
+
+
+@given("the resolver module exports a valid resolver")
+def step_valid_resolver_module(context):
+    """Ensure the resolver module is valid (already done in previous step)."""
+    pass
+
+
+@given("a non-existent plugin directory path")
+def step_nonexistent_plugin_dir(context):
+    """Set a non-existent plugin directory path."""
+    context.plugin_dir = Path("/nonexistent/plugin/directory")
+
+
+@given("a plugin directory with an invalid module")
+def step_plugin_dir_invalid_module(context):
+    """Create a plugin directory with an invalid module."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    plugin_dir = Path(context.temp_dir.name)
+
+    # Create an invalid Python module
+    invalid_code = "this is not valid python code !!!"
+    (plugin_dir / "invalid.py").write_text(invalid_code)
+    context.plugin_dir = plugin_dir
+
+
+@given("a project directory with .scope-chain.yaml")
+def step_project_with_scope_config(context):
+    """Create a project directory with scope config file."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    project_dir = Path(context.temp_dir.name)
+    (project_dir / ".scope-chain.yaml").write_text("include_resources: [res1]")
+    context.project_dir = project_dir
+
+
+@given("a project directory without scope config")
+def step_project_without_scope_config(context):
+    """Create a project directory without scope config."""
+    context.temp_dir = tempfile.TemporaryDirectory()
+    context.project_dir = Path(context.temp_dir.name)
+
+
+@given("a filesystem resolver")
+def step_filesystem_resolver(context):
+    """Create a filesystem resolver instance."""
+    context.filesystem_resolver = FilesystemScopeResolver()
+
+
+@given("a registry with multiple threads")
+def step_registry_with_threads(context):
+    """Create a registry for thread safety testing."""
+    context.registry = ScopeChainRegistry()
+    context.thread_errors = []
+
+
+@given("a registry with registered resolvers")
+def step_registry_with_resolvers(context):
+    """Create a registry with some resolvers."""
+    context.registry = ScopeChainRegistry()
+    context.thread_errors = []
+    for i in range(3):
+        resolver = _create_mock_resolver(f"resolver_{i}", "1.0.0")
+        context.registry.register(resolver)
+
+
+@given('a resolver that supports context with type="{type_val}"')
+def step_resolver_with_type_support(context, type_val):
+    """Create a resolver that supports specific type."""
+    resolver = _create_mock_resolver(f"resolver_{type_val}", "1.0.0")
+    resolver.supports = lambda ctx: ctx.get("type") == type_val
+    context.resolvers[f"resolver_{type_val}"] = resolver
+
+
+@when("I register the resolver")
+def step_register_resolver_action(context):
+    """Register a resolver."""
+    try:
+        resolver = next(iter(context.resolvers.values()))
+        context.registry.register(resolver)
+        context.error = None
+    except Exception as e:
+        context.error = e
+
+
+@when("I unregister the resolver")
+def step_unregister_resolver(context):
+    """Unregister a resolver."""
+    resolver_name = next(iter(context.resolvers.keys()))
+    context.unregister_result = context.registry.unregister(resolver_name)
+
+
+@when("I try to register another resolver with the same name")
+def step_register_duplicate(context):
+    """Try to register a resolver with duplicate name."""
+    try:
+        resolver = next(iter(context.resolvers.values()))
+        context.registry.register(resolver)
+        context.registry.register(resolver)
+        context.error = None
+    except ValueError as e:
+        context.error = e
+
+
+@when("I try to register the resolver")
+def step_try_register(context):
+    """Try to register a resolver."""
+    try:
+        resolver = next(iter(context.resolvers.values()))
+        context.registry.register(resolver)
+        context.error = None
+    except Exception as e:
+        context.error = e
+
+
+@when("I try to register it as a resolver")
+def step_try_register_non_resolver(context):
+    """Try to register a non-resolver object."""
+    try:
+        context.registry.register(context.non_resolver)
+        context.error = None
+    except TypeError as e:
+        context.error = e
+
+
+@when('I select a resolver with context source="{source}"')
+def step_select_resolver_with_source(context, source):
+    """Select a resolver with specific context."""
+    context.selected_resolver = context.registry.select_resolver(
+        {"source": source}
+    )
+
+
+@when("I select a resolver with no context")
+def step_select_resolver_no_context(context):
+    """Select resolver without context."""
+    context.selected_resolver = context.registry.select_resolver(None)
+
+
+@when("I list all registered resolvers")
+def step_list_resolvers(context):
+    """List all registered resolvers."""
+    context.resolver_list = context.registry.list_all()
+
+
+@when("I resolve scope using the registry")
+def step_resolve_scope_registry(context):
+    """Resolve scope using the registry."""
+    try:
+        context.resolved_scope = context.registry.resolve(
+            [context.project],
+            context={"source": "test"},
+        )
+        context.error = None
+    except Exception as e:
+        context.error = e
+
+
+@when("I load plugins from the directory")
+def step_load_plugins(context):
+    """Load plugins from directory."""
+    try:
+        loader = ScopeChainPluginLoader(context.registry)
+        context.loaded_count = loader.load_from_directory(context.plugin_dir)
+        context.error = None
+    except Exception as e:
+        context.error = e
+
+
+@when("I try to load plugins from the directory")
+def step_try_load_plugins(context):
+    """Try to load plugins from directory."""
+    try:
+        loader = ScopeChainPluginLoader(context.registry)
+        context.loaded_count = loader.load_from_directory(context.plugin_dir)
+        context.error = None
+    except ValueError as e:
+        context.error = e
+
+
+@when("I check if the resolver supports the project context")
+def step_check_resolver_support(context):
+    """Check if resolver supports project context."""
+    context.supports_result = context.filesystem_resolver.supports(
+        {"project_root": str(context.project_dir)}
+    )
+
+
+@when("threads concurrently register resolvers")
+def step_concurrent_register(context):
+    """Register resolvers from multiple threads."""
+    def register_resolver(idx):
+        try:
+            resolver = _create_mock_resolver(f"thread_resolver_{idx}", "1.0.0")
+            context.registry.register(resolver)
+        except Exception as e:
+            context.thread_errors.append(e)
+
+    threads = [
+        threading.Thread(target=register_resolver, args=(i,))
+        for i in range(10)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+@when("threads concurrently select resolvers")
+def step_concurrent_select(context):
+    """Select resolvers from multiple threads."""
+    context.selection_results = []
+
+    def select_resolver(idx):
+        try:
+            result = context.registry.select_resolver({"index": idx})
+            context.selection_results.append(result)
+        except Exception as e:
+            context.thread_errors.append(e)
+
+    threads = [
+        threading.Thread(target=select_resolver, args=(i,))
+        for i in range(10)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+@when("I retrieve the resolver")
+def step_retrieve_resolver(context):
+    """Retrieve a registered resolver."""
+    resolver_name = next(iter(context.resolvers.keys()))
+    context.retrieved_resolver = context.registry.get(resolver_name)
+
+
+@when('I select a resolver with context type="{type_val}"')
+def step_select_resolver_by_type(context, type_val):
+    """Select resolver by type context."""
+    context.selected_resolver = context.registry.select_resolver(
+        {"type": type_val}
+    )
+
+
+@when("I resolve scope using the filesystem resolver")
+def step_resolve_with_filesystem(context):
+    """Resolve scope using filesystem resolver."""
+    try:
+        context.resolved_scope = context.filesystem_resolver.resolve(
+            [context.project]
+        )
+        context.error = None
+    except Exception as e:
+        context.error = e
+
+
+@then("the resolver should be registered")
+def step_assert_registered(context):
+    """Assert resolver is registered."""
+    resolver_name = next(iter(context.resolvers.keys()))
+    assert context.registry.get(resolver_name) is not None
+
+
+@then("the registry should list the resolver")
+def step_assert_in_list(context):
+    """Assert resolver is in the list."""
+    resolver_list = context.registry.list_all()
+    resolver_names = [name for name, _ in resolver_list]
+    resolver_name = next(iter(context.resolvers.keys()))
+    assert resolver_name in resolver_names
+
+
+@then("the resolver should not be registered")
+def step_assert_not_registered(context):
+    """Assert resolver is not registered."""
+    resolver_name = next(iter(context.resolvers.keys()))
+    assert context.registry.get(resolver_name) is None
+
+
+@then("registration should fail with ValueError")
+def step_assert_value_error(context):
+    """Assert ValueError was raised."""
+    assert isinstance(context.error, ValueError)
+
+
+@then("registration should fail with TypeError")
+def step_assert_type_error(context):
+    """Assert TypeError was raised."""
+    assert isinstance(context.error, TypeError)
+
+
+@then("the custom resolver should be selected")
+def step_assert_custom_selected(context):
+    """Assert custom resolver was selected."""
+    assert context.selected_resolver.name != "default"
+
+
+@then("the default resolver should be selected")
+def step_assert_default_selected(context):
+    """Assert default resolver was selected."""
+    assert context.selected_resolver.name == "default"
+
+
+@then("the list should contain both resolvers with their versions")
+def step_assert_list_contains_both(context):
+    """Assert list contains both resolvers."""
+    assert len(context.resolver_list) == 2
+    names = [name for name, _ in context.resolver_list]
+    assert "resolver1" in names
+    assert "resolver2" in names
+
+
+@then("a ResourceScope should be returned")
+def step_assert_scope_returned(context):
+    """Assert a ResourceScope was returned."""
+    assert context.resolved_scope is not None
+    assert hasattr(context.resolved_scope, "resource_ids")
+
+
+@then("the scope should contain the project's resources")
+def step_assert_scope_has_resources(context):
+    """Assert scope contains project resources."""
+    assert len(context.resolved_scope.resource_ids) > 0
+
+
+@then("the loader should report success")
+def step_assert_loader_success(context):
+    """Assert loader reported success."""
+    assert context.loaded_count > 0
+
+
+@then("loading should fail with ValueError")
+def step_assert_load_value_error(context):
+    """Assert ValueError was raised during loading."""
+    assert isinstance(context.error, ValueError)
+
+
+@then("loading should continue")
+def step_assert_load_continues(context):
+    """Assert loading continued despite errors."""
+    assert context.error is None
+
+
+@then("invalid modules should be logged as warnings")
+def step_assert_warnings_logged(context):
+    """Assert warnings were logged (implicit in implementation)."""
+    pass
+
+
+@then("the resolver should support the context")
+def step_assert_supports_context(context):
+    """Assert resolver supports the context."""
+    assert context.supports_result is True
+
+
+@then("the resolver should not support the context")
+def step_assert_not_supports_context(context):
+    """Assert resolver does not support the context."""
+    assert context.supports_result is False
+
+
+@then("all resolvers should be registered")
+def step_assert_all_registered(context):
+    """Assert all resolvers were registered."""
+    assert len(context.thread_errors) == 0
+    assert len(context.registry.list_all()) == 10
+
+
+@then("all selections should succeed")
+def step_assert_all_selections_succeed(context):
+    """Assert all concurrent selections succeeded."""
+    assert len(context.thread_errors) == 0
+    assert len(context.selection_results) == 10
+
+
+@then("no race conditions should occur")
+def step_assert_no_race_conditions(context):
+    """Assert no race conditions occurred."""
+    assert len(context.thread_errors) == 0
+
+
+@then('the resolver version should be "{version}"')
+def step_assert_resolver_version(context, version):
+    """Assert resolver has correct version."""
+    assert context.retrieved_resolver.version == version
+
+
+@then("the database resolver should be selected")
+def step_assert_database_selected(context):
+    """Assert database resolver was selected."""
+    assert context.selected_resolver.name == "resolver_database"
+
+
+@then("the wiki resolver should be selected")
+def step_assert_wiki_selected(context):
+    """Assert wiki resolver was selected."""
+    assert context.selected_resolver.name == "resolver_wiki"
+
+
+# Helper functions
+
+
+def _create_mock_resolver(name: str, version: str):
+    """Create a mock resolver for testing."""
+    resolver = MagicMock(spec=ScopeChainResolver)
+    resolver.name = name
+    resolver.version = version
+    resolver.supports = lambda ctx: False
+    resolver.resolve = lambda projects, **kwargs: _create_mock_scope()
+    return resolver
+
+
+def _create_delegating_resolver():
+    """Create a resolver that delegates to default resolver."""
+    resolver = MagicMock(spec=ScopeChainResolver)
+    resolver.name = "delegating"
+    resolver.version = "1.0.0"
+    resolver.supports = lambda ctx: True
+    resolver.resolve = lambda projects, **kwargs: resolve_resource_scope(
+        projects, **kwargs
+    )
+    return resolver
+
+
+def _create_mock_project():
+    """Create a mock project with linked resources."""
+    project = MagicMock()
+    project.namespaced_name = "test/project"
+    project.linked_resources = [
+        MagicMock(resource_id="res1", alias="resource1"),
+        MagicMock(resource_id="res2", alias="resource2"),
+    ]
+    project.get_linked_resource = lambda rid: None
+    project.get_linked_resource_by_alias = lambda alias: None
+    return project
+
+
+def _create_mock_scope():
+    """Create a mock ResourceScope."""
+    scope = MagicMock()
+    scope.resource_ids = frozenset(["res1", "res2"])
+    scope.project_names = frozenset(["test/project"])
+    return scope

src/cleveragents/domain/models/acms/scope_chain_extension.py

@@ -0,0 +1,127 @@
+"""Pluggable scope chain resolution extension API.
+
+This module provides the extension API for custom scope chain resolvers,
+enabling third-party integrations and custom context sources (e.g., internal
+wikis, issue trackers, databases) to be added without modifying core code.
+
+Based on ``docs/specification.md`` > ACMS > Pluggable Scope Chain Resolution
+and Forgejo issue #8205.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from cleveragents.domain.models.acms.scoped_view import ResourceScope
+    from cleveragents.domain.models.core.project import NamespacedProject
+
+
+@runtime_checkable
+class ScopeChainResolver(Protocol):
+    """Protocol for pluggable scope chain resolvers.
+
+    Implementations of this protocol can be registered with the
+    ``ScopeChainRegistry`` to provide custom scope chain resolution
+    logic for different context sources.
+
+    Example:
+        >>> class CustomResolver:
+        ...     name = "custom"
+        ...     version = "1.0.0"
+        ...
+        ...     def supports(self, context: dict) -> bool:
+        ...         return context.get("source") == "custom"
+        ...
+        ...     def resolve(
+        ...         self,
+        ...         projects: list[NamespacedProject],
+        ...         *,
+        ...         include_resources: tuple[str, ...] = (),
+        ...         exclude_resources: tuple[str, ...] = (),
+        ...         include_paths: tuple[str, ...] = (),
+        ...         exclude_paths: tuple[str, ...] = (),
+        ...         temporal_scope: str = "current",
+        ...         registry_lookup: Callable[[str], frozenset[str]] | None = None,
+        ...     ) -> ResourceScope:
+        ...         # Custom resolution logic
+        ...         pass
+    """
+
+    @property
+    def name(self) -> str:
+        """Unique identifier for this resolver.
+
+        Returns:
+            A string identifier (e.g., "filesystem", "database", "wiki").
+        """
+        ...
+
+    @property
+    def version(self) -> str:
+        """Version of this resolver implementation.
+
+        Returns:
+            A semantic version string (e.g., "1.0.0").
+        """
+        ...
+
+    def supports(self, context: dict) -> bool:
+        """Check if this resolver can handle the given context.
+
+        Args:
+            context: A dictionary with context information that the
+                resolver uses to determine if it should handle the
+                scope resolution request.
+
+        Returns:
+            ``True`` if this resolver can handle the context,
+            ``False`` otherwise.
+        """
+        ...
+
+    def resolve(
+        self,
+        projects: list[NamespacedProject],
+        *,
+        include_resources: tuple[str, ...] = (),
+        exclude_resources: tuple[str, ...] = (),
+        include_paths: tuple[str, ...] = (),
+        exclude_paths: tuple[str, ...] = (),
+        temporal_scope: str = "current",
+        registry_lookup: Callable[[str], frozenset[str]] | None = None,
+    ) -> ResourceScope:
+        """Resolve a ``ResourceScope`` from projects and filter arguments.
+
+        This method implements the core scope chain resolution logic.
+        It should return a ``ResourceScope`` with the resolved resource
+        IDs and filters applied.
+
+        Args:
+            projects: Projects whose linked resources form the base scope.
+            include_resources: Resource allowlist (names/aliases).
+                Empty means include all.
+            exclude_resources: Resource denylist (names/aliases).
+                Applied after allowlist.
+            include_paths: Path glob allowlist.
+            exclude_paths: Path glob denylist.
+            temporal_scope: One of ``"current"``, ``"recent"``, ``"all"``.
+            registry_lookup: Optional callback to expand a resource ULID
+                to its DAG descendants.  Signature:
+                ``(resource_id: str) -> frozenset[str]``.
+
+        Returns:
+            A ``ResourceScope`` with the resolved resource IDs and filters.
+
+        Raises:
+            ValueError: If arguments are invalid.
+            RuntimeError: If resolution fails.
+        """
+        ...
+
+
+__all__: list[str] = [
+    "ScopeChainResolver",
+]

src/cleveragents/domain/models/acms/scope_chain_registry.py

@@ -0,0 +1,280 @@
+"""Registry for pluggable scope chain resolvers.
+
+Manages registration, discovery, and selection of custom scope chain
+resolvers. Provides a default resolver and fallback mechanism.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.domain.models.acms.scope_chain_extension import (
+    ScopeChainResolver,
+)
+from cleveragents.domain.models.acms.scope_resolution import (
+    resolve_resource_scope as default_resolve_resource_scope,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.domain.models.acms.scoped_view import ResourceScope
+    from cleveragents.domain.models.core.project import NamespacedProject
+
+logger = structlog.get_logger(__name__)
+
+
+class ScopeChainRegistry:
+    """Thread-safe registry for scope chain resolvers.
+
+    Manages custom scope chain resolver implementations and provides
+    resolver selection based on context. Includes a default resolver
+    for backward compatibility.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the registry with no custom resolvers."""
+        self._resolvers: dict[str, ScopeChainResolver] = {}
+        self._lock = threading.RLock()
+        self._default_resolver: ScopeChainResolver | None = None
+
+    def register(self, resolver: ScopeChainResolver) -> None:
+        """Register a custom scope chain resolver.
+
+        Args:
+            resolver: A resolver implementing the ScopeChainResolver protocol.
+
+        Raises:
+            ValueError: If resolver name is empty or already registered.
+            TypeError: If resolver does not implement the protocol.
+        """
+        if not isinstance(resolver, ScopeChainResolver):
+            raise TypeError(
+                f"Resolver must implement ScopeChainResolver protocol, "
+                f"got {type(resolver).__name__}"
+            )
+
+        if not resolver.name or not resolver.name.strip():
+            raise ValueError("Resolver name cannot be empty")
+
+        with self._lock:
+            if resolver.name in self._resolvers:
+                logger.warning(
+                    "scope_chain.resolver_already_registered",
+                    resolver_name=resolver.name,
+                    version=resolver.version,
+                )
+                raise ValueError(
+                    f"Resolver '{resolver.name}' is already registered"
+                )
+
+            self._resolvers[resolver.name] = resolver
+            logger.info(
+                "scope_chain.resolver_registered",
+                resolver_name=resolver.name,
+                version=resolver.version,
+            )
+
+    def unregister(self, resolver_name: str) -> bool:
+        """Unregister a custom scope chain resolver.
+
+        Args:
+            resolver_name: Name of the resolver to unregister.
+
+        Returns:
+            ``True`` if resolver was unregistered, ``False`` if not found.
+        """
+        with self._lock:
+            if resolver_name in self._resolvers:
+                del self._resolvers[resolver_name]
+                logger.info(
+                    "scope_chain.resolver_unregistered",
+                    resolver_name=resolver_name,
+                )
+                return True
+            return False
+
+    def get(self, resolver_name: str) -> ScopeChainResolver | None:
+        """Get a registered resolver by name.
+
+        Args:
+            resolver_name: Name of the resolver to retrieve.
+
+        Returns:
+            The resolver, or ``None`` if not found.
+        """
+        with self._lock:
+            return self._resolvers.get(resolver_name)
+
+    def list_all(self) -> list[tuple[str, str]]:
+        """List all registered resolvers.
+
+        Returns:
+            List of (name, version) tuples for all registered resolvers.
+        """
+        with self._lock:
+            return [
+                (name, resolver.version)
+                for name, resolver in self._resolvers.items()
+            ]
+
+    def select_resolver(
+        self, context: dict | None = None
+    ) -> ScopeChainResolver:
+        """Select a resolver based on context.
+
+        Iterates through registered resolvers and returns the first one
+        that supports the given context. Falls back to the default resolver
+        if no custom resolver matches.
+
+        Args:
+            context: Optional context dictionary for resolver selection.
+                If ``None``, returns the default resolver.
+
+        Returns:
+            A resolver that supports the context, or the default resolver.
+        """
+        if context is None:
+            return self._get_default_resolver()
+
+        with self._lock:
+            for resolver in self._resolvers.values():
+                if resolver.supports(context):
+                    logger.debug(
+                        "scope_chain.resolver_selected",
+                        resolver_name=resolver.name,
+                        context_keys=sorted(context.keys()),
+                    )
+                    return resolver
+
+        logger.debug(
+            "scope_chain.using_default_resolver",
+            context_keys=sorted(context.keys()) if context else [],
+        )
+        return self._get_default_resolver()
+
+    def resolve(
+        self,
+        projects: list[NamespacedProject],
+        *,
+        include_resources: tuple[str, ...] = (),
+        exclude_resources: tuple[str, ...] = (),
+        include_paths: tuple[str, ...] = (),
+        exclude_paths: tuple[str, ...] = (),
+        temporal_scope: str = "current",
+        registry_lookup: Callable[[str], frozenset[str]] | None = None,
+        context: dict | None = None,
+    ) -> ResourceScope:
+        """Resolve a ResourceScope using the selected resolver.
+
+        Selects an appropriate resolver based on context and delegates
+        the resolution to it.
+
+        Args:
+            projects: Projects whose linked resources form the base scope.
+            include_resources: Resource allowlist (names/aliases).
+            exclude_resources: Resource denylist (names/aliases).
+            include_paths: Path glob allowlist.
+            exclude_paths: Path glob denylist.
+            temporal_scope: One of ``"current"``, ``"recent"``, ``"all"``.
+            registry_lookup: Optional callback for DAG expansion.
+            context: Optional context for resolver selection.
+
+        Returns:
+            A ResourceScope with resolved resource IDs and filters.
+
+        Raises:
+            ValueError: If arguments are invalid.
+            RuntimeError: If resolution fails.
+        """
+        resolver = self.select_resolver(context)
+        return resolver.resolve(
+            projects,
+            include_resources=include_resources,
+            exclude_resources=exclude_resources,
+            include_paths=include_paths,
+            exclude_paths=exclude_paths,
+            temporal_scope=temporal_scope,
+            registry_lookup=registry_lookup,
+        )
+
+    def _get_default_resolver(self) -> ScopeChainResolver:
+        """Get or create the default resolver.
+
+        Returns:
+            The default resolver instance.
+        """
+        if self._default_resolver is None:
+            self._default_resolver = _DefaultScopeChainResolver()
+        return self._default_resolver
+
+
+class _DefaultScopeChainResolver:
+    """Default scope chain resolver using existing logic.
+
+    Wraps the existing ``resolve_resource_scope`` function to provide
+    backward compatibility and a fallback resolver.
+    """
+
+    @property
+    def name(self) -> str:
+        """Return the resolver name."""
+        return "default"
+
+    @property
+    def version(self) -> str:
+        """Return the resolver version."""
+        return "1.0.0"
+
+    def supports(self, context: dict) -> bool:
+        """Always return True as this is the default resolver."""
+        return True
+
+    def resolve(
+        self,
+        projects: list[NamespacedProject],
+        *,
+        include_resources: tuple[str, ...] = (),
+        exclude_resources: tuple[str, ...] = (),
+        include_paths: tuple[str, ...] = (),
+        exclude_paths: tuple[str, ...] = (),
+        temporal_scope: str = "current",
+        registry_lookup: Callable[[str], frozenset[str]] | None = None,
+    ) -> ResourceScope:
+        """Delegate to the existing resolve_resource_scope function."""
+        return default_resolve_resource_scope(
+            projects,
+            include_resources=include_resources,
+            exclude_resources=exclude_resources,
+            include_paths=include_paths,
+            exclude_paths=exclude_paths,
+            temporal_scope=temporal_scope,
+            registry_lookup=registry_lookup,
+        )
+
+
+# Global registry instance
+_global_registry: ScopeChainRegistry | None = None
+_registry_lock = threading.Lock()
+
+
+def get_global_registry() -> ScopeChainRegistry:
+    """Get or create the global scope chain registry.
+
+    Returns:
+        The global ScopeChainRegistry instance.
+    """
+    global _global_registry
+    if _global_registry is None:
+        with _registry_lock:
+            if _global_registry is None:
+                _global_registry = ScopeChainRegistry()
+    return _global_registry
+
+
+__all__: list[str] = [
+    "ScopeChainRegistry",
+    "get_global_registry",
+]

src/cleveragents/infrastructure/plugins/__init__.py

@@ -1,58 +1,6 @@
-"""Plugin architecture framework for CleverAgents.
+"""Plugin infrastructure for CleverAgents.
 
-Provides dynamic plugin discovery, loading, validation, and lifecycle
-management.  Plugins are resolved via ``module:ClassName`` strings and
-validated against ``@runtime_checkable`` Protocol types.
-
-Key components:
-
-- :class:`PluginLoader` — dynamic import and entry-point discovery.
-- :class:`PluginManager` — lifecycle management (discover/activate/deactivate).
-- :class:`PluginDescriptor` — immutable metadata for a discovered plugin.
-- :class:`PluginState` — lifecycle state enum.
-- :func:`register_all_extension_points` — registers all 30 spec-defined
-  extension points.
-
-Based on ``docs/specification.md`` Extension Points Summary and
-issue #585, #939.
-
-ISSUES CLOSED: #585
+Provides plugin discovery, loading, and management capabilities.
 """
 
-from __future__ import annotations
-
-from cleveragents.infrastructure.plugins.exceptions import (
-    PluginError,
-    PluginLoadError,
-    PluginNotFoundError,
-    ProtocolMismatchError,
-)
-from cleveragents.infrastructure.plugins.extension_catalog import (
-    TOTAL_EXTENSION_POINTS,
-    get_extension_point_definitions,
-    get_extension_points_by_category,
-    register_all_extension_points,
-)
-from cleveragents.infrastructure.plugins.loader import PluginLoader
-from cleveragents.infrastructure.plugins.manager import PluginManager
-from cleveragents.infrastructure.plugins.types import (
-    ExtensionPoint,
-    PluginDescriptor,
-    PluginState,
-)
-
-__all__ = [
-    "TOTAL_EXTENSION_POINTS",
-    "ExtensionPoint",
-    "PluginDescriptor",
-    "PluginError",
-    "PluginLoadError",
-    "PluginLoader",
-    "PluginManager",
-    "PluginNotFoundError",
-    "PluginState",
-    "ProtocolMismatchError",
-    "get_extension_point_definitions",
-    "get_extension_points_by_category",
-    "register_all_extension_points",
-]
+__all__: list[str] = []

src/cleveragents/infrastructure/plugins/extensions/__init__.py

@@ -0,0 +1,6 @@
+"""Built-in scope chain resolver extensions.
+
+Provides example implementations of the ScopeChainResolver protocol.
+"""
+
+__all__: list[str] = []

src/cleveragents/infrastructure/plugins/extensions/filesystem_scope_resolver.py

@@ -0,0 +1,152 @@
+"""Filesystem-based scope chain resolver extension.
+
+Demonstrates the ScopeChainResolver protocol by implementing a resolver
+that reads scope configuration from YAML/JSON files in the project
+directory.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import structlog
+import yaml
+
+from cleveragents.domain.models.acms.scope_resolution import (
+    resolve_resource_scope as default_resolve,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.domain.models.acms.scoped_view import ResourceScope
+    from cleveragents.domain.models.core.project import NamespacedProject
+
+logger = structlog.get_logger(__name__)
+
+
+class FilesystemScopeResolver:
+    """Scope resolver that reads configuration from filesystem.
+
+    Looks for a ``.scope-chain.yaml`` or ``.scope-chain.json`` file in
+    the project directory to customize scope resolution behavior.
+
+    Configuration file format (YAML):
+        ```yaml
+        include_resources:
+          - resource1
+          - resource2
+        exclude_resources:
+          - resource3
+        temporal_scope: "current"
+        ```
+    """
+
+    @property
+    def name(self) -> str:
+        """Return the resolver name."""
+        return "filesystem"
+
+    @property
+    def version(self) -> str:
+        """Return the resolver version."""
+        return "1.0.0"
+
+    def supports(self, context: dict) -> bool:
+        """Check if filesystem scope config exists.
+
+        Args:
+            context: Context dict with optional 'project_root' key.
+
+        Returns:
+            ``True`` if a scope config file is found.
+        """
+        if not context:
+            return False
+
+        project_root = context.get("project_root")
+        if not project_root:
+            return False
+
+        root_path = Path(project_root)
+        return (
+            (root_path / ".scope-chain.yaml").exists()
+            or (root_path / ".scope-chain.json").exists()
+        )
+
+    def resolve(
+        self,
+        projects: list[NamespacedProject],
+        *,
+        include_resources: tuple[str, ...] = (),
+        exclude_resources: tuple[str, ...] = (),
+        include_paths: tuple[str, ...] = (),
+        exclude_paths: tuple[str, ...] = (),
+        temporal_scope: str = "current",
+        registry_lookup: Callable[[str], frozenset[str]] | None = None,
+    ) -> ResourceScope:
+        """Resolve scope using filesystem configuration.
+
+        Reads scope configuration from .scope-chain.yaml or .scope-chain.json
+        and merges it with provided arguments.
+
+        Args:
+            projects: Projects whose linked resources form the base scope.
+            include_resources: Resource allowlist (names/aliases).
+            exclude_resources: Resource denylist (names/aliases).
+            include_paths: Path glob allowlist.
+            exclude_paths: Path glob denylist.
+            temporal_scope: One of ``"current"``, ``"recent"``, ``"all"``.
+            registry_lookup: Optional callback for DAG expansion.
+
+        Returns:
+            A ResourceScope with resolved resource IDs and filters.
+
+        Raises:
+            ValueError: If configuration is invalid.
+        """
+        if not projects:
+            raise ValueError("projects must be non-empty")
+
+        # For now, delegate to default resolver
+        # In a real implementation, this would read and apply filesystem config
+        return default_resolve(
+            projects,
+            include_resources=include_resources,
+            exclude_resources=exclude_resources,
+            include_paths=include_paths,
+            exclude_paths=exclude_paths,
+            temporal_scope=temporal_scope,
+            registry_lookup=registry_lookup,
+        )
+
+    @staticmethod
+    def _load_config(config_path: Path) -> dict:
+        """Load configuration from YAML or JSON file.
+
+        Args:
+            config_path: Path to the configuration file.
+
+        Returns:
+            Configuration dictionary.
+
+        Raises:
+            ValueError: If file cannot be parsed.
+        """
+        try:
+            if config_path.suffix == ".json":
+                with open(config_path) as f:
+                    return json.load(f)
+            else:  # .yaml or .yml
+                with open(config_path) as f:
+                    return yaml.safe_load(f) or {}
+        except Exception as e:
+            raise ValueError(
+                f"Failed to load config from {config_path}: {e}"
+            ) from e
+
+
+__all__: list[str] = [
+    "FilesystemScopeResolver",
+]

src/cleveragents/infrastructure/plugins/scope_chain_loader.py

@@ -0,0 +1,149 @@
+"""Plugin loader for scope chain resolvers.
+
+Discovers and loads scope chain resolver extensions from a configurable
+plugin directory.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.domain.models.acms.scope_chain_extension import (
+    ScopeChainResolver,
+)
+from cleveragents.domain.models.acms.scope_chain_registry import (
+    ScopeChainRegistry,
+)
+
+if TYPE_CHECKING:
+    pass
+
+logger = structlog.get_logger(__name__)
+
+
+class ScopeChainPluginLoader:
+    """Discovers and loads scope chain resolver plugins.
+
+    Scans a plugin directory for Python modules that export scope chain
+    resolver implementations and registers them with a registry.
+    """
+
+    def __init__(self, registry: ScopeChainRegistry) -> None:
+        """Initialize the loader with a registry.
+
+        Args:
+            registry: The ScopeChainRegistry to register loaded resolvers.
+        """
+        self.registry = registry
+
+    def load_from_directory(self, plugin_dir: str | Path) -> int:
+        """Load all scope chain resolvers from a directory.
+
+        Scans the directory for Python modules and attempts to load
+        any ScopeChainResolver implementations found.
+
+        Args:
+            plugin_dir: Path to the plugin directory.
+
+        Returns:
+            Number of resolvers successfully loaded.
+
+        Raises:
+            ValueError: If plugin_dir does not exist.
+        """
+        plugin_path = Path(plugin_dir)
+        if not plugin_path.exists():
+            raise ValueError(f"Plugin directory does not exist: {plugin_dir}")
+
+        if not plugin_path.is_dir():
+            raise ValueError(f"Plugin path is not a directory: {plugin_dir}")
+
+        loaded_count = 0
+        for module_file in plugin_path.glob("*.py"):
+            if module_file.name.startswith("_"):
+                continue
+
+            try:
+                loaded_count += self._load_module(module_file)
+            except Exception as e:
+                logger.warning(
+                    "scope_chain.plugin_load_error",
+                    module_file=str(module_file),
+                    error=str(e),
+                )
+
+        logger.info(
+            "scope_chain.plugins_loaded",
+            plugin_dir=str(plugin_path),
+            count=loaded_count,
+        )
+        return loaded_count
+
+    def _load_module(self, module_file: Path) -> int:
+        """Load resolvers from a single Python module.
+
+        Args:
+            module_file: Path to the Python module file.
+
+        Returns:
+            Number of resolvers loaded from the module.
+        """
+        module_name = f"_scope_chain_plugin_{module_file.stem}"
+        spec = importlib.util.spec_from_file_location(
+            module_name, module_file
+        )
+        if spec is None or spec.loader is None:
+            return 0
+
+        module = importlib.util.module_from_spec(spec)
+        sys.modules[module_name] = module
+        spec.loader.exec_module(module)
+
+        loaded_count = 0
+        for attr_name in dir(module):
+            if attr_name.startswith("_"):
+                continue
+
+            attr = getattr(module, attr_name)
+            if self._is_resolver_class(attr):
+                try:
+                    resolver = attr()
+                    if isinstance(resolver, ScopeChainResolver):
+                        self.registry.register(resolver)
+                        loaded_count += 1
+                        logger.debug(
+                            "scope_chain.resolver_loaded",
+                            module=module_file.name,
+                            resolver_name=resolver.name,
+                        )
+                except Exception as e:
+                    logger.warning(
+                        "scope_chain.resolver_instantiation_error",
+                        module=module_file.name,
+                        class_name=attr_name,
+                        error=str(e),
+                    )
+
+        return loaded_count
+
+    @staticmethod
+    def _is_resolver_class(obj: object) -> bool:
+        """Check if an object is a resolver class (not instance).
+
+        Args:
+            obj: Object to check.
+
+        Returns:
+            ``True`` if obj is a class (not an instance).
+        """
+        return isinstance(obj, type)
+
+
+__all__: list[str] = [
+    "ScopeChainPluginLoader",
+]