📘 Technical Design 029 — Member Parsing Seam Architecture¶

🧭 1. Summary¶

MemberBuilder mixes discovery, enrichment, caching, and metrics inside one monolithic class. TD-029 replaces it with a single HandleGenerationStrategy protocol and pluggable implementations, letting TypeResolver orchestrate specs via a depth-first parsing context that manages ownership chains and circular references.

At a glance:

• Single HandleGenerationStrategy protocol unifying identify + enrich responsibilities.
• Source-specific implementations (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper) compose in pipeline.
• Depth-first discovery via stack (LIFO) minimizes memory footprint.
• Immutable specs use copy() + with_*() methods for enrichment.
• Per-strategy metrics and deduplication handling for circular imports/references.
• All strategy components live in loryth.common.reflection.engine.strategies (public API via engine/__init__.py).

📚 2. Context and Problem Statement¶

MemberBuilder was designed when reflection scenarios were limited to core Python constructs. As frameworks layered metaclasses, descriptors, runtime mutation, and domain-specific metadata, the hardcoded parsing pipeline grew to ~450 lines with intertwined responsibilities. Today the class couples cache handling, discovery, metadata enrichment, and observability, making it nearly impossible to add new behavior without editing private methods. Testing pathways in isolation requires deep mocking because the only seam is the public build_* entry points.

Pain points driving this design:

1. No extensibility: custom member types (ORM columns, descriptors) require editing MemberBuilder internals.
2. Hardcoded detection: _process_callable_or_field_member() embeds method vs field heuristics that cannot be swapped.
3. Monolithic complexity: seven parsing responsibilities live in one class, exceeding our complexity guardrails.
4. Testing friction: contributors cannot unit-test specific parsing behaviors without invoking the entire builder.
5. Maintenance drag: supporting emerging Python proposals (e.g., PEP 681) requires coordinated edits across multiple private helpers.
6. Missing composition: teams cannot combine core parsing with framework-specific discovery in a controlled order.

2.1 Goals¶

1. Publish HandleGenerationStrategy protocol: single interface for identifying + enriching specs with stackability contracts and examples.
2. Extract core mappers: create source-type implementations (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper) each ≤100 lines.
3. Deterministic composition: define ordering semantics so multiple strategies refine specs safely (last-writer-wins per qualname).
4. Preserve public APIs: keep TypeResolver.resolve() signature and cache identity guarantees untouched.
5. Observable extensibility: provide per-strategy metrics namespaces (handle_generation.<slug>) so performance/failures are attributable.
6. Contract + integration coverage: deliver suites that validate pipeline orchestration and composed configurations.
7. Zero regression: prove parity with current MemberBuilder outputs across fixtures.
8. Public API clarity: export strategy components from loryth.common.reflection.engine for external customization.

2.2 Out of Scope¶

• Forward-reference resolution algorithms (handled in a later roadmap phase).
• Type-checking/type-provider redesign (Phase F of the roadmap).
• Performance optimization work beyond the complexity reductions here.
• Annotation/Type Hint Provider overhaul (Phase E of the roadmap).

2.3 Assumptions and Constraints¶

• Python 3.13+ with strict mypy, PEP 695 generics, and updated typing features.
• Immutable member identity: once discovered, qualname stays fixed; specs enrich metadata only.
• Metrics never-raise: HandleBuildingPipeline instrumentation wraps every strategy invocation and suppresses metric failures.
• Qualified addressing: all members are keyed by qualname throughout introspection phases.
• External strategy access: all strategy components are part of public API in loryth.common.reflection.engine.

🧩 3. Design Overview¶

The core insight: Spec generation is fundamentally mapping a source object (module, class, function, parameter, property) to a Spec. A single-strategy pipeline maps sources to specs via map(source, context) -> spec, where strategies can enrich the spec and enqueue discovered members for recursive processing.

Narrative flow: A caller asks TypeResolver to resolve my_app.Customer. On a cache miss, the resolver creates a HandleGenerationContext seeded with a placeholder for my_app.Customer and delegates to HandleBuildingPipeline. The context manages a depth-first stack of members to build:

1. Pop my_app.Customer (placeholder spec)
2. Import the class, run mappers via pipeline.map(Customer_class, placeholder_spec, context)
3. Mappers identify it's a type, enrich with methods/properties
4. Enrichment discovers place_order() method → calls context.ensure_member(method_spec) → pushed to stack
5. Pop my_app.Customer.place_order (placeholder spec)
6. Run mappers on the method → discovers parameters → pushed to stack
7. Continue until stack empty
8. finalize_all() materializes the graph with ownership chains intact
9. Resolver caches and returns TypeHandle to caller

3.1 Key Components¶

3.2 Architecture¶

