Skip to content

Observability Guide (Milestone 5 follow‑ups)

ARCHIVED: The loryth.common.reflection subsystem has been removed from the codebase. All modules referenced in this guide (ReflectionContext, TypeHintConfig, ReflectionMetricsCollector, etc.) no longer exist. This guide is retained for historical reference only. Do not use it as implementation guidance.

Audience: Advanced contributors and consumers of loryth.common.reflection Status: Archived (reflection subsystem removed) Last updated: 2025-09-27

Overview

The reflection subsystem provides two complementary observability channels: - Structured DEBUG logs under a single logger name to inspect forward‑reference augmentation, retries, and unresolved outcomes deterministically. - A centralized ReflectionMetrics sink used by multiple components (HighFidelityTypeHintProvider, PythonImportResolver, PackageScanner, SafeCallableInvoker). Metrics are best‑effort and must never raise.

Logs are disabled by default; enable with the debug flag and standard logging configuration. Metrics are off unless you provide a metrics sink when constructing ReflectionContext.

Logger

  • Name: loryth.common.reflection
  • Level: DEBUG for detailed traces; INFO and above are quiet by default.

Enabling logs

  • Application:
import logging
from loryth.common.reflection import ReflectionContext
from loryth.common.reflection.reflection_context import TypeHintConfig

logging.basicConfig(level=logging.INFO)  # your app baseline
logging.getLogger("loryth.common.reflection").setLevel(logging.DEBUG)

ctx = ReflectionContext(
    type_hint_config=TypeHintConfig(strict=False),
)
# HighFidelityTypeHintProvider picks up debug logging via TypeHintConfig.strict/debug knobs.
  • Tests (pytest):
import logging
import pytest


@pytest.fixture(autouse=True)
def _enable_reflection_debug_logs():
    logging.getLogger("loryth.common.reflection").setLevel(logging.DEBUG)

Event reference (DEBUG)

Events are short, consistent, and designed to be grep‑friendly. All messages are emitted only when debug=True.

  • get_type_hints_first_fail

    • Emitted when typing.get_type_hints(include_extras=True) raises. Includes the exception class and an optional list of priority tokens extracted from the first error.
  • tokens

    • Summary of gathered tokens (dotted and bare) with the priority subset first when available. Used to bound and order augmentation attempts.
  • import_attempt

    • A dotted import attempted during augmentation (e.g., "pkg.types: X").
  • sibling_attempt

    • A sibling module guess attempted inside the owner package (bounded, fixed order).
  • sibling_reuse

    • Indicates that a previously imported sibling module was reused instead of importing again.
  • get_type_hints_retry_fail

    • A retry of typing.get_type_hints after augmentation still failed, but the provider will continue with tolerant fallback if strict=False.
  • get_type_hints_retry_error

    • A retry raised an error other than NameError/KeyError; includes context to aid debugging.
  • unresolved

    • Final summary when unresolved markers are produced in tolerant mode, with slots, tokens, and attempted imports listed.

Verifying locally

  • Minimal snippet to see events:
import logging
from loryth.common.reflection import ReflectionContext
from loryth.common.reflection.reflection_context import TypeHintConfig

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("loryth.common.reflection").setLevel(logging.DEBUG)
ctx = ReflectionContext(
    type_hint_config=TypeHintConfig(strict=False),
)
# call ctx.type_hint_for_property/field/callable and inspect logs

Centralized metrics (quick start)

The reflection subsystem uses a strongly-typed metrics API with hierarchical aggregation. Several components record metrics: - HighFidelityTypeHintProvider (type hint lookups with cache tracking and augmentation awareness) - PythonImportResolver (type/module/package resolutions) - PackageScanner (module scanning operations) - SafeCallableInvoker (callable invocation operations)

A ReflectionMetricsCollector is created by default in ReflectionContext. You can retrieve snapshots on demand to observe reflection subsystem behavior.

Basic usage — manual timing

from loryth.common.reflection import ReflectionContext

# Create context (metrics collector created automatically)
context = ReflectionContext()

# Metrics are collected automatically by components
# Retrieve a snapshot to inspect state
snapshot = context.metrics.snapshot_root()

