Discover analytic functions and steps from installed packages (ISSUE-172)

PYEGERIA_ISSUES.md

@@ -172,6 +172,7 @@ re-test candidates moved into *Awaiting re-test* below the open sections.
   create-from-template) and 7 (`viewElementGUID`) are still open.
 
 **Open pyegeria items**
+- ISSUE-172: Installed packages couldn't register analytic functions or steps by installing; added the `pyegeria.analytics` entry-point group, read lazily
 - ISSUE-171: `exec_report_spec` returned an analytic result that was not measured as ordinary data or `empty`; it now returns the function's declared absence sentence
 - ISSUE-170: `AnalyticFunctionSpec` can't say what its results mean (unit, as-of, basis, absence sentences); added an optional `declaration`
 - ISSUE-169: `_resolve_analytic_function` imported any dotted path a report spec named; now registered names plus an allowlist of module prefixes
@@ -196,6 +197,26 @@ into the entry now, so it isn't rediscovered from scratch later.
 
 ---
 
+### ISSUE-172: Installed packages couldn't register analytic functions or steps by installing; added the `pyegeria.analytics` entry-point group, read lazily
+
+**Layer:** pyegeria · **Status:** in PR (branch `feat/analytic-entry-points`) · **Found:** 2026-10-30, Resource Explorer's portal-extensibility evaluation (gap G3) and the Architect's `DESIGN-ANALYTIC-LIBRARY-SPLIT-2026-10-30` (LS1: "entry-point group name and manifest loader" go to pyegeria).
+
+**What:** a package could add analytic functions to pyegeria's registry only through `PYEGERIA_ANALYTIC_FUNCTIONS_JSON`/`_MODULES` or a runtime `register_analytic_functions` call. trellis-analytics has its own `trellis.analytics` entry-point group and pushes into the registry by hand (`discovery.register()`), so the Portal sees only what something has remembered to register.
+
+**Change** (`pyegeria/view/analytic_registry.py`):
+- **Group:** `ANALYTIC_ENTRY_POINT_GROUP = "pyegeria.analytics"`, with `trellis.analytics` read as an alias during the split. An entry point names an `AnalyticFunctionSpec` (or a subclass such as trellis's manifests), a list of them, or a callable returning either. That covers both routines and steps (`kind`).
+- **Lazy loading (manifests eager, code lazy, per the design note §3):** nothing is read at import. Entry points are read on the first `get_analytic_registry()` call and cached. `refresh_analytic_functions()` forgets the cache. Only the manifests load then; a function's own module is imported on its first run by the resolver.
+- **Failures:** a broken entry point is recorded with a sentence naming it and the error. The rest still load, and the failures are logged at WARNING and returned by `analytic_entry_point_errors()` (also in `analytic_registry_payload()["entry_point_errors"]`), so "installed, not loadable" never reads as "nothing installed". An installed name that clashes with a built-in or config entry is skipped and reported, never fatal. The same manifest under both groups is not a clash.
+- **Checked with trellis's venv** (2026-10-30): both installed entry points (`context_pack`, `curation_coverage`) are discovered with no errors, and trellis's `discovery.register()` then adds nothing, harmlessly.
+
+**With ISSUE-169:** an entry-point-registered function is trusted by the resolver whatever its module, so installing is the registration path, and the allowlist only governs unregistered dotted paths.
+
+**Not done here:** the design note's split of manifest from runner inside one step's entry point, failure words on first run, and Admin "reload steps" are library/RE-host work (LS3). Whether the Portal replaces `PYEGERIA_REPORT_SPEC_MODULES` with an entry-point group is Mandy's question 2 in that note.
+
+Tests: `tests/micro-tests/test_issue_172_analytic_entry_points.py`.
+
+---
+
 ### ISSUE-171: `exec_report_spec` returned an analytic result that was not measured as ordinary data or `empty`; it now returns the function's declared absence sentence
 
 **Layer:** pyegeria · **Status:** in PR (branch `feat/state-aware-report-spec`, stacked on ISSUE-170's `feat/analytic-declaration`) · **Found:** 2026-10-30, Resource Explorer's portal-extensibility evaluation, §3 item 2.

pyegeria/view/analytic_registry.py

@@ -38,8 +38,9 @@
 from __future__ import annotations
 
 import os
+from importlib.metadata import entry_points
 from pathlib import Path
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union
 
 from loguru import logger
 from pydantic import BaseModel, Field, field_validator, model_validator
@@ -57,6 +58,9 @@
     'KNOWN_ANALYTIC_SOURCES',
     'AnalyticRegistryCollision',
     'refresh_analytic_functions',
+    'ANALYTIC_ENTRY_POINT_GROUP',
+    'ANALYTIC_ENTRY_POINT_ALIASES',
+    'analytic_entry_point_errors',
     'get_analytic_registry',
     'analytic_registry_payload',
     'register_analytic_functions',
@@ -622,6 +626,17 @@ def _strict() -> AnalyticParam:
 _RUNTIME_ANALYTIC_FUNCTIONS = AnalyticFunctionDict()
 _CONFIG_ANALYTIC_FUNCTIONS = AnalyticFunctionDict()
 
+# Installed packages register analytic functions and steps through this entry-point
+# group (ISSUE-172). trellis-analytics' group is read as an alias during the library
+# split. An entry point names an AnalyticFunctionSpec (or subclass), a list of them, or
+# a callable returning either. Read lazily on first use of the registry; only the
+# manifests are loaded then -- a function's own module is imported on its first run,
+# by format_set_executor._resolve_analytic_function.
+ANALYTIC_ENTRY_POINT_GROUP = "pyegeria.analytics"
+ANALYTIC_ENTRY_POINT_ALIASES = ("trellis.analytics",)
+_ENTRY_POINT_ANALYTIC_FUNCTIONS: Optional[AnalyticFunctionDict] = None  # None = not read yet
+_ENTRY_POINT_ERRORS: Dict[str, str] = {}
+
 
 class AnalyticRegistryCollision(ValueError):
     """Raised when two analytic-registry sources define the same function name."""
@@ -652,9 +667,11 @@ def refresh_analytic_functions() -> None:
       - PYEGERIA_ANALYTIC_FUNCTIONS_MODULES: comma-separated module callables
         (pkg.mod:func or pkg.mod.func) returning a dict/AnalyticFunctionDict
     Collisions across sources (including against BUILTINS) raise AnalyticRegistryCollision.
+    Also forgets the installed (entry-point) tier, so it is re-read on next use.
     """
-    global _CONFIG_ANALYTIC_FUNCTIONS
+    global _CONFIG_ANALYTIC_FUNCTIONS, _ENTRY_POINT_ANALYTIC_FUNCTIONS
     _CONFIG_ANALYTIC_FUNCTIONS = AnalyticFunctionDict()
+    _ENTRY_POINT_ANALYTIC_FUNCTIONS = None  # installed packages are re-read on next use
 
     json_paths = os.getenv("PYEGERIA_ANALYTIC_FUNCTIONS_JSON", "").strip()
     if json_paths:
@@ -683,12 +700,65 @@ def refresh_analytic_functions() -> None:
             _add_with_collision_check(_CONFIG_ANALYTIC_FUNCTIONS, loaded, source=f"MODULE:{m}")
 
 
+def _read_entry_points() -> Tuple[AnalyticFunctionDict, Dict[str, str]]:
+    """Load every manifest in the entry-point groups. A broken entry point is recorded
+    in the returned errors, naming it, and the rest still load."""
+    found = AnalyticFunctionDict()
+    errors: Dict[str, str] = {}
+    for group in (ANALYTIC_ENTRY_POINT_GROUP, *ANALYTIC_ENTRY_POINT_ALIASES):
+        for ep in entry_points(group=group):
+            label = f"{group}:{ep.name}"
+            try:
+                obj = ep.load()
+                if callable(obj) and not isinstance(obj, AnalyticFunctionSpec):
+                    obj = obj()
+                items = obj if isinstance(obj, (list, tuple)) else [obj]
+                for spec in items:
+                    if not isinstance(spec, AnalyticFunctionSpec):
+                        raise TypeError(f"{type(spec).__name__} is not an AnalyticFunctionSpec")
+                    prior = found.get(spec.name)
+                    if prior is not None and prior.function != spec.function:
+                        raise AnalyticRegistryCollision(
+                            f"'{spec.name}' is already provided by another entry point as {prior.function}")
+                    found[spec.name] = spec
+            except Exception as exc:  # noqa: BLE001 -- one broken package must not hide the others
+                errors[label] = f"entry point {label} ({ep.value}) could not be loaded: {exc}"
+                logger.warning(errors[label])
+    return found, errors
+
+
+def _entry_point_functions() -> AnalyticFunctionDict:
+    global _ENTRY_POINT_ANALYTIC_FUNCTIONS, _ENTRY_POINT_ERRORS
+    if _ENTRY_POINT_ANALYTIC_FUNCTIONS is None:
+        _ENTRY_POINT_ANALYTIC_FUNCTIONS, _ENTRY_POINT_ERRORS = _read_entry_points()
+    return _ENTRY_POINT_ANALYTIC_FUNCTIONS
+
+
+def analytic_entry_point_errors() -> Dict[str, str]:
+    """Entry points that are installed but could not be loaded, each with a sentence
+    naming it and the error (empty when every one loaded, or none is installed)."""
+    _entry_point_functions()
+    return dict(_ENTRY_POINT_ERRORS)
+
+
 def get_analytic_registry() -> AnalyticFunctionDict:
-    """Combine built-ins, config-loaded, and runtime-registered analytic functions.
-    Enforce no duplicate names across all sources."""
+    """Combine built-ins, config-loaded, installed (entry-point) and runtime-registered
+    analytic functions. Enforce no duplicate names across all sources, except that an
+    installed function whose name is already a built-in or config entry is skipped and
+    reported by analytic_entry_point_errors(), so an installed package can't break the
+    registry for everyone."""
     combined = AnalyticFunctionDict()
     _add_with_collision_check(combined, _BUILTINS, source="BUILTINS")
     _add_with_collision_check(combined, _CONFIG_ANALYTIC_FUNCTIONS, source="CONFIG")
+    for name, spec in _entry_point_functions().items():
+        if name in combined:
+            label = f"entry point function '{name}'"
+            if label not in _ENTRY_POINT_ERRORS:
+                _ENTRY_POINT_ERRORS[label] = (f"{label} ({spec.function}) was skipped: the name is already "
+                                              f"defined by a built-in or config entry")
+                logger.warning(_ENTRY_POINT_ERRORS[label])
+            continue
+        combined[name] = spec
     _add_with_collision_check(combined, _RUNTIME_ANALYTIC_FUNCTIONS, source="RUNTIME")
     return combined
 
@@ -700,6 +770,7 @@ def analytic_registry_payload() -> dict:
     return {
         "total": len(registry),
         "functions": {name: spec.dict() for name, spec in registry.items()},
+        "entry_point_errors": analytic_entry_point_errors(),
     }
 
 

tests/micro-tests/test_issue_172_analytic_entry_points.py

@@ -0,0 +1,121 @@
+"""ISSUE-172: installed packages register analytic functions and steps through the
+`pyegeria.analytics` entry-point group (trellis.analytics as an alias), read lazily
+on first use of the registry, with a broken entry point reported, not fatal."""
+import sys
+
+import pytest
+
+import pyegeria.view.analytic_registry as reg
+from pyegeria.view.analytic_registry import AnalyticFunctionSpec
+
+
+class FakeEntryPoint:
+    def __init__(self, name, value, obj=None, error=None):
+        self.name, self.value, self._obj, self._error = name, value, obj, error
+        self.loads = 0
+
+    def load(self):
+        self.loads += 1
+        if self._error:
+            raise self._error
+        return self._obj
+
+
+ROUTINE = AnalyticFunctionSpec(name="ep-routine", function="ep_pkg.routines.coverage")
+STEP = AnalyticFunctionSpec(name="ep-step", function="ep_pkg.steps.staleness", kind="step")
+
+
+@pytest.fixture
+def installed(monkeypatch):
+    """Install fake entry points per group; returns the dict to fill and a call counter."""
+    groups = {reg.ANALYTIC_ENTRY_POINT_GROUP: [], "trellis.analytics": []}
+    calls = []
+
+    def fake_entry_points(group):
+        calls.append(group)
+        return list(groups.get(group, []))
+
+    monkeypatch.setattr(reg, "entry_points", fake_entry_points)
+    monkeypatch.setattr(reg, "_ENTRY_POINT_ANALYTIC_FUNCTIONS", None)
+    monkeypatch.setattr(reg, "_ENTRY_POINT_ERRORS", {})
+    reg.clear_runtime_analytic_functions()
+    yield groups, calls
+    reg.clear_runtime_analytic_functions()
+
+
+def test_functions_and_steps_from_both_groups_are_registered(installed):
+    groups, _ = installed
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE))
+    groups["trellis.analytics"].append(FakeEntryPoint("two", "ep_pkg:steps", lambda: [STEP]))
+    registry = reg.get_analytic_registry()
+    assert registry["ep-routine"].function == "ep_pkg.routines.coverage"
+    assert registry["ep-step"].kind == "step"
+    assert reg.analytic_entry_point_errors() == {}
+
+
+def test_discovery_is_lazy_and_read_once(installed):
+    groups, calls = installed
+    ep = FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE)
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(ep)
+    assert calls == []  # nothing read until the registry is used
+    reg.get_analytic_registry()
+    reg.get_analytic_registry()
+    assert ep.loads == 1
+    reg.refresh_analytic_functions()
+    reg.get_analytic_registry()
+    assert ep.loads == 2  # refresh re-reads installed packages
+
+
+def test_function_module_is_not_imported_by_discovery(installed):
+    groups, _ = installed
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE))
+    reg.get_analytic_registry()
+    assert "ep_pkg.routines" not in sys.modules
+
+
+def test_broken_entry_point_is_reported_and_the_rest_still_load(installed):
+    groups, _ = installed
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP] += [
+        FakeEntryPoint("broken", "bad_pkg:manifest", error=ModuleNotFoundError("No module named 'xxhash'")),
+        FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE),
+    ]
+    assert "ep-routine" in reg.get_analytic_registry()
+    errors = reg.analytic_entry_point_errors()
+    assert list(errors) == [f"{reg.ANALYTIC_ENTRY_POINT_GROUP}:broken"]
+    assert "No module named 'xxhash'" in errors[f"{reg.ANALYTIC_ENTRY_POINT_GROUP}:broken"]
+    assert reg.analytic_registry_payload()["entry_point_errors"] == errors
+
+
+def test_an_entry_point_naming_something_else_is_an_error(installed):
+    groups, _ = installed
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(FakeEntryPoint("odd", "ep_pkg:thing", {"name": "x"}))
+    reg.get_analytic_registry()
+    assert "is not an AnalyticFunctionSpec" in reg.analytic_entry_point_errors()[
+        f"{reg.ANALYTIC_ENTRY_POINT_GROUP}:odd"]
+
+
+def test_installed_name_clashing_with_a_builtin_is_skipped_not_fatal(installed):
+    groups, _ = installed
+    builtin = next(iter(reg._BUILTINS))
+    clash = AnalyticFunctionSpec(name=builtin, function="ep_pkg.other.func")
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(FakeEntryPoint("clash", "ep_pkg:clash", clash))
+    registry = reg.get_analytic_registry()
+    assert registry[builtin].function == reg._BUILTINS[builtin].function
+    assert any("was skipped" in e for e in reg.analytic_entry_point_errors().values())
+
+
+def test_same_manifest_under_both_groups_is_not_a_clash(installed):
+    groups, _ = installed
+    groups[reg.ANALYTIC_ENTRY_POINT_GROUP].append(FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE))
+    groups["trellis.analytics"].append(FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE))
+    assert "ep-routine" in reg.get_analytic_registry()
+    assert reg.analytic_entry_point_errors() == {}
+
+
+def test_runtime_register_skipping_existing_names_still_works(installed):
+    """trellis-analytics' register() checks get_analytic_registry() and skips names present."""
+    groups, _ = installed
+    groups["trellis.analytics"].append(FakeEntryPoint("one", "ep_pkg:ROUTINE", ROUTINE))
+    existing = set(reg.get_analytic_registry())
+    new = {n: m for n, m in {"ep-routine": ROUTINE}.items() if n not in existing}
+    assert new == {}