ARCHIVED: This technical design was archived on 2025-12-22 when the reflection module was removed from the codebase. The DI module now uses Python standard library introspection (inspect, typing.get_type_hints()) instead of the custom reflection system.

📘 Technical Design 028 — Reflection Metrics Architecture¶

🧭 1. Summary¶

Purpose:

Replace the generic, loosely-typed ReflectionMetrics.emit(event, payload) interface with a strongly-typed metrics API and hierarchical aggregation architecture. This design eliminates scattered try-except blocks that increase cognitive complexity, provides deterministic metric hooks with well-defined semantics, and enables hierarchical roll-up of metrics across the module tree (e.g., metrics for loryth.common automatically aggregate metrics from all loryth.common.* submodules).

At a glance:

• AggregatorTree collection: Generic, reusable tree data structure in loryth.common.collections that auto-creates parent nodes and replaces states on leaf updates
• MetricsState dataclass: Concrete state holding counts (operations, errors), durations (ms), and derived rates (cache hit, success)
• Strongly-typed metrics API: Dedicated methods per domain (e.g., record_resolution_success(), record_scan_error()) instead of unstructured emit() calls
• Snapshot pattern: Metrics implementation exposes read-only snapshots; external sinks retrieve snapshots on demand
• Safe-by-design: No try-except needed in calling code; metrics failures never impact reflection operations

📚 2. Context and Problem Statement¶

Problem¶

The old generic ReflectionMetrics.emit(event: str, payload: Mapping[str, Any]) API had four pain points:

1. Scattered error handling: Try-except blocks scattered across components to swallow exceptions
2. Unstructured payloads: Loosely-typed dictionaries with no schema; easy to mistype event names
3. No aggregation: Components fired independent events; no way to roll up metrics to parent modules
4. Multiple implementations: NullMetrics, LoggingMetrics, HookMetrics each re-implemented error handling

Goals¶

1. Never-raise semantics: Metrics failures never propagate; enforced at API level, not call sites
2. Strongly-typed API: Dedicated methods per domain (e.g., record_resolution_success()) with IDE autocomplete
3. Hierarchical aggregation: Metrics auto-roll up from loryth.common.reflection.scanner → loryth.common for dashboards
4. External observability: Pull snapshots on demand; no coupling to internal structures

Scope¶

• In scope: Strongly-typed metrics API, AggregatorTree, hierarchical rollup, snapshot pattern
• Out of scope: Performance optimization, real-time streaming, distributed tracing, type checking strategy

Assumptions¶

• Python 3.13+ with strict mypy; immutable MetricsState; module-based paths (['loryth', 'common', 'reflection']); single metrics instance per ReflectionContext

🧩 3. Design Overview¶

3.1 Key Components¶

3.2 Architecture¶

graph LR
    AT["AggregatorTree<br/>(generic)"] --> RMC["ReflectionMetricsCollector<br/>(AggregatorTree + State)"]
    RMC --> Snapshot["MetricsSnapshot<br/>(immutable view)"]
    RMC --> RTI["ReflectionContext"]
    RTI --> Components["Components:<br/>resolve, scan, hint, etc."]
    RMC --> External["External Sinks<br/>(pull snapshots)"]

    classDef collection fill:none,stroke:#00C853,stroke-width:2px,color:#00C853
    classDef metrics fill:none,stroke:#8E24AA,stroke-width:2px,color:#8E24AA
    classDef integration fill:none,stroke:#1976D2,stroke-width:2px,color:#1976D2

    class AT collection
    class RMC,Snapshot metrics
    class RTI,Components,External integration

3.3 Interface Impact (Before → After)¶

3.4 Removal/Deprecation Policy¶

Alpha stage: Clean break. - Old ReflectionMetrics protocol, NullMetrics, LoggingMetrics, HookMetrics are removed entirely in Phase 2. - All reflection components and tests are updated to use the new typed API in the same phase. - No deprecation period; this is synchronized within the alpha development cycle.

📐 4. Implementation Details¶

4.1 AggregatorTree Protocol (loryth.common.collections)¶

from typing import Protocol, TypeGuard
from collections.abc import Hashable, Callable, Iterable