graph TD
    User["User calls<br/>resolver.resolve()"]
    CacheHit{Cached?}
    ParsingSetup["Create ParsingContext<br/>with root qualname"]
    Loop["Dequeue pending<br/>builder"]
    Import["Import source<br/>object"]
    Mappers["Run mapping<br/>strategies"]
    Enrich["Strategies identify<br/>+ enrich builder"]
    Discover["Discovered members<br/>enqueued via<br/>ensure_member()"]
    LoopCheck{More<br/>pending?}
    Finalize["finalize_all()<br/>materializes graph"]
    Cache["Cache result"]
    Return["Return Handle<br/>to caller"]

    User --> CacheHit
    CacheHit -->|yes| Return
    CacheHit -->|no| ParsingSetup
    ParsingSetup --> Loop
    Loop --> Import
    Import --> Mappers
    Mappers --> Enrich
    Enrich --> Discover
    Discover --> LoopCheck
    LoopCheck -->|yes| Loop
    LoopCheck -->|no| Finalize
    Finalize --> Cache
    Cache --> Return

    classDef process fill:none,stroke:#1976D2,stroke-width:2px,color:#1976D2;
    classDef decision fill:none,stroke:#FF9800,stroke-width:2px,color:#FF9800;
    classDef terminal fill:none,stroke:#4CAF50,stroke-width:2px,color:#4CAF50;

    class Mappers,Enrich,Discover process;
    class CacheHit,LoopCheck decision;
    class Return terminal;

3.3 Interface Impact (Before → After)¶

3.4 Removal/Deprecation Policy¶

The legacy MemberBuilder façade is removed. Callers now compose ParsingPipeline, ParsingContext, and TypeResolver instances (directly or via thin factories) to configure custom stacks. Deprecated behavior—overriding MemberBuilder internals or mutating MemberHandle directly—is no longer supported; contributors must implement strategies instead.

⚙️ 4. Runtime Flow¶

4.1 ParsingContext Lifecycle¶

ParsingContext owns the full lifecycle of builders during a cache miss. It manages a depth-first stack (LIFO) to minimize memory footprint during discovery.

class ParsingContext:
    """Manages builder lifecycle: discovery, registration, finalization.

    Uses a stack (LIFO) for depth-first discovery to minimize peak memory usage.
    When building Customer, immediately process its methods, then their parameters,
    before discovering other top-level types.
    """

    _builder_registry: dict[str, HandleBuilder]
    """Stores the latest builder (placeholder or concrete) per qualname.

    Acts as authoritative source: strategies look up existing builders here
    before creating placeholders.
    """

    _pending: list[HandleBuilder]
    """Stack (LIFO) of builders to process. Depth-first discovery.

    Use stack.pop() to get next item (LIFO). When builder discovered,
    push it onto stack to process immediately (depth-first).
    """

    _finalized: dict[str, HandleBuilder]
    """Builders that have been fully processed and materialized.

    Tracks which builders are done to detect circular references.
    next_pending() skips qualnames already in _finalized.
    """

    _cache: HierarchicalCache
    """Read-only cache for type lookups during mapping."""

    def register_builder(self, builder: HandleBuilder) -> None:
        """Add or replace a builder in the registry. Deduplicates by qualname.

        :param builder: Builder to register. If qualname exists, replaces old entry.
        """

    def next_pending(self) -> HandleBuilder | None:
        """Pop next builder from stack for processing. Skips finalized ones.

        Handles circular imports: if a qualname is already finalized, skip it.
        Returns None when stack empty.

        :return: Next HandleBuilder to process, or None if stack empty.
        """

    def finalize_all(self) -> dict[str, HandleBuilder]:
        """Materialize all builders. Called when stack is empty.

        Walks builder registry, materializes each builder that hasn't been
        finalized yet. Circular references resolve automatically since both
        handles are in the registry.

        :return: dict mapping qualname → finalized HandleBuilder.
        """

Depth-First Processing Example: Customer → place_order → parameter_order_id

1. resolve("my_app.Customer") misses cache → ParsingContext created with stack = [Customer placeholder]
2. Pop Customer placeholder, import class, run mappers
3. Mappers discover place_order() method → call context.ensure_member(method_placeholder) → push onto stack
4. Pop place_order placeholder, import method, run mappers
5. Mappers discover order_id parameter → call context.ensure_member(param_placeholder) → push onto stack
6. Pop parameter placeholder, run mappers
7. No more pending → finalize_all() materializes with ownership chains intact

Memory benefit: Stack is small throughout (only current branch depth); contrast with breadth-first where all Customer's methods would sit in queue before processing any.

Circular Reference Handling: Customer ↔ Order

1. Pop Customer, mappers discover Order reference → ensure_member(Order placeholder) → push to stack
2. Pop Order (or complete Customer first, depending on push order), mappers discover Customer reference → Customer already in registry → skip (dedup)
3. Finalization resolves both: Customer.orders → [Order], Order.customer → Customer (same finalized instance)

4.2 MappingContext¶

The context provided to each mapping strategy:

@dataclass(frozen=True)
class MappingContext:
    source: object  # Python object being mapped
    builder: HandleBuilder  # Placeholder to enhance
    parent: TypeHandleBuilder | None  # Containing type (if applicable)
    module: ModuleHandleBuilder  # Owning module (guaranteed)
    cache: HierarchicalCache  # Read-only cache
    _parsing_context: ParsingContext  # Private reference

    def ensure_member(self, builder: HandleBuilder) -> HandleBuilder:
        """Enqueue discovered member builder for recursive processing."""
        self._parsing_context.register(builder)
        self._parsing_context._pending.append(builder)
        return builder

    def get_builder(self, qualname: str) -> HandleBuilder | None:
        """Look up an already-discovered builder."""
        return self._parsing_context._registry.get(qualname)

