symfonic.capabilities.memory.port_inspection

port_inspection

A v1 MemoryCapability's writer and lifecycle must be ports the fence can see (CORE-194).

CORE-199 fences a turn by capturing an erase admission before it reads; only memory_capabilities composes that capture, and every write the turn makes presents it through the composition's own :class:GuardedMemory. A port the fence cannot inspect -- a forwarding wrapper, a subclass, an object claiming another class -- may hold raw graph-only PostgreSQL v1 memory where no type check sees it, and a hand-built capability over the raw store itself captures nothing. Either composes a turn whose writes no admission covers, so on a v1 scope each refuses at construction. Positional (v0) scopes, capabilities without ports, and the factory over an exact shipped store are unchanged.

The admission covers what the composition's own hydrator reads. So a guard is accepted only beside the hydrator and scope the factory issued it with (:func:pair_guard): the factory's guard taken out and set beside an adopter hydrator that reads another scope would write that scope's content after its forget. dataclasses.replace of the factory's capability keeps the pair.

pair_guard

pair_guard(writer: Any, hydrator: Any, scope: Any) -> Any

Record the factory's guard with the hydrator and scope it composes beside; return it.

Source code in src/symfonic/capabilities/memory/port_inspection.py

def pair_guard(writer: Any, hydrator: Any, scope: Any) -> Any:
    """Record the factory's guard with the hydrator and scope it composes beside; return it."""
    from symfonic.capabilities.memory.guarded_memory import GuardedMemory

    if type(writer) is GuardedMemory:
        _PAIRED[writer] = (hydrator, scope.path)
    return writer

refuse_uninspectable

refuse_uninspectable(scope: Any, *ports: Any, hydrator: Any = None) -> None

Refuse a v1 writer or lifecycle that is not an inspectable shipped port.

Source code in src/symfonic/capabilities/memory/port_inspection.py

def refuse_uninspectable(scope: Any, *ports: Any, hydrator: Any = None) -> None:
    """Refuse a v1 writer or lifecycle that is not an inspectable shipped port."""
    if getattr(scope, "identity_version", 0) != 1:
        return
    for port in ports:
        if port is not None:
            _inspect(scope, port, hydrator)