class AggregatorTree[Node: Hashable, State](Protocol):
    """Generic tree that auto-creates parent nodes and aggregates state on updates.

    Paths are represented as iterables of nodes (e.g., ['loryth', 'common', 'reflection']).
    When a leaf is updated, parent states are replaced via the aggregation callback.
    State is assumed immutable; updates replace (not merge) parent states.
    """

    def get(self, path: Iterable[Node]) -> State | None:
        """Retrieve state for a given path.

        :param path: Sequence of nodes forming the path (e.g., ['loryth', 'common'])
        :return: State at path, or None if path does not exist
        :rtype: State | None
        """
        ...

    def set(self, path: Iterable[Node], state: State) -> None:
        """Set state at path, auto-creating parents and triggering aggregation.

        Auto-creates parent nodes if they don't exist. After setting, applies
        the aggregation callback to update all parent states up to root.

        :param path: Sequence of nodes forming the path
        :param state: New state to set (assumed immutable)
        """
        ...

    def get_root(self) -> State | None:
        """Retrieve state at root of tree.

        :return: Root state, or None if tree is empty
        :rtype: State | None
        """
        ...

    def clear(self, path: Iterable[Node] | None = None) -> None:
        """Clear state at path or entire tree if path is None.

        :param path: Sequence of nodes forming the path, or None to clear root/all
        """
        ...

4.2 ReflectionMetrics Protocol (strictly-typed, never-raise)¶

from dataclasses import dataclass, field
from typing import Protocol

@dataclass(frozen=True)
class MetricsSnapshot:
    """Immutable snapshot of metrics at a given point in time."""
    path: tuple[str, ...]

    # Resolution, Scanning, Invocation (attempted, succeeded, failed, durations, rates)
    resolution_attempted: int = 0
    resolution_succeeded: int = 0
    resolution_failed: int = 0
    resolution_total_ms: float = 0.0
    resolution_avg_ms: float = 0.0
    resolution_success_rate: float = 0.0

    scan_attempted: int = 0
    scan_succeeded: int = 0
    scan_failed: int = 0
    scan_total_ms: float = 0.0
    scan_avg_ms: float = 0.0
    scan_success_rate: float = 0.0

    # Annotation/Hint lookups (lookups, found, cache metrics)
    annotation_lookups: int = 0
    annotation_found: int = 0
    annotation_lookup_total_ms: float = 0.0
    hint_lookups: int = 0
    hint_found: int = 0
    hint_cache_hits: int = 0
    hint_cache_hit_rate: float = 0.0
    hint_lookup_total_ms: float = 0.0
    hint_direct_resolutions: int = 0
    hint_augmented_resolutions: int = 0
    hint_augmentation_total_attempts: int = 0

    invocation_attempted: int = 0
    invocation_succeeded: int = 0
    invocation_failed: int = 0
    invocation_total_ms: float = 0.0
    invocation_success_rate: float = 0.0

    errors: dict[str, int] = field(default_factory=dict)


class ReflectionMetrics(Protocol):
    """Strongly-typed metrics API (all methods safe/never-raise)."""

    # Record-* methods (success/failure variants for each operation)
    def record_resolution_success(self, path: Iterable[str], subject: str, duration_ms: float) -> None: ...
    def record_resolution_failure(self, path: Iterable[str], subject: str, error_type: str, duration_ms: float) -> None: ...
    def record_scan_success(self, path: Iterable[str], module: str, item_count: int, duration_ms: float) -> None: ...
    def record_scan_failure(self, path: Iterable[str], module: str, error_type: str, duration_ms: float) -> None: ...
    def record_annotation_lookup(self, path: Iterable[str], subject: str, found: bool, duration_ms: float) -> None: ...
    def record_hint_lookup(self, path: Iterable[str], subject: str, found: bool, duration_ms: float, cache_hit: bool, augmentation_attempts: int = 0) -> None: ...
    def record_invocation_success(self, path: Iterable[str], subject: str, duration_ms: float) -> None: ...
    def record_invocation_failure(self, path: Iterable[str], subject: str, error_type: str, duration_ms: float) -> None: ...

    # Context managers (automatic timing)
    @contextmanager
    def resolution_timer(self, path: Iterable[str], subject: str) -> Generator[None, None, None]: ...
    @contextmanager
    def scan_timer(self, path: Iterable[str], module: str) -> Generator[object, None, None]: ...
    @contextmanager
    def hint_lookup_timer(self, path: Iterable[str], subject: str) -> Generator[object, None, None]: ...

    # Snapshot retrieval
    def snapshot(self, path: Iterable[str] | None = None) -> MetricsSnapshot: ...
    def snapshot_root(self) -> MetricsSnapshot: ...

