feat(context): implement pluggable scope chain resolution extension API #10658

Open
HAL9000 wants to merge 7 commits from feat/v360/pluggable-scope-chain-api into master
8 changed files with 643 additions and 1 deletions
+11
View File
@@ -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`
+1
View File
@@ -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.
+11
View File
@@ -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 []
+10 -1
View File
@@ -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)