print(f"Resolution attempts: {snapshot.resolution_attempted}")
print(f"Resolution successes: {snapshot.resolution_succeeded}")
print(f"Hint lookups: {snapshot.hint_lookups}")
print(f"Hint cache hit rate: {snapshot.hint_cache_hit_rate:.1%}")

Using context managers for automatic timing

Components use context managers internally for automatic timing and error classification. You can integrate metrics into your own code:

from time import perf_counter
from loryth.common.reflection import ReflectionContext

context = ReflectionContext()
path = ["my", "module", "name"]

# Automatic timing and error handling
with context.metrics.resolution_timer(path, "MyClass"):
    # Code here is timed automatically
    # Success recorded on normal exit
    # Failure recorded on exception
    result = some_resolution_logic()

# Scan with item counting
with context.metrics.scan_timer(path, "my.module") as (item_count,):
    for item in items:
        context.metrics.snapshot()  # Side effect: increments item_count
        item_count[0] += 1

# Hint lookup with cache tracking
with context.metrics.hint_lookup_timer(path, "SomeClass") as hint_ctx:
    hints = lookup_type_hints()
    hint_ctx.found = len(hints) > 0
    hint_ctx.cache_hit = was_cached

Hierarchical aggregation

Metrics automatically aggregate from child modules to parents:

from loryth.common.reflection import ReflectionContext

context = ReflectionContext()

# Record metrics at specific paths
context.metrics.record_resolution_success(
    path=["loryth", "common", "reflection", "type_resolver"],
    subject="Foo",
    duration_ms=10.5,
)
context.metrics.record_resolution_success(
    path=["loryth", "common", "reflection", "scanner"],
    subject="Bar",
    duration_ms=5.3,
)

# Retrieve aggregated snapshot at parent level
parent_snapshot = context.metrics.snapshot(["loryth", "common", "reflection"])
print(f"Module level attempts: {parent_snapshot.resolution_attempted}")  # 2
print(f"Module level successes: {parent_snapshot.resolution_succeeded}")  # 2

# Retrieve system-wide snapshot
root_snapshot = context.metrics.snapshot_root()
print(f"System-wide attempts: {root_snapshot.resolution_attempted}")  # 2

Exporting metrics to external systems

Pull snapshots and export to logging, APM platforms, or monitoring systems:

import json
import logging
from loryth.common.reflection import ReflectionContext

def export_metrics(context: ReflectionContext) -> None:
    """Export metrics snapshot to logging system."""
    snapshot = context.metrics.snapshot_root()

    log_data = {
        "resolution_attempted": snapshot.resolution_attempted,
        "resolution_succeeded": snapshot.resolution_succeeded,
        "resolution_success_rate": snapshot.resolution_success_rate,
        "scan_attempted": snapshot.scan_attempted,
        "scan_succeeded": snapshot.scan_succeeded,
        "hint_lookups": snapshot.hint_lookups,
        "hint_cache_hit_rate": snapshot.hint_cache_hit_rate,
        "hint_augmented_resolutions": snapshot.hint_augmented_resolutions,
        "hint_augmentation_attempts": snapshot.hint_augmentation_total_attempts,
        "errors": snapshot.errors,
    }
    logging.info("reflection_metrics: %s", json.dumps(log_data))

# Call on demand or on a timer
export_metrics(context)

Understanding augmentation metrics

Type hint resolution may require "augmentation" when resolving string-based forward references:

  • direct_resolutions: Type hints resolved without augmentation (found immediately in module)
  • augmented_resolutions: Type hints requiring namespace augmentation attempts
  • augmentation_total_attempts: Sum of attempts across all augmented resolutions

Example: If 5 hint lookups occur and 2 require augmentation averaging 1.5 attempts each: - hint_lookups: 5 - hint_direct_resolutions: 3 - hint_augmented_resolutions: 2 - hint_augmentation_total_attempts: 3 (1.5 + 1.5)

This helps identify modules with complex type hint patterns and opportunities for optimization.

Notes

  • Determinism: The augmentation pass and sibling guesses are bounded and ordered; DEBUG logs reflect that order.
  • Safety: The provider does not execute blocks guarded by typing.TYPE_CHECKING.
  • Never-raise semantics: All metrics operations are safe and never propagate exceptions; failures are handled internally.