4.3 Component Architecture¶

MetricsState is internally structured using sub-dataclasses (OperationMetrics, AnnotationMetrics, HintMetrics) for cleaner aggregation logic; the external API remains flat.

@dataclass
class MetricsState:
    """Mutable state holder with grouped metrics for aggregation."""
    resolution: OperationMetrics = field(default_factory=OperationMetrics)
    scan: OperationMetrics = field(default_factory=OperationMetrics)
    invocation: OperationMetrics = field(default_factory=OperationMetrics)
    annotation: AnnotationMetrics = field(default_factory=AnnotationMetrics)
    hint: HintMetrics = field(default_factory=HintMetrics)
    errors: dict[str, int] = field(default_factory=dict)

    def to_snapshot(self, path: tuple[str, ...]) -> MetricsSnapshot:
        """Convert to read-only snapshot (flattens nested structure)."""
        # Flatten nested structure: self.resolution → resolution_*, self.scan → scan_*, etc.
        ...

    @staticmethod
    def aggregate(children: list[MetricsState]) -> MetricsState:
        """Aggregate multiple child states into a single parent state."""
        # Delegates to sub-metric aggregate methods for cleaner code
        ...

ReflectionMetricsCollector implementation uses AggregatorTree to maintain hierarchical state, record metrics with never-raise semantics, and provide snapshot retrieval. See source for full implementation details.

4.4 Metrics & Observability¶

External sinks retrieve metrics via snapshot() calls and access typed attributes directly.

⚙️ 5. Semantics¶

Path representation: Iterables of strings (e.g., ["loryth", "common", "reflection"] for module-level metrics, None for system-wide).

Aggregation rules: Leaf updates trigger parent recalculation — counts summed, durations aggregated (total/avg), rates recomputed, errors merged.

Context managers: Auto-timing and error classification: - resolution_timer(path, subject): Records success/failure on exit/exception - scan_timer(path, module): Caller increments item counter during block - hint_lookup_timer(path, subject): Caller sets found and cache_hit flags

Clearing metrics: clear(path) resets at path (or None for entire tree); primarily for testing.

Thread safety: If shared across threads, underlying AggregatorTree must be thread-safe. Context managers not re-entrant.

🧪 6. Examples¶

Basic usage with manual timing:

context = ReflectionContext()
start = time.perf_counter()
try:
    result = resolve_type("SomeType")
    duration_ms = (time.perf_counter() - start) * 1000
    context.metrics.record_resolution_success(["loryth", "common", "reflection"], "SomeType", duration_ms)
except Exception as ex:
    duration_ms = (time.perf_counter() - start) * 1000
    context.metrics.record_resolution_failure(["loryth", "common", "reflection"], "SomeType", type(ex).__name__, duration_ms)
    raise

Context managers for automatic timing:

with context.metrics.resolution_timer(["loryth", "common", "reflection"], "MyClass"):
    result = complex_resolution_logic()  # Automatically timed; success/failure recorded

snapshot = context.metrics.snapshot(["loryth", "common", "reflection"])
print(f"Success rate: {snapshot.resolution_success_rate:.1%}")

Hierarchical aggregation: Record metrics from multiple components at different paths; parent paths automatically aggregate:

context.metrics.record_resolution_success(["loryth", "common", "reflection", "resolver"], "Foo", 10.5)
context.metrics.record_resolution_success(["loryth", "common", "reflection", "scanner"], "Bar", 5.3)
root = context.metrics.snapshot_root()  # Aggregates all children

External observability: Pull snapshots on demand for logging or APM:

snapshot = context.metrics.snapshot_root()
log_entry = {"resolution_attempted": snapshot.resolution_attempted, "success_rate": snapshot.resolution_success_rate, ...}

🧪 7. Test Matrix¶

Unit Coverage: - Auto-create parents, retrieve states, apply aggregation, handle empty tree and missing paths (AggregatorTree) - Construct MetricsState, convert to snapshot, verify frozen, aggregate sub-metrics (MetricsState/sub-dataclasses) - Record resolution/scan/annotation/hint/invocation with success/failure variants, context managers, never-raise guarantee, snapshot retrieval, clear() method (ReflectionMetricsCollector)

Integration Coverage: - Hierarchical aggregation from multiple components at different paths; parent/root snapshots aggregate children correctly; concurrent recording - Component integration: PythonImportResolver, PackageScanner, TypeHintProvider record metrics without try-except blocks

Regression Coverage: - Error resilience: metrics collection continues despite failures; snapshots valid after failed operations; concurrent access thread-safe

✅ 8. Acceptance Criteria¶

• ✅ AggregatorTree[Node, State] implemented with auto-create parents, callback aggregation, path-based get/set, clear(path) support
• ✅ MetricsState dataclass with typed attributes: resolution/scanning/invocation/annotation/hint metrics plus error tracking
• ✅ MetricsSnapshot frozen dataclass with same typed attributes
• ✅ ReflectionMetrics protocol with typed record_* methods, context managers, snapshot retrieval, never-raise guarantee
• ✅ ReflectionMetricsCollector implementation with all record_* methods, context managers, clear() support
• ✅ Unit tests: AggregatorTree, MetricsState, Collector (records, context managers, snapshots, clear())
• ✅ Integration tests: hierarchical aggregation, snapshot consistency, context manager paths, tree clearing
• ✅ ReflectionContext constructor accepts optional metrics parameter
• ✅ All reflection components updated to new typed API (Phase 2: PythonImportResolver, PackageScanner, TypeHintProvider, SafeCallableInvoker, HighFidelityTypeHintProvider)
• ✅ Old metrics implementations removed (NullMetrics, LoggingMetrics, HookMetrics)
• ✅ Try-except blocks for metrics removed from all components
• ✅ Mypy strict mode compliance maintained (strict generics, no Any)
• ✅ Documentation updated with usage examples and context manager patterns

🚀 9. Rollout and Migration¶

Phase 1 — Foundation: - AggregatorTree with auto-create parents and callback aggregation - MetricsState dataclass with nested sub-dataclasses - ReflectionMetricsCollector with typed API and snapshot support - Comprehensive unit/integration tests for all components - ReflectionContext constructor updated to accept metrics parameter

Phase 2 — Migration: - All reflection components updated to new typed API (PythonImportResolver, PackageScanner, TypeHintProvider, SafeCallableInvoker, HighFidelityTypeHintProvider) - Old metrics implementations removed (NullMetrics, LoggingMetrics, HookMetrics) - Try-except blocks for metrics removed from all components - Integration tests for end-to-end workflows with new API

⚠️ 10. Risks and Mitigations¶

📝 11. Changelog¶

• 2025-10-24 — Phase 2 completion: All reflection components migrated to typed metrics API; augmentation tracking implemented; old implementations (NullMetrics, LoggingMetrics, HookMetrics) removed; 811+ tests passing.
• 2025-10-24 — Phase 1 completion: Sub-dataclasses extracted (OperationMetrics, AnnotationMetrics, HintMetrics); MetricsState simplified from ~76 to ~20 lines via delegation; all complexity thresholds now pass.
• 2025-10-21 — Design refinements: MetricsState/MetricsSnapshot converted to typed attributes; clear(path) added to AggregatorTree; context managers (resolution_timer, scan_timer, hint_lookup_timer) added to protocol.
• 2025-10-21 — Initial design: AggregatorTree protocol, MetricsState dataclass, ReflectionMetrics typed API, two-phase implementation plan.

📖 13. Glossary¶

This technical design is maintained by the loryth.common reflection maintainers. See Roadmap 002 Phase B for context and timeline.