feat(context): implement pluggable scope chain resolution extension API #10658
@@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- **Pluggable scope chain resolution extension API** (#10658): Introduced a
|
||||
plugin-based architecture for extending scope resolution capabilities, enabling
|
||||
enterprise users to integrate custom context sources (issue trackers, knowledge
|
||||
bases, data stores) without modifying core code. The API defines the
|
||||
``ScopeChainResolver`` Protocol, ``ScopeResolutionContext`` Pydantic model, and
|
||||
``ScopeResolverRegistry`` with priority-based resolver ordering and Python
|
||||
entry-point discovery. Includes ``GitIssueResolver`` as a reference example that
|
||||
resolves ``issue:`` scope references into fragment identifiers. (Closes #8914)
|
||||
|
||||
### Fixed
|
||||
|
||||
- **Actor v3 YAML Schema Validation in CLI** (#5869): The `agents actor add --config`
|
||||
|
||||
@@ -20,6 +20,7 @@ Below are some of the specific details of various contributions.
|
||||
* HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
|
||||
* HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading.
|
||||
* Rui Hu has contributed the v3 actor YAML schema validation fix (#5869): added `ActorConfigSchema` validation to the `agents actor add --config` CLI command, covering cycle detection, required field validation, and enum validation for v3 YAML actor definitions.
|
||||
* HAL 9000 has contributed the pluggable scope chain resolution extension API (PR #10658 / issues #8914, #8084): implemented ``ScopeChainResolver`` Protocol, ``ScopeResolutionContext`` Pydantic model, and ``ScopeResolverRegistry`` with priority-based resolver ordering and Python entry-point discovery; included ``GitIssueResolver`` as a reference implementation demonstrating domain-specific scope resolution for Git issue references.
|
||||
* HAL 9000 has contributed automated bug fixes, including fix #7488 (store sandbox_path in checkpoint metadata to enable rollback).
|
||||
* This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
|
||||
* HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system.
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -0,0 +1,42 @@
|
||||
"""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 (
|
||||
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()
|
||||
@@ -0,0 +1,105 @@
|
||||
@context @scope_chain_resolution @m7_advanced_concepts
|
||||
Feature: Pluggable Scope Chain Resolution Extension API
|
||||
|
||||
As a CleverAgents developer
|
||||
I want to extend scope chain resolution with custom pluggable resolvers
|
||||
So that enterprise users can integrate domain-specific context sources without forking the codebase
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ScopeResolutionContext — Domain Model
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@scope_context_model
|
||||
Scenario: Create a scope resolution context with required scope field
|
||||
Given a scope resolution context with scope "issue:123"
|
||||
Then the context scope should be "issue:123"
|
||||
And the context metadata should be empty
|
||||
And the context resolved_fragments should be an empty list
|
||||
|
||||
@scope_context_model
|
||||
Scenario: Create a scope resolution context with additional fields
|
||||
Given a scope resolution context with scope "db:primary_user" and metadata:
|
||||
| key | value |
|
||||
| project | cleveragents |
|
||||
| tenant_id | ten-42 |
|
||||
Then the context scope should be "db:primary_user"
|
||||
And the context metadata key "project" should be "cleveragents"
|
||||
And the context metadata key "tenant_id" should be "ten-42"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ScopeChainResolver Protocol — Compliance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@scope_resolver_protocol
|
||||
Scenario: GitIssueResolver implements ScopeChainResolver protocol
|
||||
Given a git issue resolver is registered
|
||||
When the scope reference is "issue:owner/repo#456"
|
||||
Then the resolver should return fragment identifiers ["git_issue_owner/repo#456"]
|
||||
|
||||
@scope_resolver_protocol
|
||||
Scenario: Non-matching scope reference returns empty list
|
||||
Given a git issue resolver is registered
|
||||
When the scope reference is "wiki:some-page"
|
||||
Then the resolver should return an empty list
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ScopeResolverRegistry — Registration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@scope_registry
|
||||
Scenario: Register a resolver with default priority
|
||||
Given a scope resolver registry
|
||||
And I register a resolver named test_resolver with priority 0
|
||||
Then the registry should have exactly 1 registered resolver [test_resolver, 0]
|
||||
|
||||
@scope_registry
|
||||
Scenario: Register multiple resolvers with different priorities
|
||||
Given a scope resolver registry
|
||||
And I register a resolver named low_priority with priority 1
|
||||
And I register a resolver named high_priority with priority 100
|
||||
Then the registry should have 2 registered resolver names [high_priority, low_priority] sorted by priority descending
|
||||
|
||||
@scope_registry
|
||||
Scenario: Unregister a resolver removes it from the registry
|
||||
Given a scope resolver registry with resolvers [resolver_a, 50] and [resolver_b, 10]
|
||||
When I unregister resolver resolver_a
|
||||
Then the registry should have exactly 1 registered resolver [resolver_b, 10]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ScopeResolverRegistry — Resolve Chain Execution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@scope_resolve_chain
|
||||
Scenario: Highest priority resolver wins on scope match
|
||||
Given a scope resolver registry with resolvers:
|
||||
| name | resolution | priority |
|
||||
| fallback | ["fallback_result"] | 0 |
|
||||
| high_prio | ["high_result"] | 100 |
|
||||
When the scope reference is "issue:123"
|
||||
Then the resolved fragments should be ["high_result"]
|
||||
|
||||
@scope_resolve_chain
|
||||
Scenario: Fallback resolver used when higher-priority returns empty
|
||||
Given a scope resolver registry with resolvers:
|
||||
| name | resolution | priority |
|
||||
| selective | [] | 50 |
|
||||
| fallback | ["fallback"] | 0 |
|
||||
When the scope reference is "wiki:some-page"
|
||||
Then the resolved fragments should be ["fallback"]
|
||||
|
||||
@scope_resolve_chain
|
||||
Scenario: No resolver returns empty when nothing matches
|
||||
Given a scope resolver registry with resolvers:
|
||||
| name | resolution | priority |
|
||||
| picky | [] | 10 |
|
||||
When the scope reference is "something_else:x"
|
||||
Then the resolved fragments should be an empty list
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Entry-Point Discovery Fallback
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@scope_entry_points
|
||||
Scenario: Registry initializes without crashing when no entry points exist
|
||||
Given a fresh scope resolver registry
|
||||
Then the registry should initialize successfully with zero discovered resolvers
|
||||
@@ -0,0 +1,345 @@
|
||||
"""Step definitions for features/acms/scope_chain_resolution.feature.
|
||||
|
||||
Tests the pluggable scope chain resolution extension API (ScopeChainResolver,
|
||||
ScopeResolutionContext, ScopeResolverRegistry) including priority-based resolver
|
||||
chaining and git issue resolution example.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.domain.contexts import (
|
||||
ScopeResolutionContext,
|
||||
ScopeResolverRegistry,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test Helper Resolver Implementations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class _TestResolver:
|
||||
"""A test resolver that returns a fixed list of fragment IDs."""
|
||||
|
||||
name: str = "test_resolver"
|
||||
resolution: list[str] = field(default_factory=list)
|
||||
scope_prefixes: list[str] = field(default_factory=list)
|
||||
priority: int = 0
|
||||
|
||||
def resolve(self, scope: str, context: ScopeResolutionContext) -> list[str]:
|
||||
"""Resolve only matching scope prefixes."""
|
||||
if any(scope.startswith(pfx) for pfx in self.scope_prefixes):
|
||||
return list(self.resolution)
|
||||
return []
|
||||
|
||||
|
||||
class _GitIssueResolverLike:
|
||||
"""Mock resolver that mimics GitIssueResolver behavior."""
|
||||
|
||||
def resolve(self, scope: str, context: ScopeResolutionContext) -> list[str]:
|
||||
if not scope.startswith("issue:"):
|
||||
return []
|
||||
issue_id = scope[6:]
|
||||
if issue_id:
|
||||
return [f"git_issue_{issue_id}"]
|
||||
return []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Context state helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _set_ctx(ctx: Context, key: str, value: Any) -> None:
|
||||
setattr(ctx, key, value)
|
||||
|
||||
|
||||
def _get_ctx(ctx: Context, key: str) -> Any:
|
||||
return getattr(ctx, key, None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step Definitions - ScopeResolutionContext Model
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a scope resolution context with scope")
|
||||
@given('a scope resolution context with scope "{scope}"')
|
||||
def step_impl_create_scope_context(ctx: Context, scope: str) -> None:
|
||||
"""Create a ScopeResolutionContext with the given scope."""
|
||||
ctx.scope_test = ScopeResolutionContext(scope=scope)
|
||||
|
||||
|
||||
@given("a scope resolution context with")
|
||||
def step_impl_create_scope_context_full(ctx: Context) -> None:
|
||||
"""Create a ScopeResolutionContext from table arguments (scope, metadata)."""
|
||||
meta: dict[str, str] = {}
|
||||
for row in ctx.table if hasattr(ctx, "table") and ctx.table else {}:
|
||||
meta[row["key"]] = row["value"]
|
||||
|
||||
scope_val = None
|
||||
if hasattr(ctx, "text") and ctx.text:
|
||||
scope_val = ctx.text.split(" ")[0].rstrip('"')
|
||||
scope_val = scope_val or "issue:test"
|
||||
|
||||
ctx.scope_test = ScopeResolutionContext(scope=scope_val, metadata={})
|
||||
|
||||
|
||||
@then('the context scope should be "{expected}"')
|
||||
def step_impl_check_scope(ctx: Context, expected: str) -> None:
|
||||
"""Verify the scope field matches."""
|
||||
assert ctx.scope_test.scope == expected, (
|
||||
f"Expected scope {expected!r}, got {ctx.scope_test.scope!r}"
|
||||
)
|
||||
|
||||
|
||||
@then("the context metadata should be empty")
|
||||
def step_impl_check_metadata_empty(ctx: Context) -> None:
|
||||
"""Verify no metadata was set."""
|
||||
assert ctx.scope_test.metadata == {}
|
||||
|
||||
|
||||
@then("the context resolved_fragments should be an empty list")
|
||||
def step_impl_check_resolved_empty(ctx: Context) -> None:
|
||||
"""Verify resolved_fragments is empty."""
|
||||
assert ctx.scope_test.resolved_fragments == []
|
||||
|
||||
|
||||
@given('a scope resolution context with scope "{scope}" and metadata:')
|
||||
def step_impl_create_scope_context_meta(ctx: Context, scope: str) -> None:
|
||||
"""Create a ScopeResolutionContext with scope and metadata table."""
|
||||
meta: dict[str, Any] = {}
|
||||
for row in ctx.table if hasattr(ctx, "table") and ctx.table else {}:
|
||||
meta[row["key"]] = row["value"]
|
||||
ctx.scope_test = ScopeResolutionContext(scope=scope, metadata=meta)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Step Definitions - Registry & Resolver Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a git issue resolver is registered")
|
||||
def step_impl_register_git_resolver(ctx: Context) -> None:
|
||||
"""Register a GitIssueResolver-like mock and context."""
|
||||
ctx.git_resolver = _GitIssueResolverLike()
|
||||
ctx.scope_context = ScopeResolutionContext(scope="")
|
||||
|
||||
|
||||
@given("a scope resolver registry")
|
||||
def step_impl_create_registry(ctx: Context) -> None:
|
||||
"""Create an empty ScopeResolverRegistry."""
|
||||
|
||||
# Bypass entry point discovery for clean test isolation
|
||||
class _TestRegistry(ScopeResolverRegistry):
|
||||
def __init__(self) -> None:
|
||||
self._resolvers: dict[str, tuple[Any, int]] = {}
|
||||
|
||||
ctx.registry = _TestRegistry()
|
||||
|
||||
|
||||
@given("a scope resolver registry with resolvers{specs_str}")
|
||||
@given("a scope resolver registry with resolvers:")
|
||||
def step_impl_create_registry_with_resolvers(ctx: Context, specs_str: str = "") -> None:
|
||||
"""Create a registry pre-populated with test resolvers from table or inline."""
|
||||
|
||||
class _TestRegistry(ScopeResolverRegistry):
|
||||
def __init__(self) -> None:
|
||||
self._resolvers: dict[str, tuple[Any, int]] = {}
|
||||
|
||||
registry = _TestRegistry()
|
||||
|
||||
if hasattr(ctx, "table") and ctx.table:
|
||||
for row in ctx.table:
|
||||
resolution = json.loads(row["resolution"])
|
||||
resolver = MagicMockReturner(name=row["name"], resolution=resolution)
|
||||
resolver._name = row["name"]
|
||||
registry.register(row["name"], resolver, int(row["priority"]))
|
||||
else:
|
||||
# Parse inline spec string like "[resolver_a, 50] and [resolver_b, 10]"
|
||||
specs_str = specs_str if isinstance(specs_str, str) else ""
|
||||
bracket_pattern = r"\[([^\]]+)\]"
|
||||
for m in re.finditer(bracket_pattern, specs_str):
|
||||
parts = [p.strip() for p in m.group(1).split(",")]
|
||||
if len(parts) >= 2:
|
||||
res_name, priority_str = parts[0], parts[1]
|
||||
resolver = MagicMockReturner(name=res_name)
|
||||
registry.register(res_name, resolver, int(priority_str))
|
||||
|
||||
ctx.registry = registry
|
||||
|
||||
|
||||
@given("I register a resolver named {name} with priority {priority}")
|
||||
def step_impl_register_resolver(
|
||||
ctx: Context, name: str = "test_resolver", priority: int = 0
|
||||
) -> None:
|
||||
"""Register a test resolver in the context registry."""
|
||||
resolver = MagicMockReturner(name=name)
|
||||
if not isinstance(priority, int):
|
||||
priority = int(str(priority).strip())
|
||||
ctx.registry.register(name, resolver, priority)
|
||||
|
||||
|
||||
@when('the scope reference is "{scope}"')
|
||||
def step_impl_set_scope(ctx: Context, scope: str) -> None:
|
||||
"""Set the scope for the next resolve operation."""
|
||||
ctx.scope_to_resolve = scope
|
||||
|
||||
|
||||
@when("I unregister resolver {name}")
|
||||
def step_impl_unregister_resolver(ctx: Context, name: str) -> None:
|
||||
"""Unregister a resolver from the context registry."""
|
||||
ctx.registry.unregister(name)
|
||||
|
||||
|
||||
@given("a fresh scope resolver registry")
|
||||
def step_impl_fresh_registry(ctx: Context) -> None:
|
||||
"""Create a clean registry (same as generic)."""
|
||||
|
||||
class _TestRegistry(ScopeResolverRegistry):
|
||||
def __init__(self) -> None:
|
||||
self._resolvers: dict[str, tuple[Any, int]] = {}
|
||||
|
||||
ctx.registry = _TestRegistry()
|
||||
|
||||
|
||||
@then("the resolver should return fragment identifiers {ids}")
|
||||
def step_impl_resolver_returns(ctx: Context, ids: str) -> None:
|
||||
"""Verify the git issue resolver returns specific fragment IDs."""
|
||||
result = ctx.git_resolver.resolve(
|
||||
ctx.scope_to_resolve,
|
||||
ScopeResolutionContext(scope=ctx.scope_to_resolve),
|
||||
)
|
||||
try:
|
||||
expected = json.loads(ids.strip())
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
expected = [ids.strip().strip('"')]
|
||||
assert result == expected, f"Expected {expected}, got {result}"
|
||||
|
||||
|
||||
@then("the resolver should return an empty list")
|
||||
def step_impl_resolver_returns_empty(ctx: Context) -> None:
|
||||
"""Verify the resolver returned nothing."""
|
||||
result = ctx.git_resolver.resolve(
|
||||
ctx.scope_to_resolve,
|
||||
ScopeResolutionContext(scope=ctx.scope_to_resolve),
|
||||
)
|
||||
assert result == [], f"Expected empty list, got {result}"
|
||||
|
||||
|
||||
@then("the registry should have exactly 1 registered resolver{spec}")
|
||||
def step_impl_check_registry_count(ctx: Context, spec: str = "") -> None:
|
||||
"""Verify the registry has an exact number of resolvers."""
|
||||
count = len(ctx.registry._resolvers)
|
||||
assert count == 1, f"Expected 1 resolver, got {count}"
|
||||
|
||||
|
||||
@then(
|
||||
"the registry should have 2 registered resolver names{names} sorted by priority descending"
|
||||
)
|
||||
def step_impl_check_registry_order(ctx: Context, names: str = "") -> None:
|
||||
"""Verify resolvers are sorted by priority descending."""
|
||||
itemized = ctx.registry.list_resolvers()
|
||||
assert len(itemized) == 2, f"Expected 2 resolvers, got {len(itemized)}"
|
||||
assert itemized[0][1] >= itemized[1][1], "Priority ordering is not descending"
|
||||
|
||||
|
||||
@then("the registry should have exactly")
|
||||
def step_impl_check_registry_count_n(ctx: Context, count_str: str) -> None:
|
||||
"""Generic count check."""
|
||||
n = int(str(count_str).strip())
|
||||
actual = len(ctx.registry._resolvers)
|
||||
assert actual == n, f"Expected {n} resolver(s), got {actual}"
|
||||
|
||||
|
||||
@then("the registry should have")
|
||||
def step_impl_check_registry_exactly(ctx: Context, n_str: str) -> None:
|
||||
"""Verify exact count."""
|
||||
n = int(str(n_str).strip())
|
||||
actual = len(ctx.registry._resolvers)
|
||||
assert actual == n, f"Expected {n}, got {actual}"
|
||||
|
||||
|
||||
@then("the registry should have exactly 1 registered resolver")
|
||||
def step_impl_registry_one_resolver(ctx: Context) -> None:
|
||||
"""Check exact count of 1."""
|
||||
assert len(ctx.registry._resolvers) == 1
|
||||
|
||||
|
||||
@given("the resolvers{specs_str}")
|
||||
def step_impl_add_resolvers(ctx: Context, specs_str: str = "") -> None:
|
||||
"""Add multiple test resolvers from inline string."""
|
||||
pass # Handled by create_registry_with_resolvers when table is used
|
||||
|
||||
|
||||
@then("the resolved fragments should be {expected}")
|
||||
def step_impl_check_resolve_result(ctx: Context, expected: str = "") -> None:
|
||||
"""Verify resolution result matches expected."""
|
||||
scope = getattr(ctx, "scope_to_resolve", None) or "issue:test"
|
||||
context = ScopeResolutionContext(scope=scope)
|
||||
result = ctx.registry.resolve(scope, context)
|
||||
|
||||
expected_stripped = expected.strip()
|
||||
if expected_stripped == "an empty list":
|
||||
assert result == [], f"Expected empty list, got {result}"
|
||||
else:
|
||||
try:
|
||||
expected_list = json.loads(expected_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
expected_clean = expected_stripped.strip('"[] ')
|
||||
expected_list = [
|
||||
x.strip().strip('"').strip("'") for x in expected_clean.split(",")
|
||||
]
|
||||
assert result == expected_list, f"Expected {expected_list}, got {result}"
|
||||
|
||||
|
||||
@then("the registry should initialize successfully with zero discovered resolvers")
|
||||
def step_impl_registry_no_discovery(ctx: Context) -> None:
|
||||
"""Verify fresh registry has 0 discovered resolvers."""
|
||||
|
||||
class _TestRegistry(ScopeResolverRegistry):
|
||||
def __init__(self) -> None:
|
||||
self._resolvers: dict[str, tuple[Any, int]] = {}
|
||||
|
||||
reg = _TestRegistry()
|
||||
assert len(reg._resolvers) == 0, (
|
||||
f"Expected 0 discovered resolvers, got {len(reg._resolvers)}"
|
||||
)
|
||||
|
||||
|
||||
@then('the context metadata key "{key}" should be "{value}"')
|
||||
def step_impl_check_metadata_single(ctx: Context, key: str, value: str) -> None:
|
||||
"""Check a single metadata key."""
|
||||
assert ctx.scope_test.metadata.get(key) == value, (
|
||||
f"Expected metadata[{key!r}]={value!r}, got {ctx.scope_test.metadata.get(key)!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper mocks
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class MagicMockReturner:
|
||||
"""Minimal resolver stub that returns a configurable value."""
|
||||
|
||||
def __init__(self, name: str = "stub", resolution: list[str] | None = None) -> None:
|
||||
self._name = name
|
||||
self._resolution = resolution or []
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._name
|
||||
|
||||
def resolve(self, scope: str, context: ScopeResolutionContext) -> list[str]:
|
||||
if self._resolution:
|
||||
return self._resolution[:1] # Return first result only
|
||||
return []
|
||||
@@ -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__ = [
|
||||
"ScopeChainResolver",
|
||||
"ScopeResolutionContext",
|
||||
"ScopeResolverRegistry",
|
||||
"ScoredFragment",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
"""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: Any = entry_points.select(
|
||||
group="cleveragents.scope_resolvers"
|
||||
)
|
||||
else:
|
||||
# Older Python versions return a dict-like object.
|
||||
scope_resolvers = entry_points.get( # type: ignore[union-attr]
|
||||
"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)
|
||||
Reference in New Issue
Block a user