Immutability guarantee: Strategies cannot mutate the context. They return enhanced builders via dataclasses.replace().

4.3 ParsingPipeline Orchestration¶

class ParsingPipeline:
    def __init__(self, mapping_strategies: Sequence[HandleBuildingStrategy], metrics: ...):
        self._mapping_strategies = [ModuleHandleMapper(), TypeHandleMapper(), FunctionHandleMapper(), ...]

    def map(self, source: object, builder: HandleBuilder, parsing_context: ParsingContext,
            parent: TypeHandleBuilder | None, module: ModuleHandleBuilder) -> HandleBuilder:
        """Execute mapping strategies in sequence."""
        context = MappingContext(source, builder, parent, module, parsing_context._cache, parsing_context)

        current_builder = builder
        for strategy in self._mapping_strategies:
            current_builder = self._execute_with_metrics(
                strategy,
                lambda: strategy.map(source, dataclasses.replace(context, builder=current_builder))
            )
        return current_builder

Metrics Emission: - Strategy executes: strategy.map(source, context) - Wraps with try/finally to measure duration, count, errors - Namespace: handle_building.<strategy_slug> (e.g., handle_building.module, handle_building.type) - Metrics never raise; failures are logged but don't interrupt mapping

🧱 5. Strategy & Builder Catalog¶

5.1 HandleBuilder Contract¶

Builders are dataclasses with type-safe fields. Strategies use dataclasses.replace() for immutable enrichment.

@dataclass(frozen=True)
class HandleBuilder:
    """Immutable builder for Handle objects.

    Use dataclasses.replace() to create enhanced copies with new attributes.
    Frozen ensures thread-safety and prevents accidental mutations.
    """

    qualname: str
    """Fully qualified name (e.g., "my_app.Customer"). Set once at creation, never changes."""

    module: ModuleHandleBuilder | None = None
    """Owning module. Set during mapping, read-only."""

    parent: TypeHandleBuilder | None = None
    """Containing type for methods/fields. Set during mapping, read-only."""

    @classmethod
    def placeholder(cls, qualname: str) -> "HandleBuilder":
        """Create a lightweight placeholder builder (just qualname).

        :param qualname: Fully qualified name of the builder.
        :return: Placeholder builder with only qualname set.
        """
        return cls(qualname=qualname)

Concrete Builders (example):

@dataclass(frozen=True)
class TypeHandleBuilder(HandleBuilder):
    """Builder for TypeHandle objects."""

    bases: tuple[TypeHandleBuilder, ...] = ()
    """Base classes."""

    type_hints: dict[str, TypeHandleBuilder] | None = None
    """Type hints for type parameters."""

    methods: list[MethodHandleBuilder] = field(default_factory=list)
    """Discovered methods."""

Usage Pattern:

# Strategies enrich builders using dataclasses.replace()
builder = context.builder  # TypeHandleBuilder
# Instead of: builder.with_attr("bases", bases)
# Use:
builder = replace(builder, bases=bases)

# Chain multiple enrichments
builder = replace(builder, bases=bases, type_hints=hints)

Design principles: - Type-safe API: Each field is explicitly typed; no string keys - Immutable by design: frozen=True prevents mutations - Minimal placeholders: Just qualname at creation; other fields populated during mapping - Dataclasses.replace(): Leverages standard library; no custom clone logic needed

5.3 HandleBuildingStrategy Protocol¶

A single protocol that both identifies a source's type AND enriches the builder with discovered members. Source-specific implementations (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper) execute in sequence, each enhancing the builder.

class HandleBuildingStrategy(Protocol):
    """Maps a source object to an enhanced Handle builder.

    Responsibilities:
    1. Identify: Determine source type (module, type, function, property, etc.)
    2. Enrich: Discover members and enqueue them via context.ensure_member()
    3. Return: Enhanced builder (immutable, via with_attr() chaining)

    Multiple strategies execute in sequence. Later strategies refine earlier results.
    Strategies never return None; pass through unchanged if not applicable.
    """

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        """Transform a source object and builder into an enhanced builder.

        :param source: The Python object being mapped (module, class, function, etc.).
        :param context: Mapping context with builder, parent, module, cache, ensure_member().
        :return: Enhanced builder with discovered members and metadata.
        :raises ValueError: If source is unsupported and strategy requires strict validation.
        :rtype: HandleBuilder
        """
        ...

Contract highlights: - Single responsibility: Identify the source AND enrich with members in one pass - Pipeline composition: Multiple strategies execute in order; each refines the builder - Member enqueueing: Discovered members enqueued via context.ensure_member(placeholder) for recursive processing - Ownership guarantee: Parent type and owning module are guaranteed to be building in parallel - Immutability: Return new builder via dataclasses.replace(), never mutate input

5.4 Source-Specific Mapper Implementations¶

Core mappers handle standard Python objects (modules, types, functions). Custom mappers chain after core mappers to specialize behavior.

5.4.1 ModuleHandleMapper¶

from dataclasses import replace

