Compare commits

...

1 Commits

Author SHA1 Message Date
freemo 2235fece22 feat(resource): wire inheritance polymorphism into tool binding and auto-discovery
CI / build (pull_request) Successful in 22s
CI / helm (pull_request) Successful in 22s
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 3m20s
CI / quality (pull_request) Successful in 3m53s
CI / typecheck (pull_request) Successful in 3m58s
CI / security (pull_request) Successful in 4m8s
CI / integration_tests (pull_request) Successful in 9m14s
CI / unit_tests (pull_request) Successful in 9m21s
CI / docker (pull_request) Successful in 1m18s
CI / coverage (pull_request) Failing after 8m52s
CI / e2e_tests (pull_request) Successful in 22m12s
CI / status-check (pull_request) Failing after 1s
CI / benchmark-regression (pull_request) Successful in 55m0s
Refactor tool binding to use polymorphic resource type matching via
is_subtype_of() instead of exact string comparison or static parent_types
lists.  A tool declared for resource type 'code' now automatically binds
to 'code/python', 'code/rust', etc.

Resource auto-discovery gains discover_most_specific_type() to resolve
the deepest subtype in the hierarchy, and get_type_tree() enables the
new 'resource type tree' CLI command to visualise the full inheritance
hierarchy.

Plugins can now register custom resource types at runtime via
register_plugin_resource_type() on the registry service and
register_resource_types() on the plugin manager, extending the hierarchy
without code changes.

Resource metadata output includes the full inheritance chain for
debugging (inheritance_chain field in JSON/YAML output).

Key changes:
- BindingResolutionService._is_type_compatible(): delegates to
  is_subtype_of() for polymorphic matching (ADR-042)
- ResourceRegistryService: add discover_most_specific_type(),
  get_type_tree(), register_plugin_resource_type(), and
  get_resource_inheritance_metadata()
- PluginManager: add register_resource_types() extension point
- CLI: add 'resource type tree' command and inheritance_chain in
  resource metadata output
- 14 BDD scenarios covering polymorphic binding, auto-discovery,
  tree hierarchy, plugin registration, and metadata chain

