From 31518aa8efd03b8344f7896f1f4f7d5b0e8cc020 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Sun, 19 Apr 2026 01:19:35 +0000 Subject: [PATCH] feat(context): implement pluggable scope chain resolution extension API --- examples/scope_resolvers/__init__.py | 11 ++ .../scope_resolvers/git_issue_resolver.py | 43 +++++++ src/cleveragents/domain/contexts/__init__.py | 11 +- .../domain/contexts/scope_chain_resolver.py | 115 ++++++++++++++++++ 4 files changed, 179 insertions(+), 1 deletion(-) create mode 100644 examples/scope_resolvers/__init__.py create mode 100644 examples/scope_resolvers/git_issue_resolver.py create mode 100644 src/cleveragents/domain/contexts/scope_chain_resolver.py diff --git a/examples/scope_resolvers/__init__.py b/examples/scope_resolvers/__init__.py new file mode 100644 index 000000000..b81a5b164 --- /dev/null +++ b/examples/scope_resolvers/__init__.py @@ -0,0 +1,11 @@ +"""Example scope chain resolvers.""" + +from examples.scope_resolvers.git_issue_resolver import ( + GitIssueResolver, + create_git_issue_resolver, +) + +__all__ = [ + "GitIssueResolver", + "create_git_issue_resolver", +] diff --git a/examples/scope_resolvers/git_issue_resolver.py b/examples/scope_resolvers/git_issue_resolver.py new file mode 100644 index 000000000..2fc1a8482 --- /dev/null +++ b/examples/scope_resolvers/git_issue_resolver.py @@ -0,0 +1,43 @@ +"""Example scope chain resolver for Git issue references.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from cleveragents.domain.contexts import ( + ScopeChainResolver, + ScopeResolutionContext, + ) + + +class GitIssueResolver: + """Example resolver for Git issue scope references.""" + + def __init__(self, repo_path: str | None = None) -> None: + """Initialize the resolver.""" + self.repo_path = repo_path or "." + + def resolve( + self, + scope: str, + context: ScopeResolutionContext, + ) -> list[str]: + """Resolve a Git issue scope reference.""" + if not scope.startswith("issue:"): + return [] + + try: + issue_id = scope[6:] + if not issue_id: + return [] + + fragment_id = f"git_issue_{issue_id}" + return [fragment_id] + except Exception: + return [] + + +def create_git_issue_resolver() -> GitIssueResolver: + """Factory function for creating a GitIssueResolver instance.""" + return GitIssueResolver() diff --git a/src/cleveragents/domain/contexts/__init__.py b/src/cleveragents/domain/contexts/__init__.py index 2c9f19546..201300af0 100644 --- a/src/cleveragents/domain/contexts/__init__.py +++ b/src/cleveragents/domain/contexts/__init__.py @@ -1,11 +1,20 @@ """Contexts domain module. Contains pipeline-specific domain models for the ACMS context assembly -pipeline, including ``ScoredFragment`` for scored/ranked fragments. +pipeline, including ``ScoredFragment`` for scored/ranked fragments and +pluggable scope chain resolution extension API. """ from cleveragents.domain.contexts.fragment import ScoredFragment +from cleveragents.domain.contexts.scope_chain_resolver import ( + ScopeChainResolver, + ScopeResolutionContext, + ScopeResolverRegistry, +) __all__ = [ "ScoredFragment", + "ScopeChainResolver", + "ScopeResolutionContext", + "ScopeResolverRegistry", ] diff --git a/src/cleveragents/domain/contexts/scope_chain_resolver.py b/src/cleveragents/domain/contexts/scope_chain_resolver.py new file mode 100644 index 000000000..ca1a45a27 --- /dev/null +++ b/src/cleveragents/domain/contexts/scope_chain_resolver.py @@ -0,0 +1,115 @@ +"""Pluggable scope chain resolution extension API.""" + +from __future__ import annotations + +from typing import Any, Protocol + +from pydantic import BaseModel, Field + + +class ScopeResolutionContext(BaseModel): + """Context passed to scope resolvers during resolution.""" + + scope: str = Field( + ..., + description="The scope reference to resolve (e.g., 'issue:123').", + ) + + metadata: dict[str, Any] = Field( + default_factory=dict, + description="Additional metadata for resolution (e.g., project context).", + ) + + resolved_fragments: list[str] = Field( + default_factory=list, + description="List of already-resolved fragment identifiers.", + ) + + +class ScopeChainResolver(Protocol): + """Protocol for custom scope chain resolvers.""" + + def resolve( + self, + scope: str, + context: ScopeResolutionContext, + ) -> list[str]: + """Resolve a scope reference to a list of fragment identifiers.""" + ... + + +class ScopeResolverRegistry: + """Registry for discovering and managing scope chain resolvers.""" + + def __init__(self) -> None: + """Initialize the registry and discover resolvers from entry points.""" + self._resolvers: dict[str, tuple[ScopeChainResolver, int]] = {} + self._discover_resolvers() + + def _discover_resolvers(self) -> None: + """Discover resolvers from Python entry points.""" + try: + import importlib.metadata as metadata + except ImportError: + import importlib_metadata as metadata # type: ignore + + try: + entry_points = metadata.entry_points() + if hasattr(entry_points, "select"): + scope_resolvers = entry_points.select( + group="cleveragents.scope_resolvers" + ) + else: + scope_resolvers = entry_points.get("cleveragents.scope_resolvers", []) + + for ep in scope_resolvers: + try: + resolver_factory = ep.load() + resolver = resolver_factory() + priority = getattr(ep, "priority", 0) + self._resolvers[ep.name] = (resolver, priority) + except Exception: + pass + except Exception: + pass + + def register( + self, + name: str, + resolver: ScopeChainResolver, + priority: int = 0, + ) -> None: + """Register a resolver with the given name and priority.""" + self._resolvers[name] = (resolver, priority) + + def unregister(self, name: str) -> None: + """Unregister a resolver by name.""" + self._resolvers.pop(name, None) + + def resolve( + self, + scope: str, + context: ScopeResolutionContext, + ) -> list[str]: + """Resolve a scope using registered resolvers in priority order.""" + sorted_resolvers = sorted( + self._resolvers.values(), + key=lambda x: x[1], + reverse=True, + ) + + for resolver, _ in sorted_resolvers: + result = resolver.resolve(scope, context) + if result: + return result + + return [] + + def get_resolvers(self) -> dict[str, tuple[ScopeChainResolver, int]]: + """Get all registered resolvers with their priorities.""" + return dict(self._resolvers) + + def list_resolvers(self) -> list[tuple[str, int]]: + """List all registered resolvers with their priorities.""" + items = [(name, priority) for name, (_, priority) in self._resolvers.items()] + return sorted(items, key=lambda x: x[1], reverse=True)