class ModuleHandleMapper(HandleBuildingStrategy):
    """Maps module objects to ModuleHandle builders."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        if not isinstance(source, types.ModuleType):
            return context.builder  # Not applicable, pass through

        # Identify: core module properties
        builder = replace(context.builder,
            qualname=source.__name__,
            is_package=hasattr(source, "__path__"))

        # Enrich: discover submodules (if package)
        if hasattr(source, "__path__"):
            for _, name, is_pkg in pkgutil.iter_modules(source.__path__):
                submod = ModuleHandleBuilder.placeholder(f"{source.__name__}.{name}")
                context.ensure_member(submod)

        # Enrich: discover types and functions
        for name, obj in vars(source).items():
            if isinstance(obj, type) and obj.__module__ == source.__name__:
                type_builder = TypeHandleBuilder.placeholder(f"{source.__qualname__}.{name}")
                context.ensure_member(type_builder)
            elif inspect.isfunction(obj) and obj.__module__ == source.__name__:
                func_builder = FunctionHandleBuilder.placeholder(f"{source.__qualname__}.{name}")
                context.ensure_member(func_builder)

        return builder

5.4.2 TypeHandleMapper¶

from dataclasses import replace

class TypeHandleMapper(HandleBuildingStrategy):
    """Maps class objects to TypeHandle builders."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        if not isinstance(source, type):
            return context.builder

        # Identify
        builder = replace(context.builder,
            qualname=source.__qualname__,
            module=context.module,
            bases=source.__bases__)

        # Enrich: discover methods, fields, properties
        for name, value in vars(source).items():
            if inspect.isfunction(value):
                method = MethodHandleBuilder.placeholder(f"{source.__qualname__}.{name}")
                context.ensure_member(method)
            elif isinstance(value, property):
                prop = PropertyHandleBuilder.placeholder(f"{source.__qualname__}.{name}")
                context.ensure_member(prop)

        return builder

5.4.3 FunctionHandleMapper¶

from dataclasses import replace

class FunctionHandleMapper(HandleBuildingStrategy):
    """Maps function/method objects to FunctionHandle builders."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        if not (inspect.isfunction(source) or inspect.ismethod(source)):
            return context.builder

        # Identify
        builder = replace(context.builder,
            qualname=source.__qualname__,
            module=context.module)

        # Set parent if this is a method
        if context.parent is not None:
            builder = replace(builder, type=context.parent)

        # Enrich: discover parameters
        sig = inspect.signature(source)
        for param_name, param in sig.parameters.items():
            param_builder = ParameterHandleBuilder.placeholder(
                f"{source.__qualname__}.{param_name}"
            )
            context.ensure_member(param_builder)

        return builder

5.5 Custom Strategy Example: SQLAlchemy¶

Framework integrators extend the mapper pipeline to handle domain-specific objects. Custom mappers chain AFTER core mappers:

from dataclasses import replace

class SQLAlchemyHandleMapper(HandleBuildingStrategy):
    """Discovers SQLAlchemy ORM columns and relationships as field handles."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        # Only handle SQLAlchemy mapped classes
        if not isinstance(source, type) or not hasattr(source, "__mapper__"):
            return context.builder  # Not an ORM model, pass through

        # Mapper already processed by TypeHandleMapper; enrich with ORM details
        mapper = source.__mapper__
        builder = context.builder

        # Add ORM columns as field handles
        for column in mapper.columns:
            field = FieldHandleBuilder.placeholder(f"{source.__qualname__}.{column.name}")
            field = replace(field, orm_column=column)
            context.ensure_member(field)

        # Add ORM relationships with links
        for rel in mapper.relationships:
            field = FieldHandleBuilder.placeholder(f"{source.__qualname__}.{rel.key}")
            field = replace(field, orm_relationship=rel)
            # Enqueue the related type for processing
            target_qualname = rel.mapper.class_.__module__ + "." + rel.mapper.class_.__qualname__
            target = TypeHandleBuilder.placeholder(target_qualname)
            context.ensure_member(target)
            context.ensure_member(field)

        return builder

5.6 Type Resolver Orchestration¶

TypeResolver keeps its public API (resolve(dotted_path: str) -> TypeHandle), but internally orchestrates the ParsingContext and ParsingPipeline on cache miss:

class TypeResolver:
    def resolve(self, dotted_path: str) -> TypeHandle:
        """Resolve a type by dotted path.

        :param dotted_path: Fully qualified name (e.g., "my_app.Customer").
        :return: Fully resolved TypeHandle with ownership chains intact.
        """
        # Cache hit: return immediately
        cached = self._cache.get(dotted_path)
        if cached is not None:
            return cached

        # Cache miss: spin up parsing context + pipeline
        context = ParsingContext(root_qualname=dotted_path, cache=self._cache)

        # Depth-first processing (LIFO stack)
        while (pending_builder := context.next_pending()) is not None:
            # Import the source object
            source_obj = self._import_from_qualname(pending_builder.qualname)
            # Determine parent and module contexts
            parent, module = self._resolve_context(pending_builder.qualname)
            # Run mappers
            enhanced = self._pipeline.map(source_obj, pending_builder, context, parent, module)
            # Register back into context
            context.register(enhanced)

        # Finalize and cache
        finalized = context.finalize_all()
        self._cache.update(finalized)
        return finalized[dotted_path]

🧠 6. Semantics¶