ISSUES CLOSED: #942
2026-03-30 22:17:00 +00:00
8 changed files with 870 additions and 4 deletions
@@ -0,0 +1,97 @@
Feature: Wire inheritance polymorphism into tool binding and auto-discovery
As a developer
I want resource type inheritance wired into tool binding and auto-discovery
So that polymorphic matching works across the resource system
# Polymorphic tool binding
Scenario: Tool bound to parent type matches child resource type
Given a registry with type hierarchy "code" -> "code/python" for wiring
And a binding service using that registry for wiring
When I check type compatibility of "code/python" against required "code" for wiring
Then the types should be compatible for wiring
Scenario: Tool bound to parent type matches grandchild resource type
Given a registry with type hierarchy "file" -> "code" -> "code/python" for wiring
And a binding service using that registry for wiring
When I check type compatibility of "code/python" against required "file" for wiring
Then the types should be compatible for wiring
Scenario: Tool bound to child type does not match parent resource type
Given a registry with type hierarchy "code" -> "code/python" for wiring
And a binding service using that registry for wiring
When I check type compatibility of "code" against required "code/python" for wiring
Then the types should NOT be compatible for wiring
Scenario: Exact type match still works
Given a registry with type hierarchy "code" -> "code/python" for wiring
And a binding service using that registry for wiring
When I check type compatibility of "code/python" against required "code/python" for wiring
Then the types should be compatible for wiring
Scenario: Unrelated types are not compatible
Given a registry with type hierarchy "code" -> "code/python" for wiring
And a binding service using that registry for wiring
When I check type compatibility of "code/python" against required "data" for wiring
Then the types should NOT be compatible for wiring
# ── Resource auto-discovery: most specific type ───────────────────────────
Scenario: discover_most_specific_type picks deepest subtype
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
When I call discover_most_specific_type with candidates "file", "code", "code/python" for wiring
Then the most specific type should be "code/python" for wiring
Scenario: discover_most_specific_type with single candidate
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
When I call discover_most_specific_type with candidates "code" for wiring
Then the most specific type should be "code" for wiring
Scenario: discover_most_specific_type with unrelated types keeps first
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
When I call discover_most_specific_type with candidates "code/python", "data" for wiring
Then the most specific type should be "code/python" for wiring
Scenario: discover_most_specific_type rejects empty list
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
When I call discover_most_specific_type with empty candidates for wiring
Then a ValueError should be raised for wiring
# ── Type hierarchy tree ───────────────────────────────────────────────────
Scenario: get_type_tree returns correct hierarchy structure
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
When I call get_type_tree for wiring
Then the tree should have "file" as a root type for wiring
And the tree should show "code" as a child of "file" for wiring
And the tree should show "code/python" as a child of "code" for wiring
# ── Plugin resource type registration ─────────────────────────────────────
Scenario: Plugin registers a custom resource type extending the hierarchy
Given a registry service with hierarchy "file" -> "code" for wiring
When I register a plugin resource type "acme/code-rust" inheriting "code" for wiring
Then the type "acme/code-rust" should be registered for wiring
And "acme/code-rust" should be a subtype of "code" for wiring
Scenario: Plugin registration rejects type with missing parent
Given a registry service with hierarchy "file" -> "code" for wiring
When I register a plugin resource type "acme/orphan" inheriting "nonexistent" for wiring
Then a ValidationError should be raised for plugin registration for wiring
# ── Inheritance chain in resource metadata ────────────────────────────────
Scenario: Resource metadata includes inheritance chain
Given a registry service with hierarchy "file" -> "code" -> "code/python" for wiring
And a resource "test/my-py" of type "code/python" for wiring
When I get the resource inheritance metadata for "test/my-py" for wiring
Then the inheritance chain should be "code/python", "code", "file" for wiring
# ── Plugin manager resource type extension point ──────────────────────────
Scenario: Plugin manager registers resource types from activated plugin
Given a registry service with hierarchy "file" -> "code" for wiring
And a plugin manager for wiring
And a mock plugin "test-plugin" with resource types for wiring
When I call register_resource_types on the plugin manager for wiring
Then the plugin's resource types should be registered for wiring
@@ -106,6 +106,14 @@ def step_mock_registry(context: Context) -> None:
registry.show_resource = show_resource
registry.show_type = show_type
# Remove auto-generated MagicMock attributes that interfere with
# BindingResolutionService._is_type_compatible() — it checks
# hasattr(registry, "is_subtype_of") and MagicMock auto-creates
# any attribute, returning a truthy mock that bypasses the
# parent_types fallback path. Delete these so the service falls
# through to the show_type + parent_types code path.
del registry.is_subtype_of
del registry._load_type_registry
context.mock_registry = registry
@@ -0,0 +1,412 @@
"""Steps for resource_inheritance_wiring_942.feature — polymorphic tool binding and auto-discovery."""
from __future__ import annotations
from unittest.mock import MagicMock
from behave import given, then, when # type: ignore[import-untyped]
from behave.runner import Context # type: ignore[import-untyped]
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.application.services._resource_registry_data import spec_to_db
from cleveragents.application.services.binding_resolution_service import (
BindingResolutionService,
)
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.core.exceptions import ValidationError
from cleveragents.domain.models.core.resource_type import ResourceTypeSpec
from cleveragents.infrastructure.database.models import (
Base,
ResourceModel,
ResourceTypeModel,
)
from cleveragents.infrastructure.plugins.manager import PluginManager
from cleveragents.infrastructure.plugins.types import (
PluginDescriptor,
PluginState,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
# Prefix all test type names to avoid collisions with built-in types
# that the service bootstraps automatically.
_PFX = "w942-"
def _make_service() -> ResourceRegistryService:
"""Create a ResourceRegistryService backed by an in-memory SQLite DB."""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine)
return ResourceRegistryService(session_factory=factory)
def _register_type(
svc: ResourceRegistryService,
name: str,
*,
inherits: str | None = None,
) -> None:
"""Register a minimal test type using spec_to_db for correct column mapping."""
session = svc._session()
try:
existing = session.query(ResourceTypeModel).filter_by(name=name).first()
if existing is not None:
return
spec = ResourceTypeSpec.from_config(
{
"name": name,
"description": f"Test type {name}",
"resource_kind": "physical",
"sandbox_strategy": "none",
"user_addable": True,
"built_in": True,
"inherits": inherits,
}
)
session.add(spec_to_db(spec, source="test"))
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def _register_resource(
svc: ResourceRegistryService,
name: str,
type_name: str,
) -> None:
"""Register a minimal resource instance."""
from datetime import UTC, datetime
from ulid import ULID
session = svc._session()
try:
now_iso = datetime.now(tz=UTC).isoformat()
row = ResourceModel(
resource_id=str(ULID()),
namespaced_name=name,
namespace=name.split("/", 1)[0] if "/" in name else None,
type_name=type_name,
resource_kind="physical",
location=None,
description=None,
read_only=False,
auto_discovered=False,
sandbox_strategy=None,
content_hash=None,
properties_json=None,
metadata_json=None,
created_at=now_iso,
updated_at=now_iso,
)
session.add(row)
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
# ---------------------------------------------------------------------------
# Given steps — type hierarchy setup
# ---------------------------------------------------------------------------
@given('a registry with type hierarchy "code" -> "code/python" for wiring')
def step_registry_code_python(context: Context) -> None:
svc = _make_service()
_register_type(svc, f"{_PFX}code")
_register_type(svc, f"{_PFX}code-python", inherits=f"{_PFX}code")
_register_type(svc, f"{_PFX}data")
context.wiring_svc = svc
# Alias mapping for use in When/Then steps
context.wiring_type_map = {
"code": f"{_PFX}code",
"code/python": f"{_PFX}code-python",
"data": f"{_PFX}data",
}
@given('a registry with type hierarchy "file" -> "code" -> "code/python" for wiring')
def step_registry_file_code_python(context: Context) -> None:
svc = _make_service()
_register_type(svc, f"{_PFX}file")
_register_type(svc, f"{_PFX}code", inherits=f"{_PFX}file")
_register_type(svc, f"{_PFX}code-python", inherits=f"{_PFX}code")
_register_type(svc, f"{_PFX}data")
context.wiring_svc = svc
context.wiring_type_map = {
"file": f"{_PFX}file",
"code": f"{_PFX}code",
"code/python": f"{_PFX}code-python",
"data": f"{_PFX}data",
}
@given("a binding service using that registry for wiring")
def step_binding_service(context: Context) -> None:
context.wiring_binding_svc = BindingResolutionService(
resource_registry=context.wiring_svc,
)
@given('a registry service with hierarchy "file" -> "code" -> "code/python" for wiring')
def step_registry_service_hierarchy(context: Context) -> None:
svc = _make_service()
_register_type(svc, f"{_PFX}file")
_register_type(svc, f"{_PFX}code", inherits=f"{_PFX}file")
_register_type(svc, f"{_PFX}code-python", inherits=f"{_PFX}code")
_register_type(svc, f"{_PFX}data")
context.wiring_svc = svc
context.wiring_type_map = {
"file": f"{_PFX}file",
"code": f"{_PFX}code",
"code/python": f"{_PFX}code-python",
"data": f"{_PFX}data",
}
@given('a registry service with hierarchy "file" -> "code" for wiring')
def step_registry_service_file_code(context: Context) -> None:
svc = _make_service()
_register_type(svc, f"{_PFX}file")
_register_type(svc, f"{_PFX}code", inherits=f"{_PFX}file")
context.wiring_svc = svc
context.wiring_type_map = {
"file": f"{_PFX}file",
"code": f"{_PFX}code",
}
@given('a resource "test/my-py" of type "code/python" for wiring')
def step_register_resource(context: Context) -> None:
real_name = context.wiring_type_map["code/python"]
_register_resource(context.wiring_svc, "test/my-py", real_name)
@given("a plugin manager for wiring")
def step_plugin_manager(context: Context) -> None:
context.wiring_pm = PluginManager()
@given('a mock plugin "test-plugin" with resource types for wiring')
def step_mock_plugin(context: Context) -> None:
real_code = context.wiring_type_map["code"]
descriptor = PluginDescriptor(
name="test-plugin",
module_path="cleveragents.test_mock",
class_name="MockPlugin",
)
pm: PluginManager = context.wiring_pm
with pm._lock:
pm._plugins["test-plugin"] = descriptor
descriptor.state = PluginState.ACTIVATED
mock_instance = MagicMock()
mock_instance.resource_types = [
{
"name": "acme/w942-test-type",
"description": "Plugin test type",
"resource_kind": "physical",
"sandbox_strategy": "none",
"user_addable": True,
"inherits": real_code,
},
]
pm._instances["test-plugin"] = mock_instance
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
def _resolve_name(context: Context, name: str) -> str:
"""Resolve a friendly name to its prefixed test name."""
tmap = getattr(context, "wiring_type_map", {})
return tmap.get(name, name)
@when(
'I check type compatibility of "{actual}" against required "{required}" for wiring'
)
def step_check_compatibility(
context: Context,
actual: str,
required: str,
) -> None:
binding_svc: BindingResolutionService = context.wiring_binding_svc
real_actual = _resolve_name(context, actual)
real_required = _resolve_name(context, required)
context.wiring_compatible = binding_svc._is_type_compatible(
real_required, real_actual
)
@when("I call discover_most_specific_type with candidates {candidates_str} for wiring")
def step_discover_most_specific(
context: Context,
candidates_str: str,
) -> None:
raw = [c.strip().strip('"') for c in candidates_str.split(",")]
candidates = [_resolve_name(context, c) for c in raw]
svc: ResourceRegistryService = context.wiring_svc
result = svc.discover_most_specific_type(candidates)
# Store the friendly name for assertion
reverse_map = {v: k for k, v in context.wiring_type_map.items()}
context.wiring_result = reverse_map.get(result, result)
@when("I call discover_most_specific_type with empty candidates for wiring")
def step_discover_most_specific_empty(context: Context) -> None:
svc: ResourceRegistryService = context.wiring_svc
context.wiring_error = None
try:
svc.discover_most_specific_type([])
except ValueError as exc:
context.wiring_error = exc
@when("I call get_type_tree for wiring")
def step_get_type_tree(context: Context) -> None:
svc: ResourceRegistryService = context.wiring_svc
context.wiring_tree = svc.get_type_tree()
@when('I register a plugin resource type "{name}" inheriting "{parent}" for wiring')
def step_register_plugin_type(
context: Context,
name: str,
parent: str,
) -> None:
svc: ResourceRegistryService = context.wiring_svc
real_parent = _resolve_name(context, parent)
context.wiring_error = None
context.wiring_result = None
try:
spec = svc.register_plugin_resource_type(
{
"name": name,
"description": f"Plugin type {name}",
"resource_kind": "physical",
"sandbox_strategy": "none",
"user_addable": True,
"inherits": real_parent,
}
)
context.wiring_result = spec
except (ValidationError, ValueError) as exc:
context.wiring_error = exc
@when('I get the resource inheritance metadata for "{name}" for wiring')
def step_get_inheritance_metadata(context: Context, name: str) -> None:
svc: ResourceRegistryService = context.wiring_svc
context.wiring_meta = svc.get_resource_inheritance_metadata(name)
@when("I call register_resource_types on the plugin manager for wiring")
def step_register_resource_types_pm(context: Context) -> None:
pm: PluginManager = context.wiring_pm
svc: ResourceRegistryService = context.wiring_svc
context.wiring_registered_types = pm.register_resource_types(svc, "test-plugin")
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the types should be compatible for wiring")
def step_types_compatible(context: Context) -> None:
assert context.wiring_compatible is True, "Expected types to be compatible"
@then("the types should NOT be compatible for wiring")
def step_types_not_compatible(context: Context) -> None:
assert context.wiring_compatible is False, "Expected types to NOT be compatible"
@then('the most specific type should be "{expected}" for wiring')
def step_most_specific(context: Context, expected: str) -> None:
assert context.wiring_result == expected, (
f"Expected {expected}, got {context.wiring_result}"
)
@then("a ValueError should be raised for wiring")
def step_value_error(context: Context) -> None:
assert context.wiring_error is not None, "Expected ValueError"
assert isinstance(context.wiring_error, ValueError)
@then('the tree should have "{name}" as a root type for wiring')
def step_tree_root(context: Context, name: str) -> None:
real_name = _resolve_name(context, name)
tree = context.wiring_tree
assert real_name in tree.get("(root)", []), (
f"Expected {real_name} in root types, got {tree.get('(root)', [])}"
)
@then('the tree should show "{child}" as a child of "{parent}" for wiring')
def step_tree_child_of_parent(
context: Context,
child: str,
parent: str,
) -> None:
real_child = _resolve_name(context, child)
real_parent = _resolve_name(context, parent)
tree = context.wiring_tree
assert real_child in tree.get(real_parent, []), (
f"Expected {real_child} under {real_parent}, got {tree.get(real_parent, [])}"
)
@then('the type "{name}" should be registered for wiring')
def step_type_registered(context: Context, name: str) -> None:
svc: ResourceRegistryService = context.wiring_svc
spec = svc.show_type(name)
assert spec is not None, f"Type {name} should be registered"
@then('"{child}" should be a subtype of "{parent}" for wiring')
def step_is_subtype(context: Context, child: str, parent: str) -> None:
real_parent = _resolve_name(context, parent)
svc: ResourceRegistryService = context.wiring_svc
assert svc.is_subtype_of(child, real_parent), (
f"{child} should be a subtype of {real_parent}"
)
@then("a ValidationError should be raised for plugin registration for wiring")
def step_validation_error(context: Context) -> None:
assert context.wiring_error is not None, "Expected ValidationError"
assert isinstance(context.wiring_error, (ValidationError, ValueError))
@then("the inheritance chain should be {chain_str} for wiring")
def step_inheritance_chain(context: Context, chain_str: str) -> None:
friendly = [c.strip().strip('"') for c in chain_str.split(",")]
expected = [_resolve_name(context, c) for c in friendly]
meta = context.wiring_meta
assert meta["inheritance_chain"] == expected, (
f"Expected chain {expected}, got {meta['inheritance_chain']}"
)
@then("the plugin's resource types should be registered for wiring")
def step_plugin_types_registered(context: Context) -> None:
registered = context.wiring_registered_types
assert len(registered) > 0, "Expected at least one type registered"
assert "acme/w942-test-type" in registered
@@ -49,6 +49,7 @@ from cleveragents.domain.models.core.tool import (
ResourceSlot,
Tool,
)
from cleveragents.resource.inheritance import is_subtype_of
logger = logging.getLogger(__name__)
@@ -355,13 +356,27 @@ class BindingResolutionService:
) -> bool:
"""Check if *actual_type* satisfies *required_type*.
Exact match always passes. If *actual_type* declares
*required_type* in its ``parent_types`` hierarchy, the
check also passes (sub-type relationship).
Uses the inheritance hierarchy via ``is_subtype_of()`` for
polymorphic matching (ADR-042 §Tool Binding Polymorphism).
A tool bound to ``code`` automatically matches ``code/python``,
``code/javascript``, etc.
"""
if actual_type == required_type:
return True
# Use the registry service's is_subtype_of when available
# (delegates to the full inheritance hierarchy).
if hasattr(self._registry, "is_subtype_of"):
return self._registry.is_subtype_of(actual_type, required_type)
# Fallback: use inheritance module directly with a loaded
# type registry when the registry service exposes
# _load_type_registry (e.g. in tests with mocks).
if hasattr(self._registry, "_load_type_registry"):
type_reg = self._registry._load_type_registry()
return is_subtype_of(actual_type, required_type, type_reg)
# Last resort: check the static parent_types list on the spec.
try:
spec: ResourceTypeSpec = self._registry.show_type(
actual_type,
@@ -458,6 +458,184 @@ class ResourceRegistryService(ResourceInstanceMixin, ResourceDagMixin):
registry = self._load_type_registry()
return is_subtype_of(type_name, ancestor_name, registry)
def discover_most_specific_type(
self,
candidate_types: list[str],
) -> str:
"""Resolve the most specific type from a list of candidates.
Given a set of candidate type names (e.g. from MIME type or
extension mapping), returns the one that is the most specific
(deepest) in the inheritance hierarchy. If two candidates are
unrelated, the first one in the list wins.
This enables auto-discovery to resolve ``.py`` files to
``code/python`` instead of just ``file`` or ``code``.
Args:
candidate_types: List of candidate type names, most generic
first.
Returns:
The most specific (deepest) type name.
Raises:
ValueError: If *candidate_types* is empty.
"""
if not candidate_types:
raise ValueError("candidate_types must not be empty")
if len(candidate_types) == 1:
return candidate_types[0]
registry = self._load_type_registry()
best = candidate_types[0]
for candidate in candidate_types[1:]:
if candidate not in registry:
continue
# If candidate is a subtype of the current best, it's more
# specific. Replace best.
if is_subtype_of(candidate, best, registry):
best = candidate
# If best is a subtype of candidate, keep best (it's already
# more specific). Otherwise the two are unrelated — keep
# whichever appeared first.
return best
def get_type_tree(self) -> dict[str, list[str]]:
"""Build a tree representation of the type hierarchy.
Returns a dict mapping each type name to its list of direct
child type names. Root types (no parent) are children of a
synthetic ``"(root)"`` key.
Returns:
Dict mapping parent type name to list of child type names.
"""
registry = self._load_type_registry()
tree: dict[str, list[str]] = {"(root)": []}
for name, entry in registry.items():
parent = entry.get("inherits")
if parent is None:
tree["(root)"].append(name)
else:
tree.setdefault(parent, []).append(name)
# Sort children for deterministic output
for children in tree.values():
children.sort()
return tree
def register_plugin_resource_type(
self,
type_config: dict[str, Any],
) -> ResourceTypeSpec:
"""Register a custom resource type from a plugin configuration dict.
This is the extension point for plugins to add resource types to
the hierarchy at runtime. The config dict has the same schema as
a YAML resource type configuration file.
Args:
type_config: Dict with resource type configuration (name,
description, resource_kind, sandbox_strategy, inherits,
etc.).
Returns:
The validated ``ResourceTypeSpec``.
Raises:
ValidationError: If the config is invalid or chain rules
are violated.
"""
spec = ResourceTypeSpec.from_config(type_config)
session = self._session()
try:
existing = (
session.query(ResourceTypeModel).filter_by(name=spec.name).first()
)
if existing is not None:
raise ValidationError(
message=f"Resource type '{spec.name}' already exists",
details={"name": spec.name},
)
if spec.inherits is not None:
registry = self._load_type_registry()
try:
validate_chain(
spec.name,
spec.inherits,
registry,
is_built_in=spec.built_in,
)
except (
ResourceTypeParentNotFoundError,
ValueError,
) as exc:
raise ValidationError(
message=str(exc),
details={"name": spec.name, "inherits": spec.inherits},
) from exc
db_model = spec_to_db(spec, source="plugin")
session.add(db_model)
session.flush()
session.commit()
except ValidationError:
session.rollback()
raise
except Exception:
session.rollback()
raise
finally:
session.close()
return spec
def get_resource_inheritance_metadata(
self,
resource_name_or_id: str,
) -> dict[str, Any]:
"""Return inheritance metadata for a resource instance.
Resolves the resource, then returns its type name, the full
inheritance chain, and whether each ancestor is built-in.
Args:
resource_name_or_id: Resource name or ULID.
Returns:
Dict with ``type_name``, ``inheritance_chain``, and
``chain_details`` keys.
Raises:
NotFoundError: If the resource is not found.
"""
resource = self.show_resource(resource_name_or_id)
registry = self._load_type_registry()
chain = resolve_inheritance_chain(resource.resource_type_name, registry)
chain_details: list[dict[str, Any]] = []
for type_name in chain:
entry = registry.get(type_name, {})
chain_details.append(
{
"name": type_name,
"inherits": entry.get("inherits"),
"built_in": entry.get("built_in", False),
}
)
return {
"type_name": resource.resource_type_name,
"inheritance_chain": chain,
"chain_details": chain_details,
}
def _load_type_registry(self) -> dict[str, dict[str, Any]]:
"""Load all registered types into an in-memory dict for chain ops.
+71 -1
View File
@@ -146,7 +146,7 @@ def _resource_type_dict(spec: Any) -> dict[str, object]:
def _resource_dict(resource: Any) -> dict[str, object]:
"""Convert a Resource domain object to a plain dict for output formatting."""
return {
result: dict[str, object] = {
"resource_id": resource.resource_id,
"name": resource.name,
"type": resource.resource_type_name,
@@ -162,6 +162,19 @@ def _resource_dict(resource: Any) -> dict[str, object]:
else str(resource.updated_at),
}
# Include inheritance chain metadata for debugging
try:
service = _get_registry_service()
meta = service.get_resource_inheritance_metadata(
resource.resource_id,
)
result["inheritance_chain"] = meta["inheritance_chain"]
except Exception:
# Gracefully degrade — inheritance metadata is optional
result["inheritance_chain"] = [resource.resource_type_name]
return result
# ---------------------------------------------------------------------------
# Resource Type commands
@@ -351,6 +364,63 @@ def type_list(
raise typer.Abort() from exc
@type_app.command("tree")
def type_tree(
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Display the resource type inheritance hierarchy as a tree.
Shows all registered resource types organized by their inheritance
relationships. Root types (no parent) appear at the top level, with
child types indented below their parents.
Examples:
agents resource type tree
agents resource type tree --format json
"""
try:
service = _get_registry_service()
tree = service.get_type_tree()
if fmt != OutputFormat.RICH.value:
console.print(format_output(tree, fmt))
return
def _print_type_tree_node(
name: str,
tree_data: dict[str, list[str]],
indent: int = 0,
) -> None:
prefix = " " * indent
if indent == 0:
console.print(f"{prefix}[bold cyan]{name}[/bold cyan]")
else:
console.print(f"{prefix}[cyan]{name}[/cyan]")
for child in tree_data.get(name, []):
_print_type_tree_node(child, tree_data, indent + 1)
roots = tree.get("(root)", [])
if not roots:
console.print("[yellow]No resource types registered.[/yellow]")
return
console.print("[bold]Resource Type Hierarchy[/bold]\n")
for root in roots:
_print_type_tree_node(root, tree)
except CleverAgentsError as exc:
console.print(f"[red]Error:[/red] {exc.message}")
raise typer.Abort() from exc
except Exception as exc:
if isinstance(exc, (typer.Abort, typer.Exit)):
raise
console.print(f"[red]Unexpected error:[/red] {exc}")
raise typer.Abort() from exc
@type_app.command("show")
def type_show(
name: Annotated[
@@ -389,6 +389,84 @@ class PluginManager:
registered.append(result)
return registered
# ------------------------------------------------------------------
# Resource type extension point
# ------------------------------------------------------------------
def register_resource_types(
self,
resource_registry: Any,
plugin_name: str,
) -> list[str]:
"""Register resource types declared by an activated plugin.
Inspects the plugin instance for a ``resource_types`` attribute
(list of dicts with resource type configuration). Each type is
registered through ``ResourceRegistryService.register_plugin_resource_type()``.
Args:
resource_registry: A ``ResourceRegistryService`` instance
(or compatible object with ``register_plugin_resource_type``).
plugin_name: Name of the activated plugin whose types to
register.
Returns:
List of newly registered type names.
Raises:
PluginNotFoundError: If the plugin is not registered.
PluginError: If the plugin is not activated.
"""
with self._lock:
descriptor = self.get_plugin(plugin_name)
if descriptor.state != PluginState.ACTIVATED:
msg = (
f"Plugin '{plugin_name}' must be activated before "
"resource type registration"
)
raise PluginError(msg)
instance = self._instances.get(plugin_name)
if instance is None:
return []
type_configs: list[dict[str, Any]] = getattr(instance, "resource_types", [])
if not type_configs:
return []
registered: list[str] = []
register_fn = getattr(
resource_registry,
"register_plugin_resource_type",
None,
)
if not callable(register_fn):
self._logger.warning(
"plugin_manager.resource_registry_missing_register_fn",
plugin=plugin_name,
)
return []
for cfg in type_configs:
try:
spec: Any = register_fn(cfg)
type_name: str = spec.name
registered.append(type_name)
self._logger.info(
"plugin_manager.resource_type_registered",
plugin=plugin_name,
type_name=type_name,
)
except Exception as exc:
self._logger.warning(
"plugin_manager.resource_type_registration_failed",
plugin=plugin_name,
config=cfg,
error=str(exc),
)
return registered
# ------------------------------------------------------------------
# Cleanup
# ------------------------------------------------------------------
+8
View File
@@ -1182,6 +1182,14 @@ workspace_folder # noqa: B018, F821
execute_tool # noqa: B018, F821
wrap_service_method # noqa: B018, F821
# Resource inheritance wiring — public API (#942)
type_tree # noqa: B018, F821
discover_most_specific_type # noqa: B018, F821
get_type_tree # noqa: B018, F821
register_plugin_resource_type # noqa: B018, F821
get_resource_inheritance_metadata # noqa: B018, F821
register_resource_types # noqa: B018, F821
# LSP functional runtime — public API (#826)
StdioTransport # noqa: B018, F821
LspClient # noqa: B018, F821