• Mapping-centric design: Handle creation is fundamentally about mapping a source object (module, class, function, property) to a HandleBuilder. Identification and enrichment happen in a single pass per source type.
• Single-strategy protocol: HandleBuildingStrategy unifies identify + enrich responsibilities. Multiple source-specific implementations (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper) execute in sequence, each enhancing the builder.
• Breadth-first discovery: ParsingContext maintains a queue of pending builders. When a strategy discovers a member, it enqueues a placeholder via context.ensure_member(). The queue ensures breadth-first processing: all of Customer's methods are discovered before any method's parameters are discovered.
• Ownership chain guarantee: The parsing context ensures that when building a MethodHandle, its containing TypeHandle is already in the builder registry (and its containing ModuleHandle is guaranteed). This enables context.parent and context.module to be set correctly during mapping.
• Scope expansion (Option A): Resolving my_app.Customer expands to build Customer AND all its methods AND all their parameters AND any referenced types. Full scope expansion ensures complete ownership chains without gaps. Deferred lazy-hydration (Phase C of roadmap-002) will optimize this.
• Deterministic stacking: Strategies execute in pipeline order. Later strategies refine earlier results (last-writer-wins per qualname). Each source gets mapped exactly once per resolve cycle.
• Builder immutability: Every with_attr() returns a new builder; strategies never mutate input builders. This preserves references and enables safe composition.
• Circular reference handling: When Customer → Order → Customer, both types get enqueued. When dequeued, the second reference skips (already in registry). Finalization resolves links automatically.
• Metrics isolation: _execute_with_metrics() wraps each strategy, measuring duration/count/errors under handle_building.<slug>. Metrics never raise; failures are logged but don't interrupt mapping.
• Resolver boundary: TypeResolver.resolve(path) handles cache hits, instantiates ParsingContext + ParsingPipeline on miss, manages imports, and persists results. External API unchanged; internal orchestration is cleaner.

🧪 7. Examples¶

7.1 Custom SQLAlchemy Mapper¶

Framework integrators extend the core mapping pipeline by chaining custom mappers after the core set. Custom mappers refine builders without rediscovering core members.

from dataclasses import replace

class SQLAlchemyHandleMapper(HandleBuildingStrategy):
    """Enriches type handles with SQLAlchemy ORM metadata."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        # Only handle SQLAlchemy mapped classes
        if not isinstance(source, type) or not hasattr(source, "__mapper__"):
            return context.builder  # Pass through if not ORM

        mapper = source.__mapper__
        builder = context.builder

        # Discover ORM columns as field handles
        for column in mapper.columns:
            field = FieldHandleBuilder.placeholder(f"{source.__qualname__}.{column.name}")
            field = replace(field, orm_column=column)
            context.ensure_member(field)

        # Discover ORM relationships with type linking
        for rel in mapper.relationships:
            field = FieldHandleBuilder.placeholder(f"{source.__qualname__}.{rel.key}")
            field = replace(field, orm_relationship=rel)
            # Enqueue the related type
            target_type = TypeHandleBuilder.placeholder(rel.mapper.class_.__qualname__)
            context.ensure_member(target_type)
            context.ensure_member(field)

        return builder

Usage: Chain after core mappers to add ORM semantics.

7.2 Custom Descriptor Mapper¶

User-defined descriptors can be treated as first-class members via a custom mapper:

from dataclasses import replace

class DescriptorHandleMapper(HandleBuildingStrategy):
    """Discovers descriptor implementations on classes."""

    def map(self, source: object, context: MappingContext) -> HandleBuilder:
        if not isinstance(source, type):
            return context.builder

        builder = context.builder

        # Find descriptor implementations (classes with __get__ and __set__)
        for name, value in vars(source).items():
            if hasattr(value, "__get__") and hasattr(value, "__set__"):
                descriptor = PropertyHandleBuilder.placeholder(f"{source.__qualname__}.{name}")
                descriptor = replace(descriptor, descriptor_type=type(value).__name__)
                context.ensure_member(descriptor)

        return builder

7.3 ReflectionContext Integration¶

Wire custom mappers into the parsing pipeline by composing them after core mappers:

from loryth.common.reflection import ReflectionContext
from loryth.common.collections import ConcurrentHierarchicalCache

# Create pipeline with core + custom mappers
cache = ConcurrentHierarchicalCache()
pipeline = ParsingPipeline(
    mapping_strategies=[
        # Core mappers (always first)
        ModuleHandleMapper(),
        TypeHandleMapper(),
        FunctionHandleMapper(),
        # Custom mappers (chain after core)
        SQLAlchemyHandleMapper(),
        DescriptorHandleMapper(),
    ],
    metrics=ReflectionMetricsCollector("handle_building"),
)

# Create resolver with pipeline
resolver = TypeResolver(cache=cache, pipeline=pipeline)

# Use in ReflectionContext
context = ReflectionContext(type_resolver=resolver)

# Resolve a type: builds Customer + methods + parameters + ORM columns + relationships
customer_handle = context.resolve_type("my_app.models.Customer")

Expected behavior: - Customer handle includes standard methods/properties (from TypeHandleMapper) - Customer handle includes ORM columns/relationships (from SQLAlchemyHandleMapper) - All related types (Order, User, etc.) are fully resolved via breadth-first queueing - Ownership chains intact: Customer.orders[0].customer points back to Customer (shared finalized instance)

🧪 8. Test Matrix¶

8.1 Unit Coverage¶

8.2 Integration Coverage¶

8.3 Edge Case Coverage (Required)¶

8.4 Accepted Risks (Deferred to Future Phases)¶

✅ 9. Acceptance Criteria¶

• HandleBuildingStrategy protocol: Single protocol with map(source, context) -> builder signature published and documented; source-specific implementations (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper) with ≥80% unit coverage.
• MappingContext immutability: Context is frozen dataclass; strategies cannot mutate it; all enrichment via with_attr() builder API; ≥90% coverage.
• ParsingContext queue management: next_pending() correctly deduplicates (skips finalized qualnames); ensure_member() and register_builder() work correctly; circular deduplication tested; ≥90% coverage.
• Edge Case 1 (Circular Imports): Module A ↔ Module B circular import handled without infinite loops; both in final registry; test assertion verifies deduplication.
• Edge Case 3 (Deeply Nested): 5+ level nesting (Module → Type → Method → Parameter → Type → Method) builds correctly with intact ownership chains; finalization completes without stack overflow.
• Edge Case 7 (Circular Types): Customer ↔ Order bidirectional references resolve correctly; both enqueued once; finalized graph shows shared instances (Customer.orders[0].customer == original Customer).
• Accepted Risks Documented: Risks 2, 4, 5, 6 are documented with current behavior, mitigation strategy, and target phase for resolution.
• Runtime flow: TypeResolver + ParsingContext breadth-first processing works end-to-end; cache hits bypass pipeline; cache misses expand full scope correctly.
• Default strategy parity: Core mappers produce outputs matching current golden fixtures (regression parity).
• Metrics instrumentation: Metrics emitted under handle_building.<slug> namespace with never-raise semantics; ≥90% coverage.
• Migration guide: Custom mapper authoring guide with SQLAlchemy and descriptor examples; contributors can extend pipeline safely.

🚀 10. Rollout and Migration¶

10.1 Implementation Phases¶

Phase 1 — Foundation & Core Strategy Extraction (⏳ Pending) - Create reflection/strategies/ package, define protocols, contexts, and ParsingPipeline. - Extract class/module/package inspectors plus type-hint/visibility/docstring annotators from MemberBuilder. - Update TypeResolver internals to instantiate a parsing context + pipeline on cache miss while preserving the public resolve() signature. - Wire default strategies into the pipeline, seed builder factories, and run golden regression fixtures for parity. - Instrument the pipeline with _execute_with_metrics() and validate namespaces via local dashboards. - Outcome: the default builder path uses strategies with no observable API change.

Phase 2 — Composability, Merging & Contract Tests (⏳ Pending) - Implement metadata + member merging utilities (qualname dedupe, dictionary merge) and provenance tracking. - Extend the builder index so inspectors can emit placeholders and annotators can resolve related members safely. - Ship reusable inspection/annotation contract tests plus stacked-strategy integration suites (core + custom mixes). - Document deterministic ordering rules, builder placeholder semantics, and strategy registration; update metrics coverage to include custom slugs. - Outcome: stacking semantics are proven with automated tests guarding regressions, and external contributors have a contract harness to validate their strategies.

Phase 3 — Examples, Documentation & Migration (⏳ Pending) - Deliver example strategies (SQLAlchemy, descriptors, runtime attributes) plus an authoring guide and FAQ. - Provide migration documentation with before/after snippets, DI/runtime integration tests, and performance + memory benchmarks. - Update contributor docs (reflection-architecture.md, roadmap) and add monitoring dashboards for the new metrics namespaces. - Partner with framework teams to pilot custom strategies and capture telemetry/ergonomics feedback. - Outcome: maintainers and integrators have prescriptive guidance plus telemetry to adopt the seam safely, backed by pilot implementations.

Phase 4 — Remove Legacy MemberBuilder (⏳ Pending) - Identify and update all tests that depend on MemberBuilder internals; migrate to builder-based assertions or remove/replace tests as needed. - Remove MemberBuilder class and all internal parsing methods from the codebase. - Update any remaining DI or reflection code that may reference deprecated helpers (e.g., build_type, build_module). - Run full regression suite to ensure no observable API changes for end-users (public TypeResolver API remains unchanged). - Outcome: legacy monolithic builder is completely removed; codebase is cleaner and uses only the strategy-based architecture.

10.2 Migration Strategy¶

Existing Reflection Codebase - Keep calling TypeResolver.resolve() / ReflectionContext.resolve_type(); the resolver now spins up parsing contexts internally, so no call-site changes are required. - When legacy code touches MemberBuilder internals, migrate to HandleBuildingStrategy implementations and register with the shared pipeline. - Replace any direct MemberHandle mutation with dataclasses.replace() on builder instances; regression fixtures should compare finalized graphs before/after.

Framework / ORM Integrators - Provide custom mappers for framework-specific discovery (e.g., SQLAlchemyHandleMapper) by implementing map() and enqueueing specialized builders via context.ensure_member(). - Use dataclasses.replace() to enrich builders with framework metadata rather than mutating handles directly. - Compose pipelines/resolvers through small factories (as shown in §7.3) so DI containers and reflection entry points can swap strategy stacks without subclassing. - Use before/after verification by comparing finalized MemberHandle graphs with and without the new strategies, and rely on pipeline metrics to monitor custom hook performance.

⚠️ 11. Risks and Mitigations¶

📈 12. Progress¶

Legend: ⏳ Pending · 🚧 In Progress · ✅ Completed

• Phase 1 — Foundation & Core Strategy Extraction — ✅ Complete (2025-10-27)
• Phase 2 — Composability, Merging & Contract Tests — ⏳
• Phase 3 — Examples, Documentation & Migration — ⏳
• Phase 4 — Remove Legacy MemberBuilder — ⏳

📝 13. Changelog¶

• 2025-11-09 — Architecture consolidation and terminology alignment:
• Consolidated all strategy components under loryth.common.reflection.engine.strategies (moved from reflection/strategies)
• Replaced builders with specs (immutable classes using copy() + with_*() pattern instead of frozen dataclasses)
• Renamed HandleBuildingStrategy → HandleGenerationStrategy for clarity
• Renamed ParsingContext → HandleGenerationContext for consistency
• Created BaseSpec base class with qualname, docstring, meta attributes
• Implemented all spec types: ModuleSpec, FunctionSpec, ClassSpec, ClassMethodSpec, ClassPropertySpec, ClassFieldSpec, ParameterSpec
• Updated HandleBuildingPipeline to work with specs instead of builders
• Created comprehensive public API exports in engine/init.py
• Updated TD-029 document to reflect Generation terminology and engine location
• Backward compatibility maintained via re-exports in deprecated handle_generation_strategy.py
• Ready for Phase 2: Test coverage implementation
• 2025-10-27 — Phase 1 implementation complete:
• Created strategies package with all core components (~1,500 LOC)
• Implemented HandleBuildingStrategy protocol (protocols.py)
• Created ParsingContext and MappingContext (contexts.py)
• Implemented 8 frozen dataclass builders (handle_builder.py, module_builder.py, type_builder.py, function_builder.py, method_builder.py, parameter_builder.py, field_builder.py, property_builder.py)
• Extracted 3 core mappers (ModuleHandleMapper, TypeHandleMapper, FunctionHandleMapper)
• Implemented ParsingPipeline with metrics instrumentation (pipeline.py)
• Created default pipeline factory: create_default_parsing_pipeline() in reflection/init.py
• Updated RuntimeTypeResolver to support new architecture while maintaining backward compatibility
• All 791 existing tests pass with zero regressions
• New strategies package fully exported in reflection/init.py
• Used from __future__ import annotations to resolve circular imports between builder modules
• Mark Phase 1 ✅ Complete
• 2025-10-26 — Final consistency review and fixes:
• Fixed Section 3.1 table: Updated to reference dataclasses.replace() instead of with_attr()
• Deleted Section 5.2 (Specialized Member Builders): Removed outdated fluent API documentation that contradicted the design
• Fixed all mapper code examples (Sections 5.4.1, 5.4.2, 5.4.3, 5.5): Replaced with_attr() chains with dataclasses.replace() calls
• Added from dataclasses import replace imports to all mapper examples
• Fixed Section 5.3: Updated contract highlights to reference dataclasses.replace() instead of with_attr()
• Fixed Section 5.6: Changed comment from "breadth-first processing" to "depth-first processing (LIFO stack)"
• Fixed Section 7.1 and 7.2: Updated SQLAlchemy and Descriptor mapper examples to use dataclasses.replace()
• Completely rewrote Appendix A: Removed outdated two-strategy model (inspection/annotation); replaced with current single-strategy architecture (Appendix A.1-A.4 now document ParsingContext, MappingContext, ParsingPipeline, and TypeResolver correctly)
• All code examples now consistent with design decision to use frozen dataclasses with dataclasses.replace()
• 2025-10-26 — Major design refinement:
• Renamed HandleMappingStrategy → HandleBuildingStrategy to better express "building" handles
• Changed from breadth-first queue to depth-first stack (LIFO) to reduce peak memory usage
• Replaced with_attr() fluent API with dataclasses.replace() for type-safe builder enrichment
• Updated Section 4.1: ParsingContext documented with code + docstrings (removed table)
• Added Phase 4 (Remove Legacy MemberBuilder) to rollout plan with test migration details
• Removed "previous design" references from Section 3
• Updated metrics namespace from handle_mapping to handle_building
• Status changed to Draft (from Proposed)
• 2025-10-26 — Updated to reflect TD-030 completion (Handle hierarchy):
• Updated all *Info class references to *Handle (MemberInfo → MemberHandle, TypeInfo → TypeHandle, ModuleInfo → ModuleHandle, etc.)
• Added TD-030 to Related section (foundation for this design)
• Revised Section 5.8 (PackageInspectionStrategy) to reflect unified ModuleHandle with is_package property (no separate PackageHandle)
• Updated code examples to use Handle terminology
• Note: PackageHandle no longer exists; packages are now ModuleHandle with is_package=True
• 2025-10-25 — Initial draft created (status Proposed) with single HandleBuildingStrategy protocol, phase plan, contract tests, and examples.

📖 14. Glossary¶

📎 Appendix A — Reference Sketches¶

A.1 ParsingContext Implementation¶

from dataclasses import dataclass

class ParsingContext:
    """Manages builder lifecycle during cache miss.

    Uses LIFO stack for depth-first discovery.
    """

    def __init__(self, root_qualname: str, cache: HierarchicalCache) -> None:
        self._cache = cache
        self._builder_registry: dict[str, HandleBuilder] = {}
        self._pending: list[HandleBuilder] = [HandleBuilder.placeholder(root_qualname)]
        self._finalized: dict[str, HandleBuilder] = {}

    def register_builder(self, builder: HandleBuilder) -> None:
        """Add or replace a builder in the registry."""
        self._builder_registry[builder.qualname] = builder

    def next_pending(self) -> HandleBuilder | None:
        """Pop next builder from stack (LIFO). Skip if finalized."""
        while self._pending:
            candidate = self._pending.pop()  # Pop from end (LIFO)
            if candidate.qualname not in self._finalized:
                return candidate
        return None

    def finalize_all(self) -> dict[str, HandleBuilder]:
        """Materialize all builders in registry."""
        for qualname, builder in self._builder_registry.items():
            if qualname not in self._finalized:
                self._finalized[qualname] = builder  # In practice, call builder.finalize()
        return self._finalized

A.2 MappingContext Implementation¶

from dataclasses import dataclass


@dataclass(frozen=True)
class MappingContext:
    """Immutable context provided to mapping strategies."""

    source: object
    builder: HandleBuilder
    parent: TypeHandleBuilder | None
    module: ModuleHandleBuilder
    cache: HierarchicalCache
    _parsing_context: ParsingContext

    def ensure_member(self, builder: HandleBuilder) -> HandleBuilder:
        """Enqueue discovered member for recursive processing."""
        self._parsing_context.register(builder)
        self._parsing_context._pending.append(builder)  # Push to stack
        return builder

    def get_builder(self, qualname: str) -> HandleBuilder | None:
        """Look up an already-discovered builder."""
        return self._parsing_context._registry.get(qualname)

A.3 ParsingPipeline Implementation¶

import time
from typing import Sequence

class ParsingPipeline:
    """Orchestrates mapping strategies in sequence."""

    def __init__(
        self,
        mapping_strategies: Sequence[HandleBuildingStrategy],
        metrics: ReflectionMetricsCollector,
    ) -> None:
        self._mapping_strategies = list(mapping_strategies)
        self._metrics = metrics

    def map(
        self,
        source: object,
        builder: HandleBuilder,
        parsing_context: ParsingContext,
        parent: TypeHandleBuilder | None,
        module: ModuleHandleBuilder,
    ) -> HandleBuilder:
        """Execute mapping strategies in sequence."""
        context = MappingContext(source, builder, parent, module, parsing_context._cache, parsing_context)

        current_builder = builder
        for strategy in self._mapping_strategies:
            current_builder = self._execute_with_metrics(
                strategy,
                lambda: strategy.map(source, dataclasses.replace(context, builder=current_builder))
            )
        return current_builder

    def _execute_with_metrics(
        self,
        strategy: HandleBuildingStrategy,
        func,
    ) -> HandleBuilder:
        """Execute strategy with metrics wrapping."""
        start = time.perf_counter()
        errors = 0
        result = None
        try:
            result = func()
            return result
        except Exception:
            errors = 1
            raise
        finally:
            duration = time.perf_counter() - start
            slug = strategy.__class__.__name__.removesuffix("Mapper").lower()
            namespace = f"handle_building.{slug}"
            self._metrics.record(namespace=namespace, duration=duration, count=1, errors=errors)

A.4 Complete TypeResolver Example¶

import importlib
from typing import Callable


class TypeResolver:
    """Public API for resolving types to TypeHandle builders."""

    def __init__(
            self,
            cache: HierarchicalCache,
            pipeline: ParsingPipeline,
    ) -> None:
        self._cache = cache
        self._pipeline = pipeline

    def resolve(self, dotted_path: str) -> TypeHandle:
        """Resolve a type by dotted path.

        Returns cached result on hit; builds full scope on miss.
        """
        # Cache hit: return immediately
        cached = self._cache.get(dotted_path)
        if cached is not None:
            return cached

        # Cache miss: spin up parsing context
        context = ParsingContext(dotted_path, self._cache)

        # Depth-first processing (LIFO stack)
        while (pending_builder := context.next_pending()) is not None:
            # Import source object
            source_obj = self._import(pending_builder.qualname)

            # Resolve parent and module contexts
            parent, module = self._resolve_context(pending_builder.qualname)

            # Run mappers
            enhanced = self._pipeline.map(source_obj, pending_builder, context, parent, module)

            # Register back into context
            context.register(enhanced)

        # Finalize and cache
        finalized = context.finalize_all()
        self._cache.update(finalized)
        return finalized[dotted_path]

    def _import(self, dotted_path: str) -> object:
        """Import a Python object by fully qualified dotted path."""
        module_name, _, attr = dotted_path.rpartition(".")
        module = importlib.import_module(module_name)
        return getattr(module, attr) if attr else module

    def _resolve_context(self, qualname: str) -> tuple[TypeHandleBuilder | None, ModuleHandleBuilder]:
        """Resolve parent type and owning module for a qualname."""
        # Implementation: parse qualname, look up parent and module in registry
        # Return (parent_type_handle, module_handle)
        pass